| """ |
| Thin async Postgres access layer + card (de)serialization. |
| Kept separate so mcp_server, conductor, and reaper share one pool + one mapper. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| from datetime import datetime, timedelta, timezone |
| from typing import Optional |
|
|
| import asyncpg |
|
|
| from .contract import Card, Column, Role, ThreadEntry |
|
|
| _POOL: Optional[asyncpg.Pool] = None |
|
|
|
|
| async def pool() -> asyncpg.Pool: |
| global _POOL |
| if _POOL is None: |
| dsn = os.environ.get( |
| "DATABASE_URL", |
| "postgresql://kanbot:kanbot@localhost:5432/kanbot", |
| ) |
| _POOL = await asyncpg.create_pool(dsn, min_size=1, max_size=10) |
| return _POOL |
|
|
|
|
| def _row_to_card(row: asyncpg.Record) -> Card: |
| return Card( |
| id=row["id"], |
| title=row["title"], |
| column=Column(row["column"]), |
| agent_role=Role(row["agent_role"]), |
| spec=row["spec"], |
| acceptance_criteria=json.loads(row["acceptance_criteria"]), |
| assigned_worker=row["assigned_worker"], |
| parent=row["parent"], |
| children=json.loads(row["children"]), |
| thread=[ThreadEntry(**e) for e in json.loads(row["thread"])], |
| lease_expires_at=row["lease_expires_at"].isoformat() |
| if row["lease_expires_at"] else None, |
| attempt=row["attempt"], |
| requires_human=row["requires_human"], |
| tags=json.loads(row["tags"]), |
| created_at=row["created_at"].isoformat(), |
| updated_at=row["updated_at"].isoformat(), |
| ) |
|
|
|
|
| async def get_card(card_id: str) -> Optional[Card]: |
| async with (await pool()).acquire() as c: |
| row = await c.fetchrow("SELECT * FROM cards WHERE id = $1", card_id) |
| return _row_to_card(row) if row else None |
|
|
|
|
| async def list_column(column: Column) -> list[Card]: |
| async with (await pool()).acquire() as c: |
| rows = await c.fetch( |
| 'SELECT * FROM cards WHERE "column" = $1 ORDER BY created_at', column.value |
| ) |
| return [_row_to_card(r) for r in rows] |
|
|
|
|
| async def next_card_id() -> str: |
| async with (await pool()).acquire() as c: |
| n = await c.fetchval("SELECT count(*) FROM cards") |
| return f"CARD-{n + 1}" |
|
|
|
|
| async def insert_card(card: Card) -> Card: |
| async with (await pool()).acquire() as c: |
| await c.execute( |
| """INSERT INTO cards |
| (id,title,"column",agent_role,spec,acceptance_criteria,parent, |
| children,thread,tags,requires_human,attempt) |
| VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)""", |
| card.id, card.title, card.column.value, card.agent_role.value, |
| card.spec, json.dumps(card.acceptance_criteria), card.parent, |
| json.dumps(card.children), json.dumps([e.__dict__ for e in card.thread]), |
| json.dumps(card.tags), card.requires_human, card.attempt, |
| ) |
| return card |
|
|
|
|
| async def append_thread(card_id: str, entry: ThreadEntry) -> None: |
| async with (await pool()).acquire() as c: |
| await c.execute( |
| "UPDATE cards SET thread = thread || $2::jsonb WHERE id = $1", |
| card_id, json.dumps([entry.__dict__]), |
| ) |
|
|
|
|
| async def set_lease(card_id: str, worker: str, seconds: int) -> None: |
| expiry = datetime.now(timezone.utc) + timedelta(seconds=seconds) |
| async with (await pool()).acquire() as c: |
| await c.execute( |
| "UPDATE cards SET assigned_worker=$2, lease_expires_at=$3 WHERE id=$1", |
| card_id, worker, expiry, |
| ) |
|
|
|
|
| async def clear_lease(card_id: str) -> None: |
| async with (await pool()).acquire() as c: |
| await c.execute( |
| "UPDATE cards SET assigned_worker=NULL, lease_expires_at=NULL WHERE id=$1", |
| card_id, |
| ) |
|
|
|
|
| async def add_child(parent_id: str, child_id: str) -> None: |
| async with (await pool()).acquire() as c: |
| await c.execute( |
| "UPDATE cards SET children = children || $2::jsonb WHERE id = $1", |
| parent_id, json.dumps([child_id]), |
| ) |
|
|
