60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
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, CookieTransport
|
|
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")
|
|
cookie_transport = CookieTransport(cookie_name="rpkapiusersauth")
|
|
|
|
|
|
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=cookie_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)
|