Coverage for core/dependencies.py: 100.00%

1import logging 

2 

3from fastapi import HTTPException 

4from sqlalchemy.ext.asyncio import AsyncSession 

5 

6from core.database import AsyncSessionLocal, SessionLocal 

7 

8logger = logging.getLogger("dependencies") 

9 

10 

11# Async DB dependency 

12async def get_db() -> AsyncSession: 

13 async with AsyncSessionLocal() as db: 

14 try: 

15 yield db 

16 except HTTPException: 

17 await db.rollback() 

18 raise 

19 except Exception as e: 

20 logger.error("Database error: %s", e, exc_info=True) 

21 await db.rollback() 

22 raise e 

23 

24 

25# Sync DB dependency (for migration/schedule) 

26def get_sync_db(): 

27 db = SessionLocal() 

28 try: 

29 yield db 

30 except Exception as e: 

31 logger.error("Database error: %s", e, exc_info=True) 

32 db.rollback() 

33 raise e 

34 finally: 

35 db.close()