Coverage for src/case_router/service.py: 100%

1"""Use cases: submit, read, list, and close cases, plus the queue health report.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from datetime import UTC, datetime 

7 

8from .db import CaseRepository, Database, RuleRepository, TenantRepository 

9from .deadlines import ResponsePolicy 

10from .errors import UnknownTenant 

11from .metrics import Metrics 

12from .models import Case, CaseDraft, QueueHealth, Status, Tenant 

13from .rules import RuleSet, build_rule 

14 

15UtcClock = Callable[[], datetime] 

16 

17 

18def utc_now() -> datetime: 

19 return datetime.now(UTC) 

20 

21 

22class CaseService: 

23 def __init__( 

24 self, 

25 db: Database, 

26 *, 

27 metrics: Metrics, 

28 policy: ResponsePolicy | None = None, 

29 clock: UtcClock = utc_now, 

30 ) -> None: 

31 self.tenants = TenantRepository(db) 

32 self.rules = RuleRepository(db) 

33 self.cases = CaseRepository(db) 

34 self.metrics = metrics 

35 self.policy = policy or ResponsePolicy() 

36 self._clock = clock 

37 

38 def tenant(self, tenant_id: str) -> Tenant: 

39 tenant = self.tenants.get(tenant_id) 

40 if tenant is None: 

41 raise UnknownTenant(tenant_id) 

42 return tenant 

43 

44 def rule_set(self, tenant_id: str) -> RuleSet: 

45 rows = self.rules.for_tenant(tenant_id) 

46 return RuleSet([build_rule(r.kind, r.value, r.queue) for r in rows]) 

47 

48 def submit(self, tenant: Tenant, draft: CaseDraft) -> Case: 

49 with self.metrics.routing_seconds.labels(tenant.id).time(): 

50 decision = self.rule_set(tenant.id).route(draft) 

51 created_at = self._clock() 

52 respond_by = self.policy.respond_by(created_at, draft.priority, draft.tier) 

53 case = self.cases.insert(tenant.id, draft, decision, created_at, respond_by) 

54 self.metrics.cases_routed.labels(tenant.id, decision.queue, decision.rule).inc() 

55 return case 

56 

57 def get(self, tenant: Tenant, case_id: int) -> Case: 

58 return self.cases.get(tenant.id, case_id) 

59 

60 def find( 

61 self, tenant: Tenant, *, queue: str | None, status: Status | None, limit: int 

62 ) -> list[Case]: 

63 return self.cases.find(tenant.id, queue=queue, status=status, limit=limit) 

64 

65 def close(self, tenant: Tenant, case_id: int) -> Case: 

66 case = self.cases.close(tenant.id, case_id, self._clock()) 

67 assert case.closed_at is not None 

68 on_time = "yes" if case.closed_at <= case.respond_by else "no" 

69 self.metrics.cases_closed.labels(tenant.id, case.queue, on_time).inc() 

70 return case 

71 

72 def queue_health(self, tenant: Tenant) -> list[QueueHealth]: 

73 return self.cases.queue_health(tenant.id, self._clock())