eduard/cobra-boards
Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.
A remix template · 1 remixes · Remix eduard/cobra-boards
app/boards/infrastructure/dodil.py
"""The Dodil side: a service-account token, DataK3 on the Postgres wire, and Dodil Models.
The remix generated the service account and put its id and secret into the repo's CI secrets;
.dodil/deploy.yaml hands them to the app as environment variables.
"""
import os
import threading
import time
from contextlib import contextmanager
import httpx
import psycopg
from boards.application.ports import AssistantUnavailable
TOKEN_URL = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
PG_HOST = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
MODELS_BASE = os.environ.get("MODELS_BASE", "https://api.dodil.io/v1").rstrip("/")
CHAT_MODEL = os.environ.get("CHAT_MODEL", "kimi-k2.6")
_lock, _tok = threading.Lock(), {"v": None, "exp": 0.0}
def token() -> str:
"""A short-lived service-account token (client_credentials), cached until near expiry.
It is the DataK3 password AND the Models bearer."""
with _lock:
if _tok["v"] and time.time() < _tok["exp"] - 60:
return _tok["v"]
r = httpx.post(TOKEN_URL, timeout=15, data={
"grant_type": "client_credentials",
"client_id": os.environ["DODIL_SERVICE_ACCOUNT_ID"],
"client_secret": os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]})
r.raise_for_status()
d = r.json()
_tok.update(v=d["access_token"], exp=time.time() + float(d.get("expires_in", 300)))
return _tok["v"]
_local = threading.local()
# DataK3 serves 16 concurrent reads per bucket; stay under it (dodil guide: DB_CONCURRENCY 8-10).
_gate = threading.BoundedSemaphore(int(os.environ.get("DB_CONCURRENCY", "8")))
def _open() -> psycopg.Connection:
"""DataK3 speaks Postgres: the database is the bucket, TLS is required."""
return psycopg.connect(host=PG_HOST, port=5432, dbname=os.environ["BUCKET"], user="token",
password=token(), sslmode="require", autocommit=True,
connect_timeout=15)
@contextmanager
def connect():
"""A connection kept per worker thread and reused, so a page that needs eight reads pays for
one TLS handshake, not eight. It is replaced when the token it signed in with is about to
expire, or when the server dropped it."""
with _gate:
conn, exp = getattr(_local, "conn", None), getattr(_local, "exp", 0.0)
if conn is None or conn.closed or conn.broken or time.time() > exp - 60:
if conn is not None and not conn.closed:
conn.close()
conn = _open()
_local.conn, _local.exp = conn, _tok["exp"]
try:
yield conn
except psycopg.OperationalError:
conn.close() # a dropped link: the next call opens a fresh one
_local.conn = None
raise
EMBED_MODEL = os.environ.get("EMBED_MODEL", "arctic-embed-m-v2")
EMBED_DIM = 768 # matches VECTOR(768) in schema.sql
def _unwrap(body: dict) -> dict:
"""The gateway wraps the OpenAI body as {"status": ..., "data": {...}}."""
inner = body.get("data")
return inner if isinstance(inner, dict) else body
def chat(prompt: str, system: str = "") -> str:
"""One chat call to Dodil Models (OpenAI-compatible)."""
messages = ([{"role": "system", "content": system}] if system else []) + \
[{"role": "user", "content": prompt}]
r = httpx.post(f"{MODELS_BASE}/chat/completions", timeout=120,
headers={"Authorization": f"Bearer {token()}"},
json={"model": CHAT_MODEL, "messages": messages})
r.raise_for_status()
return _unwrap(r.json())["choices"][0]["message"]["content"] or ""
def embed(text: str) -> list:
r = httpx.post(f"{MODELS_BASE}/embeddings", timeout=90,
headers={"Authorization": f"Bearer {token()}"},
json={"model": EMBED_MODEL, "input": [text[:4000]]})
r.raise_for_status()
body = r.json()
inner = body.get("data", body)
items = inner if isinstance(inner, list) else inner.get("data", [])
vec = items[0]["embedding"]
return [float(x) for x in vec[:EMBED_DIM]]
OBJECTS_BASE = os.environ.get("OBJECTS_BASE", "https://object.uk-lon-1.dodil.io").rstrip("/")
class DodilObjects:
"""Attachment bytes in the bucket's own object storage, with the app's service account
(k3.editor on the bucket covers objects too). Nothing is public: the app streams a file only
to someone who can see its board."""
def __init__(self, bucket: str = ""):
self.bucket = bucket # read from BUCKET on first use, like the database connection
def _url(self, key: str) -> str:
return f"{OBJECTS_BASE}/{self.bucket or os.environ['BUCKET']}/{key}"
def put(self, key: str, data: bytes, content_type: str) -> None:
r = httpx.put(self._url(key), content=data, timeout=60,
headers={"Authorization": f"Bearer {token()}",
"Content-Type": content_type or "application/octet-stream"})
r.raise_for_status()
def get(self, key: str) -> bytes:
r = httpx.get(self._url(key), timeout=60, headers={"Authorization": f"Bearer {token()}"})
r.raise_for_status()
return r.content
def delete(self, key: str) -> None:
r = httpx.delete(self._url(key), timeout=30,
headers={"Authorization": f"Bearer {token()}"})
if r.status_code not in (200, 202, 204, 404):
r.raise_for_status()
def _model_call(fn, *args, **kwargs):
"""A model call that failed on the network or the server becomes one clear error, so the use
case can tell the person "try again" instead of the page showing a 500."""
try:
return fn(*args, **kwargs)
except (httpx.HTTPError, KeyError, IndexError, ValueError) as e:
raise AssistantUnavailable(str(e)[:200]) from e
class ModelsAssistant:
"""The board's AI, on Dodil Models only. It suggests and summarises; it never writes."""
def checklist(self, title: str, description: str) -> list:
text = _model_call(chat,"Task: " + title + ("\nDetails: " + description if description else ""),
system="Break the task into 3 to 6 concrete, checkable steps a small team "
"would actually do. One step per line, imperative, no numbering, "
"no commentary.")
return [line.strip("-*•0123456789.) ").strip() for line in text.splitlines()
if line.strip()][:8]
def summary(self, board_title: str, facts: str) -> str:
return _model_call(chat, facts, system=f"You write the morning briefing for the team board "
f"'{board_title}'. Use ONLY the facts given; never invent a "
f"card, a person or a number. 3 to 5 short sentences: what's "
f"at risk first (overdue, blocked), then what's due today, "
f"then progress. Refer to cards as #number.").strip()
def embed(self, text: str) -> list:
return _model_call(embed, text)