Coverage for app\db.py: 96%
24 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-09-10 15:54 +0900
« 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
7def _load_env_once():
8 # Load .env from project root
9 load_dotenv()
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")
21 if not all([user, password, host, port, dbname]):
22 raise RuntimeError("Postgres connection parameters are not fully set in environment")
24 return psycopg2.connect(
25 user=user,
26 password=password,
27 host=host,
28 port=int(port),
29 dbname=dbname,
30 )
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()