Spaces:
Build error
Build error
File size: 1,564 Bytes
bd09498 fc1a5f0 bd09498 fc1a5f0 2e30fe9 fc1a5f0 2e30fe9 fc1a5f0 bd09498 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """
Engine SQLAlchemy async (aiosqlite) et session factory.
Utilisation dans les endpoints FastAPI :
async def my_endpoint(db: AsyncSession = Depends(get_db)):
...
Les tables sont créées au démarrage de l'application (voir main.py lifespan).
"""
# 1. stdlib
import logging
# 2. third-party
from sqlalchemy import event
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
# 3. local
from app.config import settings
logger = logging.getLogger(__name__)
engine = create_async_engine(
settings.database_url,
echo=False,
connect_args={"check_same_thread": False},
)
# Activer les clés étrangères SQLite (désactivées par défaut).
# Nécessaire pour que ondelete="CASCADE" / "SET NULL" fonctionne.
# Note : on n'appelle PAS cursor.close() car avec aiosqlite le curseur
# retourne une coroutine pour close(), ce qui provoque un RuntimeWarning
# « coroutine 'Cursor.close' was never awaited ». Le curseur PRAGMA est
# éphémère et libéré automatiquement.
@event.listens_for(engine.sync_engine, "connect")
def _set_sqlite_pragma(dbapi_conn, _connection_record):
dbapi_conn.execute("PRAGMA foreign_keys=ON")
async_session_factory = async_sessionmaker(
engine,
expire_on_commit=False,
class_=AsyncSession,
)
class Base(DeclarativeBase):
pass
async def get_db():
"""Dépendance FastAPI — injecte une AsyncSession par requête."""
async with async_session_factory() as session:
yield session
|