Coverage for src/case_router/db.py: 100%
109 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-12 18:59 +0000
1"""SQLite storage with hand-written SQL. Every case query is scoped by tenant_id."""
3from __future__ import annotations
5import sqlite3
6from collections.abc import Iterator, Sequence
7from contextlib import contextmanager
8from dataclasses import dataclass
9from datetime import UTC, datetime
10from importlib.resources import files
11from pathlib import Path
13from .errors import CaseAlreadyClosed, CaseNotFound
14from .models import (
15 Case,
16 CaseDraft,
17 Channel,
18 Priority,
19 QueueHealth,
20 RoutingDecision,
21 Status,
22 Tenant,
23 Tier,
24)
25from .rules import build_rule
27SCHEMA = files("case_router").joinpath("schema.sql").read_text(encoding="utf-8")
28_TS_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
31def to_db_time(value: datetime) -> str:
32 """Fixed-width UTC text, so SQL can compare timestamps as strings."""
33 if value.tzinfo is None:
34 raise ValueError("timestamps must include a time zone")
35 return value.astimezone(UTC).strftime(_TS_FORMAT)
38def from_db_time(value: str) -> datetime:
39 return datetime.strptime(value, _TS_FORMAT).replace(tzinfo=UTC)
42class Database:
43 def __init__(self, path: str | Path) -> None:
44 if str(path) == ":memory:":
45 raise ValueError("use a file path; each connection to :memory: is a separate database")
46 self.path = str(path)
47 with self.connect() as conn:
48 conn.execute("PRAGMA journal_mode = WAL")
49 conn.executescript(SCHEMA)
51 @contextmanager
52 def connect(self) -> Iterator[sqlite3.Connection]:
53 """One connection per unit of work: commit on success, roll back on error."""
54 conn = sqlite3.connect(self.path, timeout=10)
55 conn.row_factory = sqlite3.Row
56 conn.execute("PRAGMA foreign_keys = ON")
57 try:
58 with conn:
59 yield conn
60 finally:
61 conn.close()
64class TenantRepository:
65 def __init__(self, db: Database) -> None:
66 self._db = db
68 def get(self, tenant_id: str) -> Tenant | None:
69 with self._db.connect() as conn:
70 row = conn.execute(
71 "SELECT id, name, requests_per_minute FROM tenants WHERE id = ?", (tenant_id,)
72 ).fetchone()
73 return None if row is None else Tenant(row["id"], row["name"], row["requests_per_minute"])
75 def save(self, tenant: Tenant) -> None:
76 with self._db.connect() as conn:
77 conn.execute(
78 """
79 INSERT INTO tenants (id, name, requests_per_minute) VALUES (?, ?, ?)
80 ON CONFLICT (id) DO UPDATE SET
81 name = excluded.name,
82 requests_per_minute = excluded.requests_per_minute
83 """,
84 (tenant.id, tenant.name, tenant.requests_per_minute),
85 )
88@dataclass(frozen=True, slots=True)
89class RuleRow:
90 kind: str
91 value: str
92 queue: str
95class RuleRepository:
96 def __init__(self, db: Database) -> None:
97 self._db = db
99 def for_tenant(self, tenant_id: str) -> list[RuleRow]:
100 with self._db.connect() as conn:
101 rows = conn.execute(
102 "SELECT kind, value, queue FROM routing_rules "
103 "WHERE tenant_id = ? ORDER BY position",
104 (tenant_id,),
105 ).fetchall()
106 return [RuleRow(r["kind"], r["value"], r["queue"]) for r in rows]
108 def replace(self, tenant_id: str, rules: Sequence[RuleRow]) -> None:
109 """Swap a tenant's whole rule list in one transaction.
111 Each row is built into a rule object first, so a bad row is rejected before any
112 existing rule is deleted.
113 """
114 for row in rules:
115 build_rule(row.kind, row.value, row.queue)
116 with self._db.connect() as conn:
117 conn.execute("DELETE FROM routing_rules WHERE tenant_id = ?", (tenant_id,))
118 conn.executemany(
119 "INSERT INTO routing_rules (tenant_id, position, kind, value, queue) "
120 "VALUES (?, ?, ?, ?, ?)",
121 [(tenant_id, i, r.kind, r.value, r.queue) for i, r in enumerate(rules, start=1)],
122 )
125_CASE_COLUMNS = (
126 "id, tenant_id, subject, description, priority, channel, tier, queue, rule, status, "
127 "created_at, respond_by, closed_at"
128)
131def _case_from_row(row: sqlite3.Row) -> Case:
132 return Case(
133 id=row["id"],
134 tenant_id=row["tenant_id"],
135 subject=row["subject"],
136 description=row["description"],
137 priority=Priority(row["priority"]),
138 channel=Channel(row["channel"]),
139 tier=Tier(row["tier"]),
140 queue=row["queue"],
141 rule=row["rule"],
142 status=Status(row["status"]),
143 created_at=from_db_time(row["created_at"]),
144 respond_by=from_db_time(row["respond_by"]),
145 closed_at=None if row["closed_at"] is None else from_db_time(row["closed_at"]),
146 )
149class CaseRepository:
150 def __init__(self, db: Database) -> None:
151 self._db = db
153 def insert(
154 self,
155 tenant_id: str,
156 draft: CaseDraft,
157 decision: RoutingDecision,
158 created_at: datetime,
159 respond_by: datetime,
160 ) -> Case:
161 with self._db.connect() as conn:
162 cursor = conn.execute(
163 """
164 INSERT INTO cases (tenant_id, subject, description, priority, channel, tier,
165 queue, rule, status, created_at, respond_by)
166 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'ROUTED', ?, ?)
167 """,
168 (
169 tenant_id,
170 draft.subject,
171 draft.description,
172 draft.priority,
173 draft.channel,
174 draft.tier,
175 decision.queue,
176 decision.rule,
177 to_db_time(created_at),
178 to_db_time(respond_by),
179 ),
180 )
181 case_id = cursor.lastrowid
182 assert case_id is not None
183 return self._fetch(conn, tenant_id, case_id)
185 def get(self, tenant_id: str, case_id: int) -> Case:
186 with self._db.connect() as conn:
187 return self._fetch(conn, tenant_id, case_id)
189 def find(
190 self,
191 tenant_id: str,
192 *,
193 queue: str | None = None,
194 status: Status | None = None,
195 limit: int = 50,
196 ) -> list[Case]:
197 """Cases for one tenant, soonest deadline first."""
198 sql = f"SELECT {_CASE_COLUMNS} FROM cases WHERE tenant_id = ?"
199 params: list[object] = [tenant_id]
200 if queue is not None:
201 sql += " AND queue = ?"
202 params.append(queue)
203 if status is not None:
204 sql += " AND status = ?"
205 params.append(status)
206 sql += " ORDER BY respond_by, id LIMIT ?"
207 params.append(limit)
208 with self._db.connect() as conn:
209 return [_case_from_row(r) for r in conn.execute(sql, params).fetchall()]
211 def close(self, tenant_id: str, case_id: int, closed_at: datetime) -> Case:
212 """Close a case. The status check sits inside the UPDATE, so two clients racing to
213 close the same case cannot both succeed."""
214 with self._db.connect() as conn:
215 cursor = conn.execute(
216 "UPDATE cases SET status = 'CLOSED', closed_at = ? "
217 "WHERE id = ? AND tenant_id = ? AND status = 'ROUTED'",
218 (to_db_time(closed_at), case_id, tenant_id),
219 )
220 case = self._fetch(conn, tenant_id, case_id)
221 if cursor.rowcount == 0:
222 raise CaseAlreadyClosed(case_id)
223 return case
225 def queue_health(self, tenant_id: str, now: datetime) -> list[QueueHealth]:
226 """Open and overdue counts per queue, busiest queue first."""
227 with self._db.connect() as conn:
228 rows = conn.execute(
229 """
230 SELECT queue,
231 SUM(status = 'ROUTED') AS open_cases,
232 SUM(status = 'ROUTED' AND respond_by < ?) AS overdue_cases,
233 MIN(CASE WHEN status = 'ROUTED' THEN created_at END) AS oldest_open_at
234 FROM cases
235 WHERE tenant_id = ?
236 GROUP BY queue
237 ORDER BY open_cases DESC, queue
238 """,
239 (to_db_time(now), tenant_id),
240 ).fetchall()
241 return [
242 QueueHealth(
243 queue=r["queue"],
244 open_cases=r["open_cases"],
245 overdue_cases=r["overdue_cases"],
246 oldest_open_at=None
247 if r["oldest_open_at"] is None
248 else from_db_time(r["oldest_open_at"]),
249 )
250 for r in rows
251 ]
253 @staticmethod
254 def _fetch(conn: sqlite3.Connection, tenant_id: str, case_id: int) -> Case:
255 row = conn.execute(
256 f"SELECT {_CASE_COLUMNS} FROM cases WHERE id = ? AND tenant_id = ?",
257 (case_id, tenant_id),
258 ).fetchone()
259 if row is None:
260 raise CaseNotFound(case_id)
261 return _case_from_row(row)