Coverage for app_modules/db.py: 92%

1""" 

2db.py 

3 

4Purpose: 

5 Encapsulate SQLite connection lifecycle, providing `get_db()` for obtaining a 

6 request-scoped connection and `register_teardown(app)` to close it after the 

7 request. This mirrors Flask's application context pattern and keeps DB logic 

8 out of `app.py`. 

9 

10Exports: 

11 - get_db(): returns a sqlite3.Row-based connection stored on flask.g 

12 - register_teardown(app): registers a teardown handler to close the DB 

13""" 

14 

15import sqlite3 

16import os 

17from flask import g 

18from .paths import ROOT_DIR 

19 

20 

21def get_db(): 

22 db = getattr(g, '_database', None) 

23 if db is None: 

24 # Use environment variable if set (for testing), otherwise use default 

25 db_path = os.getenv('DATABASE_PATH') or os.path.join(ROOT_DIR, 'theialogin.db') 

26 db = sqlite3.connect(db_path) 

27 db.row_factory = sqlite3.Row 

28 # Connection-wide pragmas for reliability and performance 

29 try: 

30 db.execute('PRAGMA journal_mode=WAL') 

31 db.execute('PRAGMA synchronous=NORMAL') 

32 db.execute('PRAGMA foreign_keys=ON') 

33 except Exception: 

34 pass 

35 g._database = db 

36 return db 

37 

38 

39def register_teardown(app): 

40 @app.teardown_appcontext 

41 def close_connection(exception): 

42 db = getattr(g, '_database', None) 

43 if db is not None: 

44 db.close() 

45 

46