Deployed dc4f73f to dev with MkDocs 1.2.4 and mike 1.1.2

dev/src/db_sqlalchemy_access_tokens.py

@@ -0,0 +1,46 @@
+from typing import AsyncGenerator
+
+from fastapi import Depends
+from fastapi_users.db import SQLAlchemyBaseUserTable, SQLAlchemyUserDatabase
+from fastapi_users_db_sqlalchemy.access_token import (
+    SQLAlchemyAccessTokenDatabase,
+    SQLAlchemyBaseAccessTokenTable,
+)
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from .models import AccessToken, UserDB
+
+DATABASE_URL = "sqlite+aiosqlite:///./test.db"
+Base: DeclarativeMeta = declarative_base()
+
+
+class UserTable(Base, SQLAlchemyBaseUserTable):
+    pass
+
+
+class AccessTokenTable(SQLAlchemyBaseAccessTokenTable, Base):
+    pass
+
+
+engine = create_async_engine(DATABASE_URL)
+async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+
+async def create_db_and_tables():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
+    async with async_session_maker() as session:
+        yield session
+
+
+async def get_user_db(session: AsyncSession = Depends(get_async_session)):
+    yield SQLAlchemyUserDatabase(UserDB, session, UserTable)
+
+
+async def get_access_token_db(session: AsyncSession = Depends(get_async_session)):
+    yield SQLAlchemyAccessTokenDatabase(AccessToken, session, AccessTokenTable)