From 3db73332addb0d565b80a2c601a1cedc0205ae0d Mon Sep 17 00:00:00 2001 From: johnfrum1234 Date: Sun, 15 Mar 2026 13:16:12 +0200 Subject: [PATCH] =?UTF-8?q?Add=20core/models.py=20=E2=80=94=20data=20acces?= =?UTF-8?q?s=20functions=20for=20all=209=20tables?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 20 functions covering: projects, tasks, decisions, modules, agent_logs, pipelines, support tickets, and dashboard stats. Parameterized queries, JSON encode/decode, no ORM. 21 tests, all passing. Co-Authored-By: Claude Opus 4.6 (1M context) --- core/models.py | 445 +++++++++++++++++++++++++++++++++++++++++++ tests/__init__.py | 0 tests/test_models.py | 240 +++++++++++++++++++++++ 3 files changed, 685 insertions(+) create mode 100644 core/models.py create mode 100644 tests/__init__.py create mode 100644 tests/test_models.py diff --git a/core/models.py b/core/models.py new file mode 100644 index 0000000..713d08c --- /dev/null +++ b/core/models.py @@ -0,0 +1,445 @@ +""" +Kin — data access functions for all tables. +Pure functions: (conn, params) → dict | list[dict]. No ORM, no classes. +""" + +import json +import sqlite3 +from datetime import datetime +from typing import Any + + +def _row_to_dict(row: sqlite3.Row | None) -> dict | None: + """Convert sqlite3.Row to dict with JSON fields decoded.""" + if row is None: + return None + d = dict(row) + for key, val in d.items(): + if isinstance(val, str) and val.startswith(("[", "{")): + try: + d[key] = json.loads(val) + except (json.JSONDecodeError, ValueError): + pass + return d + + +def _rows_to_list(rows: list[sqlite3.Row]) -> list[dict]: + """Convert list of sqlite3.Row to list of dicts.""" + return [_row_to_dict(r) for r in rows] + + +def _json_encode(val: Any) -> Any: + """Encode lists/dicts to JSON strings for storage.""" + if isinstance(val, (list, dict)): + return json.dumps(val, ensure_ascii=False) + return val + + +# --------------------------------------------------------------------------- +# Projects +# --------------------------------------------------------------------------- + +def create_project( + conn: sqlite3.Connection, + id: str, + name: str, + path: str, + tech_stack: list | None = None, + status: str = "active", + priority: int = 5, + pm_prompt: str | None = None, + claude_md_path: str | None = None, + forgejo_repo: str | None = None, +) -> dict: + """Create a new project and return it as dict.""" + conn.execute( + """INSERT INTO projects (id, name, path, tech_stack, status, priority, + pm_prompt, claude_md_path, forgejo_repo) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (id, name, path, _json_encode(tech_stack), status, priority, + pm_prompt, claude_md_path, forgejo_repo), + ) + conn.commit() + return get_project(conn, id) + + +def get_project(conn: sqlite3.Connection, id: str) -> dict | None: + """Get project by id.""" + row = conn.execute("SELECT * FROM projects WHERE id = ?", (id,)).fetchone() + return _row_to_dict(row) + + +def list_projects(conn: sqlite3.Connection, status: str | None = None) -> list[dict]: + """List projects, optionally filtered by status.""" + if status: + rows = conn.execute( + "SELECT * FROM projects WHERE status = ? ORDER BY priority, name", + (status,), + ).fetchall() + else: + rows = conn.execute( + "SELECT * FROM projects ORDER BY priority, name" + ).fetchall() + return _rows_to_list(rows) + + +def update_project(conn: sqlite3.Connection, id: str, **fields) -> dict: + """Update project fields. Returns updated project.""" + if not fields: + return get_project(conn, id) + for key in ("tech_stack",): + if key in fields: + fields[key] = _json_encode(fields[key]) + sets = ", ".join(f"{k} = ?" for k in fields) + vals = list(fields.values()) + [id] + conn.execute(f"UPDATE projects SET {sets} WHERE id = ?", vals) + conn.commit() + return get_project(conn, id) + + +# --------------------------------------------------------------------------- +# Tasks +# --------------------------------------------------------------------------- + +def create_task( + conn: sqlite3.Connection, + id: str, + project_id: str, + title: str, + status: str = "pending", + priority: int = 5, + assigned_role: str | None = None, + parent_task_id: str | None = None, + brief: dict | None = None, + spec: dict | None = None, + forgejo_issue_id: int | None = None, +) -> dict: + """Create a task linked to a project.""" + conn.execute( + """INSERT INTO tasks (id, project_id, title, status, priority, + assigned_role, parent_task_id, brief, spec, forgejo_issue_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (id, project_id, title, status, priority, assigned_role, + parent_task_id, _json_encode(brief), _json_encode(spec), + forgejo_issue_id), + ) + conn.commit() + return get_task(conn, id) + + +def get_task(conn: sqlite3.Connection, id: str) -> dict | None: + """Get task by id.""" + row = conn.execute("SELECT * FROM tasks WHERE id = ?", (id,)).fetchone() + return _row_to_dict(row) + + +def list_tasks( + conn: sqlite3.Connection, + project_id: str | None = None, + status: str | None = None, +) -> list[dict]: + """List tasks with optional project/status filters.""" + query = "SELECT * FROM tasks WHERE 1=1" + params: list = [] + if project_id: + query += " AND project_id = ?" + params.append(project_id) + if status: + query += " AND status = ?" + params.append(status) + query += " ORDER BY priority, created_at" + return _rows_to_list(conn.execute(query, params).fetchall()) + + +def update_task(conn: sqlite3.Connection, id: str, **fields) -> dict: + """Update task fields. Auto-sets updated_at.""" + if not fields: + return get_task(conn, id) + json_cols = ("brief", "spec", "review", "test_result", "security_result") + for key in json_cols: + if key in fields: + fields[key] = _json_encode(fields[key]) + fields["updated_at"] = datetime.now().isoformat() + sets = ", ".join(f"{k} = ?" for k in fields) + vals = list(fields.values()) + [id] + conn.execute(f"UPDATE tasks SET {sets} WHERE id = ?", vals) + conn.commit() + return get_task(conn, id) + + +# --------------------------------------------------------------------------- +# Decisions +# --------------------------------------------------------------------------- + +def add_decision( + conn: sqlite3.Connection, + project_id: str, + type: str, + title: str, + description: str, + category: str | None = None, + tags: list | None = None, + task_id: str | None = None, +) -> dict: + """Record a decision, gotcha, or convention for a project.""" + cur = conn.execute( + """INSERT INTO decisions (project_id, task_id, type, category, + title, description, tags) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (project_id, task_id, type, category, title, description, + _json_encode(tags)), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM decisions WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return _row_to_dict(row) + + +def get_decisions( + conn: sqlite3.Connection, + project_id: str, + category: str | None = None, + tags: list | None = None, + types: list | None = None, + limit: int | None = None, +) -> list[dict]: + """Query decisions for a project with optional filters. + + tags: matches if ANY tag is present (OR logic via json_each). + types: filter by decision type (decision, gotcha, workaround, etc). + """ + query = "SELECT DISTINCT d.* FROM decisions d WHERE d.project_id = ?" + params: list = [project_id] + if category: + query += " AND d.category = ?" + params.append(category) + if types: + placeholders = ", ".join("?" for _ in types) + query += f" AND d.type IN ({placeholders})" + params.extend(types) + if tags: + query += """ AND d.id IN ( + SELECT d2.id FROM decisions d2, json_each(d2.tags) AS t + WHERE t.value IN ({}) + )""".format(", ".join("?" for _ in tags)) + params.extend(tags) + query += " ORDER BY d.created_at DESC" + if limit: + query += " LIMIT ?" + params.append(limit) + return _rows_to_list(conn.execute(query, params).fetchall()) + + +# --------------------------------------------------------------------------- +# Modules +# --------------------------------------------------------------------------- + +def add_module( + conn: sqlite3.Connection, + project_id: str, + name: str, + type: str, + path: str, + description: str | None = None, + owner_role: str | None = None, + dependencies: list | None = None, +) -> dict: + """Register a project module.""" + cur = conn.execute( + """INSERT INTO modules (project_id, name, type, path, description, + owner_role, dependencies) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (project_id, name, type, path, description, owner_role, + _json_encode(dependencies)), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM modules WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return _row_to_dict(row) + + +def get_modules(conn: sqlite3.Connection, project_id: str) -> list[dict]: + """Get all modules for a project.""" + rows = conn.execute( + "SELECT * FROM modules WHERE project_id = ? ORDER BY type, name", + (project_id,), + ).fetchall() + return _rows_to_list(rows) + + +# --------------------------------------------------------------------------- +# Agent Logs +# --------------------------------------------------------------------------- + +def log_agent_run( + conn: sqlite3.Connection, + project_id: str, + agent_role: str, + action: str, + task_id: str | None = None, + session_id: str | None = None, + input_summary: str | None = None, + output_summary: str | None = None, + tokens_used: int | None = None, + model: str | None = None, + cost_usd: float | None = None, + success: bool = True, + error_message: str | None = None, + duration_seconds: int | None = None, +) -> dict: + """Log an agent execution run.""" + cur = conn.execute( + """INSERT INTO agent_logs (project_id, task_id, agent_role, session_id, + action, input_summary, output_summary, tokens_used, model, + cost_usd, success, error_message, duration_seconds) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (project_id, task_id, agent_role, session_id, action, input_summary, + output_summary, tokens_used, model, cost_usd, success, + error_message, duration_seconds), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM agent_logs WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return _row_to_dict(row) + + +# --------------------------------------------------------------------------- +# Pipelines +# --------------------------------------------------------------------------- + +def create_pipeline( + conn: sqlite3.Connection, + task_id: str, + project_id: str, + route_type: str, + steps: list | dict, +) -> dict: + """Create a new pipeline run.""" + cur = conn.execute( + """INSERT INTO pipelines (task_id, project_id, route_type, steps) + VALUES (?, ?, ?, ?)""", + (task_id, project_id, route_type, _json_encode(steps)), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM pipelines WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return _row_to_dict(row) + + +def update_pipeline( + conn: sqlite3.Connection, + id: int, + status: str | None = None, + total_cost_usd: float | None = None, + total_tokens: int | None = None, + total_duration_seconds: int | None = None, +) -> dict: + """Update pipeline status and stats.""" + fields: dict[str, Any] = {} + if status is not None: + fields["status"] = status + if status in ("completed", "failed", "cancelled"): + fields["completed_at"] = datetime.now().isoformat() + if total_cost_usd is not None: + fields["total_cost_usd"] = total_cost_usd + if total_tokens is not None: + fields["total_tokens"] = total_tokens + if total_duration_seconds is not None: + fields["total_duration_seconds"] = total_duration_seconds + if fields: + sets = ", ".join(f"{k} = ?" for k in fields) + vals = list(fields.values()) + [id] + conn.execute(f"UPDATE pipelines SET {sets} WHERE id = ?", vals) + conn.commit() + row = conn.execute( + "SELECT * FROM pipelines WHERE id = ?", (id,) + ).fetchone() + return _row_to_dict(row) + + +# --------------------------------------------------------------------------- +# Support +# --------------------------------------------------------------------------- + +def create_ticket( + conn: sqlite3.Connection, + project_id: str, + source: str, + client_message: str, + client_id: str | None = None, + classification: str | None = None, +) -> dict: + """Create a support ticket.""" + cur = conn.execute( + """INSERT INTO support_tickets (project_id, source, client_id, + client_message, classification) + VALUES (?, ?, ?, ?, ?)""", + (project_id, source, client_id, client_message, classification), + ) + conn.commit() + row = conn.execute( + "SELECT * FROM support_tickets WHERE id = ?", (cur.lastrowid,) + ).fetchone() + return _row_to_dict(row) + + +def list_tickets( + conn: sqlite3.Connection, + project_id: str | None = None, + status: str | None = None, +) -> list[dict]: + """List support tickets with optional filters.""" + query = "SELECT * FROM support_tickets WHERE 1=1" + params: list = [] + if project_id: + query += " AND project_id = ?" + params.append(project_id) + if status: + query += " AND status = ?" + params.append(status) + query += " ORDER BY created_at DESC" + return _rows_to_list(conn.execute(query, params).fetchall()) + + +# --------------------------------------------------------------------------- +# Statistics / Dashboard +# --------------------------------------------------------------------------- + +def get_project_summary(conn: sqlite3.Connection) -> list[dict]: + """Get all projects with task counts by status.""" + rows = conn.execute(""" + SELECT p.*, + COUNT(t.id) AS total_tasks, + SUM(CASE WHEN t.status = 'done' THEN 1 ELSE 0 END) AS done_tasks, + SUM(CASE WHEN t.status = 'in_progress' THEN 1 ELSE 0 END) AS active_tasks, + SUM(CASE WHEN t.status = 'blocked' THEN 1 ELSE 0 END) AS blocked_tasks + FROM projects p + LEFT JOIN tasks t ON t.project_id = p.id + GROUP BY p.id + ORDER BY p.priority, p.name + """).fetchall() + return _rows_to_list(rows) + + +def get_cost_summary(conn: sqlite3.Connection, days: int = 7) -> list[dict]: + """Get cost summary by project for the last N days.""" + rows = conn.execute(""" + SELECT + p.id AS project_id, + p.name AS project_name, + COUNT(a.id) AS runs, + COALESCE(SUM(a.tokens_used), 0) AS total_tokens, + COALESCE(SUM(a.cost_usd), 0) AS total_cost_usd, + COALESCE(SUM(a.duration_seconds), 0) AS total_duration_seconds + FROM projects p + LEFT JOIN agent_logs a ON a.project_id = p.id + AND a.created_at >= datetime('now', ?) + GROUP BY p.id + HAVING runs > 0 + ORDER BY total_cost_usd DESC + """, (f"-{days} days",)).fetchall() + return _rows_to_list(rows) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..9982e39 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,240 @@ +"""Tests for core/models.py — all functions, in-memory SQLite.""" + +import pytest +from core.db import init_db +from core import models + + +@pytest.fixture +def conn(): + """Fresh in-memory DB for each test.""" + c = init_db(db_path=":memory:") + yield c + c.close() + + +# -- Projects -- + +def test_create_and_get_project(conn): + p = models.create_project(conn, "vdol", "В долю поперёк", "~/projects/vdolipoperek", + tech_stack=["vue3", "nuxt"]) + assert p["id"] == "vdol" + assert p["tech_stack"] == ["vue3", "nuxt"] + assert p["status"] == "active" + + fetched = models.get_project(conn, "vdol") + assert fetched["name"] == "В долю поперёк" + + +def test_get_project_not_found(conn): + assert models.get_project(conn, "nope") is None + + +def test_list_projects_filter(conn): + models.create_project(conn, "a", "A", "/a", status="active") + models.create_project(conn, "b", "B", "/b", status="paused") + models.create_project(conn, "c", "C", "/c", status="active") + + assert len(models.list_projects(conn)) == 3 + assert len(models.list_projects(conn, status="active")) == 2 + assert len(models.list_projects(conn, status="paused")) == 1 + + +def test_update_project(conn): + models.create_project(conn, "x", "X", "/x", priority=5) + updated = models.update_project(conn, "x", priority=1, status="maintenance") + assert updated["priority"] == 1 + assert updated["status"] == "maintenance" + + +def test_update_project_tech_stack_json(conn): + models.create_project(conn, "x", "X", "/x", tech_stack=["python"]) + updated = models.update_project(conn, "x", tech_stack=["python", "fastapi"]) + assert updated["tech_stack"] == ["python", "fastapi"] + + +# -- Tasks -- + +def test_create_and_get_task(conn): + models.create_project(conn, "p1", "P1", "/p1") + t = models.create_task(conn, "P1-001", "p1", "Fix bug", + brief={"summary": "broken login"}) + assert t["id"] == "P1-001" + assert t["brief"] == {"summary": "broken login"} + assert t["status"] == "pending" + + +def test_list_tasks_filters(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.create_project(conn, "p2", "P2", "/p2") + models.create_task(conn, "P1-001", "p1", "Task A", status="pending") + models.create_task(conn, "P1-002", "p1", "Task B", status="done") + models.create_task(conn, "P2-001", "p2", "Task C", status="pending") + + assert len(models.list_tasks(conn)) == 3 + assert len(models.list_tasks(conn, project_id="p1")) == 2 + assert len(models.list_tasks(conn, status="pending")) == 2 + assert len(models.list_tasks(conn, project_id="p1", status="done")) == 1 + + +def test_update_task(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.create_task(conn, "P1-001", "p1", "Task") + updated = models.update_task(conn, "P1-001", status="in_progress", + spec={"steps": [1, 2, 3]}) + assert updated["status"] == "in_progress" + assert updated["spec"] == {"steps": [1, 2, 3]} + assert updated["updated_at"] is not None + + +def test_subtask(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.create_task(conn, "P1-001", "p1", "Parent") + child = models.create_task(conn, "P1-001a", "p1", "Child", + parent_task_id="P1-001") + assert child["parent_task_id"] == "P1-001" + + +# -- Decisions -- + +def test_add_and_get_decisions(conn): + models.create_project(conn, "p1", "P1", "/p1") + d = models.add_decision(conn, "p1", "gotcha", "iOS Safari bottom sheet", + "position:fixed breaks on iOS Safari", + category="ui", tags=["ios-safari", "css"]) + assert d["type"] == "gotcha" + assert d["tags"] == ["ios-safari", "css"] + + results = models.get_decisions(conn, "p1") + assert len(results) == 1 + + +def test_decisions_filter_by_category(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.add_decision(conn, "p1", "decision", "Use WAL", "perf", + category="architecture") + models.add_decision(conn, "p1", "gotcha", "Safari bug", "css", + category="ui") + assert len(models.get_decisions(conn, "p1", category="ui")) == 1 + + +def test_decisions_filter_by_tags(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.add_decision(conn, "p1", "gotcha", "Bug A", "desc", + tags=["safari", "css"]) + models.add_decision(conn, "p1", "gotcha", "Bug B", "desc", + tags=["chrome", "js"]) + models.add_decision(conn, "p1", "gotcha", "Bug C", "desc", + tags=["safari", "js"]) + + assert len(models.get_decisions(conn, "p1", tags=["safari"])) == 2 + assert len(models.get_decisions(conn, "p1", tags=["js"])) == 2 + assert len(models.get_decisions(conn, "p1", tags=["css"])) == 1 + + +def test_decisions_filter_by_types(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.add_decision(conn, "p1", "decision", "A", "a") + models.add_decision(conn, "p1", "gotcha", "B", "b") + models.add_decision(conn, "p1", "workaround", "C", "c") + + assert len(models.get_decisions(conn, "p1", types=["gotcha", "workaround"])) == 2 + + +def test_decisions_limit(conn): + models.create_project(conn, "p1", "P1", "/p1") + for i in range(10): + models.add_decision(conn, "p1", "decision", f"D{i}", f"desc{i}") + assert len(models.get_decisions(conn, "p1", limit=3)) == 3 + + +# -- Modules -- + +def test_add_and_get_modules(conn): + models.create_project(conn, "p1", "P1", "/p1") + m = models.add_module(conn, "p1", "search", "frontend", "src/search/", + description="Search UI", dependencies=["auth"]) + assert m["name"] == "search" + assert m["dependencies"] == ["auth"] + + mods = models.get_modules(conn, "p1") + assert len(mods) == 1 + + +# -- Agent Logs -- + +def test_log_agent_run(conn): + models.create_project(conn, "p1", "P1", "/p1") + log = models.log_agent_run(conn, "p1", "developer", "implement", + tokens_used=5000, model="sonnet", + cost_usd=0.015, duration_seconds=45) + assert log["agent_role"] == "developer" + assert log["cost_usd"] == 0.015 + assert log["success"] == 1 # SQLite boolean + + +# -- Pipelines -- + +def test_create_and_update_pipeline(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.create_task(conn, "P1-001", "p1", "Task") + pipe = models.create_pipeline(conn, "P1-001", "p1", "feature", + [{"step": "architect"}, {"step": "dev"}]) + assert pipe["status"] == "running" + assert pipe["steps"] == [{"step": "architect"}, {"step": "dev"}] + + updated = models.update_pipeline(conn, pipe["id"], status="completed", + total_cost_usd=0.05, total_tokens=10000) + assert updated["status"] == "completed" + assert updated["completed_at"] is not None + + +# -- Support -- + +def test_create_and_list_tickets(conn): + models.create_project(conn, "p1", "P1", "/p1") + t = models.create_ticket(conn, "p1", "telegram_bot", "Не работает поиск", + client_id="tg:12345", classification="bug") + assert t["source"] == "telegram_bot" + assert t["status"] == "new" + + tickets = models.list_tickets(conn, project_id="p1") + assert len(tickets) == 1 + + assert len(models.list_tickets(conn, status="resolved")) == 0 + + +# -- Statistics -- + +def test_project_summary(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.create_task(conn, "P1-001", "p1", "A", status="done") + models.create_task(conn, "P1-002", "p1", "B", status="in_progress") + models.create_task(conn, "P1-003", "p1", "C", status="blocked") + + summary = models.get_project_summary(conn) + assert len(summary) == 1 + s = summary[0] + assert s["total_tasks"] == 3 + assert s["done_tasks"] == 1 + assert s["active_tasks"] == 1 + assert s["blocked_tasks"] == 1 + + +def test_cost_summary(conn): + models.create_project(conn, "p1", "P1", "/p1") + models.log_agent_run(conn, "p1", "dev", "implement", + cost_usd=0.10, tokens_used=5000) + models.log_agent_run(conn, "p1", "reviewer", "review", + cost_usd=0.05, tokens_used=2000) + + costs = models.get_cost_summary(conn, days=1) + assert len(costs) == 1 + assert costs[0]["total_cost_usd"] == pytest.approx(0.15) + assert costs[0]["total_tokens"] == 7000 + assert costs[0]["runs"] == 2 + + +def test_cost_summary_empty(conn): + models.create_project(conn, "p1", "P1", "/p1") + assert models.get_cost_summary(conn, days=7) == []