Coverage for database.py: 44%

1import duckdb 

2import os 

3from typing import Generator 

4 

5# In production this will point to the duckdb file 

6DATABASE_URL = os.getenv("DATABASE_URL", "whipstats_production.duckdb") 

7 

8# Global connection to be initialized in lifespan 

9db_conn: duckdb.DuckDBPyConnection | None = None 

10 

11def init_db(path: str = DATABASE_URL): 

12 global db_conn 

13 # Open read-only connection. DuckDB allows multiple read-only connections  

14 # to the same file, and connection is thread-safe for reading. 

15 db_conn = duckdb.connect(path, read_only=True) 

16 

17def close_db(): 

18 global db_conn 

19 if db_conn: 

20 db_conn.close() 

21 db_conn = None 

22 

23def get_db() -> Generator[duckdb.DuckDBPyConnection, None, None]: 

24 """ 

25 FastAPI dependency that yields a thread-local cursor. 

26 This allows concurrent read queries without blocking each other. 

27 """ 

28 if not db_conn: 

29 raise RuntimeError("Database connection is not initialized") 

30 cursor = db_conn.cursor() 

31 try: 

32 yield cursor 

33 finally: 

34 cursor.close()