58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
import uuid
|
|
from typing import Optional
|
|
|
|
from fastapi import Depends, Request
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from fastapi_users import BaseUserManager, FastAPIUsers
|
|
from fastapi_users.authentication import AuthenticationBackend, BearerTransport, JWTStrategy
|
|
from fastapi_users.db import SQLAlchemyUserDatabase
|
|
|
|
from database import get_async_session
|
|
from models import User
|
|
from config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
class UserManager(BaseUserManager[User, uuid.UUID]):
|
|
reset_password_token_secret = settings.SECRET_KEY
|
|
verification_token_secret = settings.SECRET_KEY
|
|
|
|
def parse_id(self, value: any) -> uuid.UUID:
|
|
"""Parse string ID to UUID"""
|
|
if isinstance(value, uuid.UUID):
|
|
return value
|
|
return uuid.UUID(value)
|
|
|
|
async def on_after_register(self, user: User, request: Optional[Request] = None):
|
|
print(f"User {user.id} has registered.")
|
|
|
|
# You can add more hooks for password reset, verification etc.
|
|
# For this example, we'll keep it minimal.
|
|
|
|
async def get_user_db(session: AsyncSession = Depends(get_async_session)):
|
|
yield SQLAlchemyUserDatabase(session, User)
|
|
|
|
async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
|
|
yield UserManager(user_db)
|
|
|
|
bearer_transport = BearerTransport(tokenUrl="api/auth/jwt/login")
|
|
|
|
def get_jwt_strategy() -> JWTStrategy:
|
|
return JWTStrategy(secret=settings.SECRET_KEY, lifetime_seconds=settings.TOKEN_LIFETIME)
|
|
|
|
auth_backend = AuthenticationBackend(
|
|
name="jwt",
|
|
transport=bearer_transport,
|
|
get_strategy=get_jwt_strategy,
|
|
)
|
|
|
|
fastapi_users = FastAPIUsers[User, uuid.UUID](
|
|
get_user_manager,
|
|
[auth_backend],
|
|
)
|
|
|
|
# Dependencies for protecting endpoints
|
|
current_active_user = fastapi_users.current_user(active=True)
|
|
current_superuser = fastapi_users.current_user(active=True, superuser=True)
|