53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
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)
|