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

1"""HTTP API. Every /v1 request names its tenant in the X-Tenant-Id header.""" 

2 

3from __future__ import annotations 

4 

5import math 

6import os 

7import time 

8from collections.abc import Awaitable, Callable 

9from dataclasses import dataclass 

10from datetime import datetime 

11from typing import Annotated, Literal, cast 

12 

13from fastapi import ( 

14 APIRouter, 

15 Body, 

16 Depends, 

17 FastAPI, 

18 Header, 

19 HTTPException, 

20 Query, 

21 Request, 

22 Response, 

23) 

24from fastapi.responses import JSONResponse 

25from prometheus_client import CONTENT_TYPE_LATEST, generate_latest 

26from pydantic import BaseModel, ConfigDict, Field 

27 

28from . import __version__ 

29from .db import Database, RuleRow 

30from .errors import CaseAlreadyClosed, CaseNotFound, UnknownTenant 

31from .limits import Clock, TenantLimiter 

32from .metrics import Metrics 

33from .models import Case, CaseDraft, Channel, Priority, QueueHealth, Status, Tenant, Tier 

34from .seed import seed_demo_data 

35from .service import CaseService, UtcClock, utc_now 

36 

37 

38@dataclass(frozen=True) 

39class Settings: 

40 database_path: str = "case_router.db" 

41 seed_demo_data: bool = True 

42 

43 @classmethod 

44 def from_env(cls) -> Settings: 

45 return cls( 

46 database_path=os.environ.get("CASE_ROUTER_DB", cls.database_path), 

47 seed_demo_data=os.environ.get("CASE_ROUTER_SEED", "1") != "0", 

48 ) 

49 

50 

51# Control characters have no place in a case and can forge extra lines in logs. The 

52# description may still hold tabs and line breaks. 

53NO_CONTROL_CHARS = r"^[^\x00-\x1f\x7f]*$" 

54NO_CONTROL_CHARS_EXCEPT_LINES = r"^[^\x00-\x08\x0b\x0c\x0e-\x1f\x7f]*$" 

55 

56 

57class CaseIn(BaseModel): 

58 model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) 

59 

60 subject: str = Field(min_length=3, max_length=200, pattern=NO_CONTROL_CHARS) 

61 description: str = Field(default="", max_length=5000, pattern=NO_CONTROL_CHARS_EXCEPT_LINES) 

62 priority: Priority = Priority.MEDIUM 

63 channel: Channel = Channel.WEB 

64 tier: Tier = Tier.STANDARD 

65 

66 def to_draft(self) -> CaseDraft: 

67 return CaseDraft(self.subject, self.description, self.priority, self.channel, self.tier) 

68 

69 

70class CaseOut(BaseModel): 

71 id: int 

72 subject: str 

73 description: str 

74 priority: Priority 

75 channel: Channel 

76 tier: Tier 

77 queue: str 

78 rule: str 

79 status: Status 

80 created_at: datetime 

81 respond_by: datetime 

82 closed_at: datetime | None 

83 

84 @classmethod 

85 def of(cls, case: Case) -> CaseOut: 

86 return cls( 

87 id=case.id, 

88 subject=case.subject, 

89 description=case.description, 

90 priority=case.priority, 

91 channel=case.channel, 

92 tier=case.tier, 

93 queue=case.queue, 

94 rule=case.rule, 

95 status=case.status, 

96 created_at=case.created_at, 

97 respond_by=case.respond_by, 

98 closed_at=case.closed_at, 

99 ) 

100 

101 

102class QueueHealthOut(BaseModel): 

103 queue: str 

104 open_cases: int 

105 overdue_cases: int 

106 oldest_open_at: datetime | None 

107 

108 @classmethod 

109 def of(cls, health: QueueHealth) -> QueueHealthOut: 

110 return cls( 

111 queue=health.queue, 

112 open_cases=health.open_cases, 

113 overdue_cases=health.overdue_cases, 

114 oldest_open_at=health.oldest_open_at, 

115 ) 

116 

117 

118class RuleIn(BaseModel): 

119 model_config = ConfigDict(extra="forbid") 

120 

121 kind: Literal["priority", "tier", "channel", "keyword", "any"] 

122 value: str = Field(default="", max_length=500) 

123 queue: str = Field(min_length=1, max_length=100) 

124 

125 

126class RuleOut(BaseModel): 

127 position: int 

128 name: str 

129 queue: str 

130 

131 

132@dataclass(frozen=True) 

133class AppState: 

134 db: Database 

135 service: CaseService 

136 limiter: TenantLimiter 

137 metrics: Metrics 

138 

139 

140def app_state(request: Request) -> AppState: 

141 return cast(AppState, request.app.state.case_router) 

142 

143 

144StateDep = Annotated[AppState, Depends(app_state)] 

145 

146 

147def current_tenant( 

148 state: StateDep, x_tenant_id: Annotated[str, Header(min_length=1, max_length=64)] 

149) -> Tenant: 

150 """Look up the tenant and spend one token from its request allowance.""" 

151 tenant = state.service.tenant(x_tenant_id) 

152 acquisition = state.limiter.acquire(tenant) 

153 if not acquisition.allowed: 

154 state.metrics.requests_rejected.labels(tenant.id, "rate_limit").inc() 

155 raise HTTPException( 

156 status_code=429, 

157 detail="Request limit reached for this tenant", 

158 headers={"Retry-After": str(max(1, math.ceil(acquisition.retry_after)))}, 

159 ) 

160 return tenant 

161 

162 

163TenantDep = Annotated[Tenant, Depends(current_tenant)] 

164 

165ops = APIRouter(tags=["ops"]) 

166v1 = APIRouter(prefix="/v1") 

167 

168 

169@ops.get("/healthz") 

170def health(state: StateDep) -> dict[str, str]: 

171 with state.db.connect() as conn: 

172 conn.execute("SELECT 1") 

173 return {"status": "ok"} 

174 

175 

176@ops.get("/metrics") 

177def prometheus_metrics(state: StateDep) -> Response: 

178 return Response(generate_latest(state.metrics.registry), media_type=CONTENT_TYPE_LATEST) 

179 

180 

181@v1.post("/cases", status_code=201, tags=["cases"]) 

182def submit_case(body: CaseIn, tenant: TenantDep, state: StateDep) -> CaseOut: 

183 return CaseOut.of(state.service.submit(tenant, body.to_draft())) 

184 

185 

186@v1.get("/cases", tags=["cases"]) 

187def list_cases( 

188 tenant: TenantDep, 

189 state: StateDep, 

190 queue: str | None = None, 

191 status: Status | None = None, 

192 limit: Annotated[int, Query(ge=1, le=200)] = 50, 

193) -> list[CaseOut]: 

194 cases = state.service.find(tenant, queue=queue, status=status, limit=limit) 

195 return [CaseOut.of(c) for c in cases] 

196 

197 

198@v1.get("/cases/{case_id}", tags=["cases"]) 

199def get_case(case_id: int, tenant: TenantDep, state: StateDep) -> CaseOut: 

200 return CaseOut.of(state.service.get(tenant, case_id)) 

201 

202 

203@v1.post("/cases/{case_id}/close", tags=["cases"]) 

204def close_case(case_id: int, tenant: TenantDep, state: StateDep) -> CaseOut: 

205 return CaseOut.of(state.service.close(tenant, case_id)) 

206 

207 

208@v1.get("/queues", tags=["queues"]) 

209def queue_health(tenant: TenantDep, state: StateDep) -> list[QueueHealthOut]: 

210 return [QueueHealthOut.of(q) for q in state.service.queue_health(tenant)] 

211 

212 

213def _rules_out(state: AppState, tenant: Tenant) -> list[RuleOut]: 

214 rules = state.service.rule_set(tenant.id) 

215 return [RuleOut(position=i, name=r.name, queue=r.queue) for i, r in enumerate(rules, 1)] 

216 

217 

218@v1.get("/rules", tags=["rules"]) 

219def get_rules(tenant: TenantDep, state: StateDep) -> list[RuleOut]: 

220 return _rules_out(state, tenant) 

221 

222 

223@v1.put("/rules", tags=["rules"]) 

224def replace_rules( 

225 body: Annotated[list[RuleIn], Body(min_length=1, max_length=50)], 

226 tenant: TenantDep, 

227 state: StateDep, 

228) -> list[RuleOut]: 

229 try: 

230 state.service.rules.replace(tenant.id, [RuleRow(r.kind, r.value, r.queue) for r in body]) 

231 except ValueError as exc: 

232 raise HTTPException(status_code=422, detail=str(exc)) from exc 

233 return _rules_out(state, tenant) 

234 

235 

236async def _unknown_tenant(_request: Request, _exc: Exception) -> JSONResponse: 

237 return JSONResponse({"detail": "Unknown tenant"}, status_code=404) 

238 

239 

240async def _case_not_found(_request: Request, _exc: Exception) -> JSONResponse: 

241 return JSONResponse({"detail": "Case not found"}, status_code=404) 

242 

243 

244async def _already_closed(_request: Request, _exc: Exception) -> JSONResponse: 

245 return JSONResponse({"detail": "Case is already closed"}, status_code=409) 

246 

247 

248def create_app( 

249 settings: Settings | None = None, 

250 *, 

251 clock: UtcClock = utc_now, 

252 monotonic: Clock = time.monotonic, 

253) -> FastAPI: 

254 """Build an app with its own database, limiter, and metrics registry. 

255 

256 Clocks are injected so tests can move time forward instead of sleeping. 

257 """ 

258 settings = settings or Settings.from_env() 

259 db = Database(settings.database_path) 

260 if settings.seed_demo_data: 

261 seed_demo_data(db) 

262 metrics = Metrics() 

263 state = AppState( 

264 db=db, 

265 service=CaseService(db, metrics=metrics, clock=clock), 

266 limiter=TenantLimiter(clock=monotonic), 

267 metrics=metrics, 

268 ) 

269 

270 app = FastAPI( 

271 title="Case Router", 

272 version=__version__, 

273 description="Routes support cases to queues per tenant, with request limits and metrics.", 

274 ) 

275 app.state.case_router = state 

276 app.include_router(ops) 

277 app.include_router(v1) 

278 app.add_exception_handler(UnknownTenant, _unknown_tenant) 

279 app.add_exception_handler(CaseNotFound, _case_not_found) 

280 app.add_exception_handler(CaseAlreadyClosed, _already_closed) 

281 

282 @app.middleware("http") 

283 async def record_latency( 

284 request: Request, call_next: Callable[[Request], Awaitable[Response]] 

285 ) -> Response: 

286 started = time.perf_counter() 

287 response = await call_next(request) 

288 route = getattr(request.scope.get("route"), "path", "unmatched") 

289 metrics.request_seconds.labels(request.method, route, str(response.status_code)).observe( 

290 time.perf_counter() - started 

291 ) 

292 return response 

293 

294 return app