rahulpawar/taskdeck

Lanebook, a team task board (boards, lists, cards, roles, history) you run from the web or from Claude/ChatGPT over MCP. FastAPI + DataK3 + appid + Models.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix rahulpawar/taskdeck

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. No outside database, no
outside AI key: the same short-lived token is the DataK3 password and the Models bearer.
"""
import os
import threading
import time

import httpx
import psycopg

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")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "jina-embeddings-v4")

_lock, _tok = threading.Lock(), {"v": None, "exp": 0.0}


def token() -> str:
    """A short-lived service-account token (client_credentials), cached until near expiry."""
    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"]


def _open(tok: str) -> 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=tok, sslmode="require", autocommit=True, connect_timeout=15)


_local = threading.local()


class connect:
    """`with connect() as c:` hands out this thread's open DataK3 connection, opening a new one
    when there is none, it broke, or the token was renewed. Reusing it saves a TLS handshake and
    a sign-in per query; a connection that failed is dropped so the next call starts clean."""

    def __enter__(self) -> psycopg.Connection:
        tok = token()
        c = getattr(_local, "conn", None)
        if c is None or c.closed or c.broken or getattr(_local, "tok", None) != tok:
            if c is not None and not c.closed:
                try:
                    c.close()
                except Exception:
                    pass
            _local.conn, _local.tok = _open(tok), tok
        return _local.conn

    def __exit__(self, exc_type, exc, tb):
        if exc_type is not None and issubclass(exc_type, psycopg.OperationalError):
            try:
                _local.conn.close()
            finally:
                _local.conn = None
        return False


def _unwrap(body: dict) -> dict:
    """The gateway wraps the OpenAI body as {"status": ..., "data": {...}}."""
    inner = body.get("data")
    if isinstance(inner, dict) and ("choices" in inner or "data" in inner):
        return inner
    return body


def chat(prompt: str, system: str = "") -> str:
    msgs = ([{"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": msgs})
    r.raise_for_status()
    return _unwrap(r.json())["choices"][0]["message"]["content"] or ""


def embed(texts: list) -> list:
    r = httpx.post(f"{MODELS_BASE}/embeddings", timeout=120,
                   headers={"Authorization": f"Bearer {token()}"},
                   json={"model": EMBED_MODEL, "input": [t[:2000] for t in texts]})
    r.raise_for_status()
    rows = _unwrap(r.json())["data"]
    return [row["embedding"] for row in sorted(rows, key=lambda x: x.get("index", 0))]


class ModelsAssistant:
    def checklist(self, card_title: str, description: str = "") -> list:
        text = chat("Break this task into a short checklist of 3 to 6 concrete steps. "
                    "One step per line, no numbering, nothing else.\n\nTask: " + card_title
                    + (f"\nDetails: {description[:1500]}" if description else ""))
        steps = [line.strip().lstrip("-*•[]0123456789.) ").strip() for line in text.splitlines()]
        return [s for s in steps if s][:8]

    def summary(self, board_title: str, facts: str) -> str:
        return chat(facts, system=(
            f"You write the daily summary of the team board '{board_title}'. In at most 6 short "
            "lines: what moved recently, what is overdue (and whose), and what looks next. "
            "Plain text, no headings, only facts from the data given.")).strip()

    def embed(self, texts: list) -> list:
        return embed(texts)