Coverage for app\db.py: 96%

24 statements  

« prev     ^ index     » next       coverage.py v7.10.3, created at 2025-09-10 15:54 +0900

1from dotenv import load_dotenv 

2import os 

3from typing import List, Tuple 

4import psycopg2 

5 

6 

7def _load_env_once(): 

8 # Load .env from project root 

9 load_dotenv() 

10 

11 

12def get_conn(): 

13 """Create and return a psycopg2 connection using env vars.""" 

14 _load_env_once() 

15 user = os.getenv("PG_USER") 

16 password = os.getenv("PG_PASSWORD") 

17 host = os.getenv("PG_HOST") 

18 port = os.getenv("PG_PORT") 

19 dbname = os.getenv("PG_DB") 

20 

21 if not all([user, password, host, port, dbname]): 

22 raise RuntimeError("Postgres connection parameters are not fully set in environment") 

23 

24 return psycopg2.connect( 

25 user=user, 

26 password=password, 

27 host=host, 

28 port=int(port), 

29 dbname=dbname, 

30 ) 

31 

32 

33def list_tables() -> List[Tuple[str, str]]: 

34 """Return list of (schema, table_name) for user tables.""" 

35 conn = get_conn() 

36 try: 

37 with conn.cursor() as cur: 

38 cur.execute( 

39 """ 

40 SELECT table_schema, table_name 

41 FROM information_schema.tables 

42 WHERE table_type='BASE TABLE' AND table_schema NOT IN ('pg_catalog','information_schema') 

43 ORDER BY table_schema, table_name; 

44 """ 

45 ) 

46 rows = cur.fetchall() 

47 return [(r[0], r[1]) for r in rows] 

48 finally: 

49 conn.close()