Spaces:
Runtime error
Runtime error
File size: 2,792 Bytes
18ade12 | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """
AsyncSession dependency for FastAPI.
This module provides database session management with:
- AsyncSession context manager
- FastAPI dependency injection helper
- Graceful error handling
"""
import logging
from typing import AsyncGenerator
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlmodel.ext.asyncio.session import AsyncSession
from src.models.base import get_engine
logger = logging.getLogger(__name__)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
"""
FastAPI dependency that provides an AsyncSession.
Usage:
@app.get("/items")
async def get_items(session: AsyncSession = Depends(get_session)):
...
Yields:
AsyncSession: Database session for the request
Raises:
SQLAlchemyError: If database connection fails (handled gracefully)
"""
engine = get_engine()
async with AsyncSession(engine) as session:
try:
yield session
await session.commit()
except OperationalError as e:
await session.rollback()
# Log error without exposing credentials
logger.error(
"Database connection error: Unable to connect to database. "
"Please verify your DATABASE_URL configuration."
)
raise
except SQLAlchemyError as e:
await session.rollback()
logger.error(f"Database error: {type(e).__name__}")
raise
except Exception as e:
await session.rollback()
logger.error(f"Unexpected error during database operation: {type(e).__name__}")
raise
class DatabaseSessionManager:
"""
Context manager for manual database session handling.
Usage:
async with DatabaseSessionManager() as session:
result = await session.exec(select(User))
"""
def __init__(self):
self.engine = get_engine()
self.session: AsyncSession | None = None
async def __aenter__(self) -> AsyncSession:
self.session = AsyncSession(self.engine)
return self.session
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
if exc_type is not None:
await self.session.rollback()
# Log error without exposing sensitive details
if exc_type == OperationalError:
logger.error(
"Database connection error during session. "
"Connection may have been lost."
)
else:
logger.error(f"Database error: {exc_type.__name__}")
else:
await self.session.commit()
await self.session.close()
|