41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
from typing import Annotated, AsyncGenerator
|
|
|
|
from fastapi import Depends
|
|
from fastapi_users.authentication.strategy import AccessTokenDatabase, DatabaseStrategy
|
|
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
|
|
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.db.database import async_session_maker
|
|
from src.models import UsersORM, AccessToken
|
|
from src.settings import settings
|
|
|
|
|
|
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
|
async with async_session_maker as db:
|
|
yield db
|
|
|
|
|
|
DBDep = Annotated[AsyncSession, Depends(get_db)]
|
|
|
|
|
|
async def get_user_db(
|
|
session: DBDep
|
|
):
|
|
yield SQLAlchemyUserDatabase(session, UsersORM)
|
|
|
|
|
|
async def get_access_token_db(
|
|
session: DBDep,
|
|
):
|
|
yield SQLAlchemyAccessTokenDatabase(session, AccessToken)
|
|
|
|
|
|
ATDep = Annotated[AccessTokenDatabase[AccessToken], Depends(get_access_token_db)]
|
|
|
|
|
|
def get_database_strategy(
|
|
access_token_db: ATDep,
|
|
) -> DatabaseStrategy:
|
|
return DatabaseStrategy(access_token_db, lifetime_seconds=settings.lifetime)
|