dmslanes/lanes
Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT
A remix template · 0 remixes · Remix dmslanes/lanes
app/infrastructure/k3.py
"""Repository on DataK3, Dodil's tables engine, over the Postgres wire.
The bucket is the database; user is the literal `token`; the password is a
short-lived token minted from the app's service account (deployed) or
DODIL_TOKEN (local). Each call runs in autocommit: DataK3 has no
read-your-writes inside an open transaction, and every use case here is a few
small statements.
"""
from __future__ import annotations
import json
import os
import threading
import time
from datetime import date, datetime, timezone
import httpx
import psycopg
from app.domain.model import Activity, Board, BoardList, Card, Conflict, Membership, Role, Workspace
TOKEN_URL = os.environ.get("DODIL_OIDC_TOKEN_URL", "https://id.dodil.io/realms/dodil/protocol/openid-connect/token")
_token_lock = threading.Lock()
_token = {"value": None, "exp": 0.0}
def access_token() -> str:
"""Service-account token (client credentials), cached until a minute before it expires."""
sa_id, sa_secret = os.environ.get("DODIL_SERVICE_ACCOUNT_ID"), os.environ.get("DODIL_SERVICE_ACCOUNT_SECRET")
if not (sa_id and sa_secret):
return os.environ["DODIL_TOKEN"]
with _token_lock:
if _token["value"] and time.time() < _token["exp"] - 60:
return _token["value"]
r = httpx.post(TOKEN_URL, data={"grant_type": "client_credentials", "client_id": sa_id,
"client_secret": sa_secret}, timeout=15)
r.raise_for_status()
d = r.json()
_token.update(value=d["access_token"], exp=time.time() + float(d.get("expires_in", 300)))
return _token["value"]
def _iso(v):
return v.isoformat() if isinstance(v, (date, datetime)) else v
def _dt(v):
return datetime.fromisoformat(v) if v else None
class K3Repository:
def __init__(self, bucket: str | None = None):
self.bucket = bucket or os.environ["K3_BUCKET"] # no default: never write into a guessed bucket
self.host = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
self.port = int(os.environ.get("PG_PORT", "5432"))
self._local = threading.local()
self._gate = threading.BoundedSemaphore(int(os.environ.get("DB_CONCURRENCY", "8"))) # below K3's 16 reads
# ---------- plumbing ----------
def _conn(self):
c = getattr(self._local, "conn", None)
if c is None or c.closed:
c = psycopg.connect(host=self.host, port=self.port, dbname=self.bucket, user="token",
password=access_token(), sslmode="require", autocommit=True,
connect_timeout=15)
self._local.conn = c
return c
def _run(self, sql: str, params=(), fetch: bool = True):
for attempt in range(3):
try:
with self._gate, self._conn().cursor() as cur:
cur.execute(sql, params)
rows = cur.fetchall() if fetch and cur.description else []
return rows, cur.rowcount
except psycopg.errors.UniqueViolation:
raise
except (psycopg.OperationalError, psycopg.errors.SerializationFailure):
self._local.conn = None
if attempt == 2:
raise
time.sleep(0.3 * (attempt + 1))
def _q(self, sql, params=()):
return self._run(sql, params)[0]
def _x(self, sql, params=()):
return self._run(sql, params, fetch=False)[1]
def self_check(self) -> dict:
tables = ("workspaces", "memberships", "boards", "lists", "cards", "card_numbers", "activity")
return {t: self._q(f"SELECT count(*) FROM {t}")[0][0] for t in tables}
# ---------- workspaces and people ----------
def add_workspace(self, ws):
self._x("INSERT INTO workspaces (id, name, created_by, created_at) VALUES (%s, %s, %s, %s)",
(ws.id, ws.name, ws.created_by, _iso(ws.created_at)))
def get_workspace(self, workspace_id):
r = self._q("SELECT id, name, created_by, created_at FROM workspaces WHERE id = %s", (workspace_id,))
return Workspace(r[0][0], r[0][1], r[0][2], _dt(r[0][3])) if r else None
@staticmethod
def _m(row):
return Membership(row[0], row[1], Role(row[2]))
def get_membership(self, workspace_id, email):
r = self._q("SELECT workspace_id, email, role FROM memberships WHERE id = %s", (f"{workspace_id}:{email}",))
return self._m(r[0]) if r else None
def memberships_of_workspace(self, workspace_id):
return [self._m(r) for r in self._q(
"SELECT workspace_id, email, role FROM memberships WHERE workspace_id = %s", (workspace_id,))]
def memberships_of_person(self, email):
return [self._m(r) for r in self._q(
"SELECT workspace_id, email, role FROM memberships WHERE email = %s", (email,))]
def save_membership(self, m):
self._x("INSERT INTO memberships (id, workspace_id, email, role) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET role = EXCLUDED.role",
(f"{m.workspace_id}:{m.email}", m.workspace_id, m.email, m.role.value))
def delete_membership(self, workspace_id, email):
self._x("DELETE FROM memberships WHERE id = %s", (f"{workspace_id}:{email}",))
# ---------- boards and lists ----------
def add_board(self, b):
self._x("INSERT INTO boards (id, workspace_id, name, created_at) VALUES (%s, %s, %s, %s)",
(b.id, b.workspace_id, b.name, _iso(b.created_at)))
def get_board(self, board_id):
r = self._q("SELECT id, workspace_id, name, created_at FROM boards WHERE id = %s", (board_id,))
return Board(r[0][0], r[0][1], r[0][2], _dt(r[0][3])) if r else None
def boards_of_workspaces(self, workspace_ids):
if not workspace_ids:
return []
rows = self._q("SELECT id, workspace_id, name, created_at FROM boards WHERE workspace_id = ANY(%s)",
(list(workspace_ids),))
return [Board(r[0], r[1], r[2], _dt(r[3])) for r in rows]
def add_list(self, l):
self._x("INSERT INTO lists (id, board_id, name, position) VALUES (%s, %s, %s, %s)",
(l.id, l.board_id, l.name, l.position))
def get_list(self, list_id):
r = self._q("SELECT id, board_id, name, position FROM lists WHERE id = %s", (list_id,))
return BoardList(*r[0]) if r else None
def lists_of_board(self, board_id):
rows = self._q("SELECT id, board_id, name, position FROM lists WHERE board_id = %s", (board_id,))
return sorted([BoardList(*r) for r in rows], key=lambda l: l.position)
# ---------- cards ----------
_CARD_COLS = ("id, workspace_id, board_id, list_id, number, title, description, assignee, due_date, "
"labels, position, created_at, updated_at, version, blocked_by")
@staticmethod
def _card(r):
return Card(id=r[0], workspace_id=r[1], board_id=r[2], list_id=r[3], number=int(r[4]), title=r[5],
description=r[6] or "", assignee=r[7], due_date=date.fromisoformat(r[8]) if r[8] else None,
labels=json.loads(r[9]) if r[9] else [], position=int(r[10]), created_at=_dt(r[11]),
updated_at=_dt(r[12]), version=int(r[13]),
blocked_by=json.loads(r[14]) if r[14] else [])
def claim_card_number(self):
# A claim is a plain INSERT: under a race one writer wins, the other gets 23505 and tries the next.
# The next, not max()+1 again: DataK3 keeps a deleted key taken (measured 2026-10-29), so a
# number freed by a DELETE still collides and re-reading max() would loop on it.
n = int(self._q("SELECT max(number) FROM card_numbers")[0][0] or 0) + 1
for _ in range(50):
try:
self._x("INSERT INTO card_numbers (number, claimed_at) VALUES (%s, %s)",
(n, datetime.now(timezone.utc).isoformat()))
return n
except psycopg.errors.UniqueViolation:
n += 1
raise Conflict("could not number the card, try again")
def _card_params(self, c):
return (c.workspace_id, c.board_id, c.list_id, c.number, c.title, c.description, c.assignee,
_iso(c.due_date), json.dumps(c.labels), c.position, _iso(c.created_at), _iso(c.updated_at),
c.version, json.dumps(c.blocked_by))
def add_card(self, c):
self._x(f"INSERT INTO cards ({self._CARD_COLS}) VALUES ({', '.join(['%s'] * 15)})",
(c.id, *self._card_params(c)))
def save_card(self, c, expected_version):
current = self.get_card(c.id)
if current is None or current.version != expected_version:
raise Conflict("this card changed since you opened it; reload and try again")
n = self._x("UPDATE cards SET workspace_id=%s, board_id=%s, list_id=%s, number=%s, title=%s, "
"description=%s, assignee=%s, due_date=%s, labels=%s, position=%s, created_at=%s, "
"updated_at=%s, version=%s, blocked_by=%s WHERE id=%s AND version=%s",
(*self._card_params(c), c.id, expected_version))
if n == 0:
raise Conflict("this card changed since you opened it; reload and try again")
def delete_card(self, card_id):
self._x("DELETE FROM cards WHERE id = %s", (card_id,))
def get_card(self, card_id):
r = self._q(f"SELECT {self._CARD_COLS} FROM cards WHERE id = %s", (card_id,))
return self._card(r[0]) if r else None
def get_card_by_number(self, number):
r = self._q(f"SELECT {self._CARD_COLS} FROM cards WHERE number = %s", (number,))
return self._card(r[0]) if r else None
def cards_of_board(self, board_id):
return [self._card(r) for r in self._q(f"SELECT {self._CARD_COLS} FROM cards WHERE board_id = %s",
(board_id,))]
def cards_of_workspaces(self, workspace_ids):
if not workspace_ids:
return []
return [self._card(r) for r in self._q(
f"SELECT {self._CARD_COLS} FROM cards WHERE workspace_id = ANY(%s)", (list(workspace_ids),))]
# ---------- history ----------
def add_activity(self, a):
self._x("INSERT INTO activity (id, card_id, board_id, actor, happened_at, action, changes) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
(a.id, a.card_id, a.board_id, a.actor, _iso(a.at), a.action, json.dumps(a.changes)))
@staticmethod
def _a(r):
return Activity(r[0], r[1], r[2], r[3], _dt(r[4]), r[5], json.loads(r[6]) if r[6] else {})
def activity_of_card(self, card_id):
return [self._a(r) for r in self._q(
"SELECT id, card_id, board_id, actor, happened_at, action, changes FROM activity WHERE card_id = %s", (card_id,))]
def activity_of_board(self, board_id, limit):
rows = self._q("SELECT id, card_id, board_id, actor, happened_at, action, changes FROM activity "
"WHERE board_id = %s ORDER BY happened_at DESC LIMIT %s", (board_id, limit))
return [self._a(r) for r in rows]