Coverage for opt/mealie/lib/python3.12/site-packages/mealie/db/db_setup.py: 87%

1from collections.abc import Generator 

2from contextlib import contextmanager 

3 

4import sqlalchemy as sa 

5from sqlalchemy.engine import Engine 

6from sqlalchemy.event import listens_for 

7from sqlalchemy.orm import sessionmaker 

8from sqlalchemy.orm.session import Session 

9 

10from mealie.core.config import get_app_settings 

11 

12settings = get_app_settings() 

13 

14 

15@listens_for(Engine, "connect") 

16def set_sqlite_pragma_journal_wal(dbapi_connection, connection_record): 

17 """ 

18 Automatically enables SQLite's WAL journal mode if the setting is activated. 

19 

20 This is a persistent setting, so turning it off down the line doesn't revert back to the original journal mode. 

21 

22 Write-Ahead-Log enables sqlite to be used concurrently by multiple readers and writers (writes still happen 

23 sequentially). 

24 """ 

25 global settings 

26 if settings.DB_ENGINE != "sqlite" or not settings.SQLITE_MIGRATE_JOURNAL_WAL: 26 ↛ 28line 26 didn't jump to line 28 because the condition on line 26 was always true

27 return 

28 cursor = dbapi_connection.cursor() 

29 cursor.execute("PRAGMA journal_mode=WAL") 

30 cursor.close() 

31 

32 

33def sql_global_init(db_url: str): 

34 connect_args = {} 

35 if "sqlite" in db_url: 35 ↛ 38line 35 didn't jump to line 38 because the condition on line 35 was always true

36 connect_args["check_same_thread"] = False 

37 

38 engine = sa.create_engine(db_url, echo=False, connect_args=connect_args, pool_pre_ping=True, future=True) 

39 

40 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True) 

41 

42 return SessionLocal, engine 

43 

44 

45SessionLocal, engine = sql_global_init(settings.DB_URL) # type: ignore 

46 

47 

48@contextmanager 

49def session_context() -> Generator[Session, None, None]: 

50 """ 

51 session_context() provides a managed session to the database that is automatically 

52 closed when the context is exited. This is the preferred method of accessing the 

53 database. 

54 

55 Note: use `generate_session` when using the `Depends` function from FastAPI 

56 """ 

57 global SessionLocal 

58 sess = SessionLocal() 

59 try: 

60 yield sess 

61 finally: 

62 sess.close() 

63 

64 

65def generate_session() -> Generator[Session, None, None]: 

66 """ 

67 WARNING: This function should _only_ be called when used with 

68 using the `Depends` function from FastAPI. This function will leak 

69 sessions if used outside of the context of a request. 

70 

71 Use `with_session` instead. That function will allow you to use the 

72 session within a context manager 

73 """ 

74 global SessionLocal 

75 db = SessionLocal() 

76 try: 

77 yield db 

78 finally: 

79 db.close()