fez5587andClaude Fable 5
c8bfa0c0ab
- docker-compose.yml: Postgres 16 (schema auto-applied) + Redis, nonstandard host ports 5433/6380. - engine/db.py: async psycopg pool, append_event in caller's transaction. - engine/souls.py: versioned role charters (manager/engineer/qa) + reporting protocol appended to every soul. - engine/main.py: POST/GET companies, employee hire with Hermes profile creation (outside tx, retryable; DB authoritative), event feed, archive with profile cleanup, global pause/resume in Redis. - Verified live end-to-end: create -> 4 events + 3 real Hermes profiles; pause flips health; archive 204 removes profiles, keeps rows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
"""Postgres access — one async pool, event append, thin query helpers."""
|
|
import json
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from psycopg.rows import dict_row
|
|
from psycopg.types.json import Json
|
|
from psycopg_pool import AsyncConnectionPool
|
|
|
|
from engine.config import settings
|
|
|
|
pool = AsyncConnectionPool(settings.database_url, open=False, kwargs={"row_factory": dict_row})
|
|
|
|
|
|
async def open_pool() -> None:
|
|
await pool.open()
|
|
|
|
|
|
async def close_pool() -> None:
|
|
await pool.close()
|
|
|
|
|
|
async def fetch_all(query: str, *args: Any) -> list[dict]:
|
|
async with pool.connection() as conn:
|
|
cur = await conn.execute(query, args)
|
|
return await cur.fetchall()
|
|
|
|
|
|
async def fetch_one(query: str, *args: Any) -> dict | None:
|
|
async with pool.connection() as conn:
|
|
cur = await conn.execute(query, args)
|
|
return await cur.fetchone()
|
|
|
|
|
|
async def execute(query: str, *args: Any) -> None:
|
|
async with pool.connection() as conn:
|
|
await conn.execute(query, args)
|
|
|
|
|
|
async def append_event(
|
|
conn: Any,
|
|
*,
|
|
actor: str,
|
|
event_type: str,
|
|
company_id: UUID | None = None,
|
|
work_item_id: UUID | None = None,
|
|
correlation_id: UUID | None = None,
|
|
source: str = "engine",
|
|
payload: dict | None = None,
|
|
policy_decision: dict | None = None,
|
|
provenance: dict | None = None,
|
|
) -> dict:
|
|
"""Append to the immutable log. Caller passes its open connection so the
|
|
event lands in the same transaction as the projection update."""
|
|
cur = await conn.execute(
|
|
"""
|
|
INSERT INTO event (company_id, actor, event_type, work_item_id,
|
|
correlation_id, source, payload, policy_decision, provenance)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING event_id, created_at
|
|
""",
|
|
(
|
|
company_id, actor, event_type, work_item_id, correlation_id, source,
|
|
Json(payload or {}),
|
|
Json(policy_decision) if policy_decision else None,
|
|
Json(provenance) if provenance else None,
|
|
),
|
|
)
|
|
return await cur.fetchone()
|
|
|
|
|
|
def as_json(row: dict) -> dict:
|
|
"""Make a DB row JSON-serializable (UUID/datetime to str)."""
|
|
return json.loads(json.dumps(row, default=str))
|