Coverage for src/database/db.py: 46%

1import redis.asyncio as aioredis 

2from sqlalchemy.exc import SQLAlchemyError 

3 

4from fastapi import HTTPException 

5from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine 

6from src.conf.config import settings 

7 

8DB_URL = settings.sqlalchemy_database_url 

9REDIS_URL = settings.redis_url 

10 

11engine = create_async_engine(DB_URL) 

12local_session = async_sessionmaker(autocommit=False, autoflush=False, bind=engine) 

13 

14 

15async def get_db(): 

16 """ 

17 Asynchronously creates a database session using SQLAlchemy's async_sessionmaker. 

18 

19 Parameters: 

20 None 

21 

22 Returns: 

23 async_generator: An asynchronous generator yielding a SQLAlchemy session. 

24 

25 Raises: 

26 HTTPException: If a SQLAlchemyError occurs during session creation, a 500 error is raised. 

27 """ 

28 try: 

29 async with local_session() as session: 

30 yield session 

31 except SQLAlchemyError as e: 

32 print(f"Database session creation failed: {e}") 

33 raise HTTPException(status_code=500, detail="Database session creation failed") 

34 

35async def get_redis_client(): 

36 """ 

37 Asynchronously creates a Redis client using aioredis. 

38 

39 Parameters: 

40 None 

41 

42 Returns: 

43 aioredis.Redis: An asynchronous Redis client. 

44 

45 Raises: 

46 HTTPException: If a RedisError occurs during client creation, a 500 error is raised. 

47 """ 

48 redis_client = aioredis.from_url(REDIS_URL) 

49 try: 

50 yield redis_client 

51 except Exception as e: 

52 print(e) 

53 raise HTTPException(status_code=500, detail="Database is not configured correctly!") 

54 finally: 

55 await redis_client.close()