import os import uuid from beanie import PydanticObjectId from fastapi import Depends from fastapi_users import UUIDIDMixin, BaseUserManager, FastAPIUsers, schemas from fastapi_users.authentication import AuthenticationBackend, BearerTransport from fastapi_users.authentication.strategy import AccessTokenDatabase, DatabaseStrategy from fastapi_users_db_beanie.access_token import BeanieBaseAccessTokenDocument, BeanieAccessTokenDatabase from httpx_oauth.clients.google import GoogleOAuth2 from httpx_oauth.clients.discord import DiscordOAuth2 from hub.user import User, get_user_db SECRET = os.getenv("FASTAPI_USERS_SECRET") google_oauth_client = GoogleOAuth2(os.getenv("GOOGLE_CLIENT_ID"), os.getenv("GOOGLE_CLIENT_SECRET")) discord_oauth_client = DiscordOAuth2(os.getenv("DISCORD_CLIENT_ID"), os.getenv("DISCORD_CLIENT_SECRET")) TOKEN_LIFETIME = 3600 bearer_transport = BearerTransport(tokenUrl="auth/login") class AccessToken(BeanieBaseAccessTokenDocument): pass async def get_access_token_db(): yield BeanieAccessTokenDatabase(AccessToken) def get_database_strategy( access_token_db: AccessTokenDatabase[AccessToken] = Depends(get_access_token_db), ) -> DatabaseStrategy: return DatabaseStrategy(access_token_db, lifetime_seconds=TOKEN_LIFETIME) class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]): reset_password_token_secret = SECRET verification_token_secret = SECRET async def get_user_manager(user_db=Depends(get_user_db)): yield UserManager(user_db) auth_backend = AuthenticationBackend( name="db", transport=bearer_transport, get_strategy=get_database_strategy, ) fastapi_users = FastAPIUsers[User, PydanticObjectId](get_user_manager, [auth_backend]) auth_router = fastapi_users.get_auth_router(auth_backend, requires_verification=True) register_router = fastapi_users.get_register_router(schemas.BaseUser, schemas.BaseUserCreate) password_router = fastapi_users.get_reset_password_router() verification_router = fastapi_users.get_verify_router(schemas.BaseUser) users_router = fastapi_users.get_users_router(schemas.BaseUser, schemas.BaseUserUpdate) google_oauth_router = fastapi_users.get_oauth_router(google_oauth_client, auth_backend, SECRET) discord_oauth_router = fastapi_users.get_oauth_router(discord_oauth_client, auth_backend, SECRET)