17759ef564
- apps/api: psycopg v3 repositories, schema.sql + migration runner, state machine implementing the data-model transition table, approval gate with sha256 hash match + 24h expiry enforced at send time, EchoTransport pluggable sender, LLM calls routed through packages.llm-gateway with mock mode, 47 pytest tests green - apps/api/Dockerfile.test: python 3.13-slim test image (DinD-safe: migrations copied as directory) - docker-compose.yml: add api-test service on compose network (host port publishing is broken in this sandbox; container-to-container networking used) - Fix: replace masked placeholder password in config.py/conftest.py defaults
80 lines
No EOL
1.9 KiB
Python
80 lines
No EOL
1.9 KiB
Python
"""Database connection pool and helpers (psycopg v3)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
from psycopg_pool import ConnectionPool
|
|
|
|
from app.config import DATABASE_URL
|
|
|
|
_pool: ConnectionPool | None = None
|
|
|
|
|
|
def get_pool() -> ConnectionPool:
|
|
global _pool
|
|
if _pool is None:
|
|
_pool = ConnectionPool(
|
|
conninfo=DATABASE_URL,
|
|
kwargs={"row_factory": dict_row},
|
|
min_size=1,
|
|
max_size=8,
|
|
open=True,
|
|
)
|
|
return _pool
|
|
|
|
|
|
def close_pool() -> None:
|
|
global _pool
|
|
if _pool is not None:
|
|
_pool.close()
|
|
_pool = None
|
|
|
|
|
|
def get_conn() -> psycopg.Connection:
|
|
"""Get a raw connection (for use as context manager)."""
|
|
return psycopg.connect(DATABASE_URL, row_factory=dict_row)
|
|
|
|
|
|
# --- repository helpers ---
|
|
|
|
|
|
def fetch_one(sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
|
|
pool = get_pool()
|
|
with pool.connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params)
|
|
return cur.fetchone()
|
|
|
|
|
|
def fetch_all(sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
|
|
pool = get_pool()
|
|
with pool.connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params)
|
|
return cur.fetchall()
|
|
|
|
|
|
def execute(sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
|
|
pool = get_pool()
|
|
with pool.connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, params)
|
|
conn.commit()
|
|
if cur.description:
|
|
return cur.fetchone()
|
|
return None
|
|
|
|
|
|
# JSON adaptation for psycopg v3
|
|
def adapt_jsonb(value: Any) -> str:
|
|
return json.dumps(value)
|
|
|
|
|
|
def convert_uuid(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
return str(value) |