Coverage for src / database / session.py: 100%

1"""Gerenciamento de engine e sessoes SQLModel.""" 

2 

3from functools import lru_cache 

4 

5from sqlalchemy.orm import sessionmaker 

6from sqlmodel import Session, create_engine 

7 

8from src.core.config import get_database_url, get_settings 

9 

10 

11def _get_echo_value(settings) -> bool: 

12 value = settings.get("database.echo", None) 

13 if value is not None: 

14 return bool(value) 

15 

16 database_settings = settings.get("database", {}) 

17 if isinstance(database_settings, dict): 

18 return bool(database_settings.get("echo", False)) 

19 

20 return False 

21 

22 

23@lru_cache 

24def get_engine(): 

25 """Cria engine SQLAlchemy com base nas configuracoes ativas.""" 

26 settings = get_settings() 

27 database_url = get_database_url(settings) 

28 echo = _get_echo_value(settings) 

29 

30 return create_engine(database_url, echo=echo, pool_pre_ping=True) 

31 

32 

33@lru_cache 

34def get_session_local(): 

35 """Retorna factory de sessoes para o banco configurado.""" 

36 return sessionmaker( 

37 bind=get_engine(), 

38 autocommit=False, 

39 autoflush=False, 

40 class_=Session, 

41 ) 

42 

43 

44def get_db(): 

45 """Fornece sessao transiente para injecao de dependencia no FastAPI.""" 

46 session = get_session_local()() 

47 try: 

48 yield session 

49 finally: 

50 session.close()