import os import uuid from fastapi import Depends from fastapi_users import UUIDIDMixin, BaseUserManager 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 fastapi_users.fastapi_users import get_oauth_router 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")) 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, ) oauth_router = get_oauth_router(google_oauth_client, auth_backend, get_user_manager, SECRET)