refactor: make API registry authoritative

2026-07-29 18:07:34 +08:00
parent b0cbd25f64
commit 93fc4505a9
9 changed files with 230 additions and 67 deletions
@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal, cast
+
+from app_config import APP_DIR
+
+
+AccessRole = Literal["public", "authenticated", "member", "admin"]
+MatchType = Literal["exact", "regex"]
+
+
+class RouteRegistryError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class ApiRoute:
+    method: str
+    path: str
+    match: MatchType
+    feature: str
+    access: AccessRole
+
+    def matches(self, method: str, path: str) -> bool:
+        if self.method != method.upper():
+            return False
+        return self.path == path if self.match == "exact" else re.fullmatch(self.path, path) is not None
+
+
+class ApiRouteRegistry:
+    def __init__(self, routes: tuple[ApiRoute, ...]) -> None:
+        self.routes = routes
+        self._exact: dict[tuple[str, str], ApiRoute] = {}
+        regex_routes: list[ApiRoute] = []
+        seen: set[tuple[str, str]] = set()
+        for route in routes:
+            key = (route.method, route.path)
+            if key in seen:
+                raise RouteRegistryError(f"Duplicate API route: {route.method} {route.path}")
+            seen.add(key)
+            if route.match == "exact":
+                self._exact[key] = route
+            else:
+                try:
+                    re.compile(route.path)
+                except re.error as exc:
+                    raise RouteRegistryError(f"Invalid API route regex: {route.path}") from exc
+                regex_routes.append(route)
+        self._regex = tuple(regex_routes)
+
+    @classmethod
+    def load(cls, path: Path | None = None) -> "ApiRouteRegistry":
+        config_path = path or APP_DIR / "config" / "api.config.json"
+        payload = json.loads(config_path.read_text(encoding="utf-8"))
+        routes = tuple(
+            ApiRoute(
+                method=str(item["method"]).upper(),
+                path=str(item["path"]),
+                match=cast(MatchType, str(item["match"])),
+                feature=str(item["feature"]),
+                access=cast(AccessRole, str(item["access"])),
+            )
+            for item in payload.get("routes") or []
+        )
+        if not routes:
+            raise RouteRegistryError("API route registry is empty")
+        return cls(routes)
+
+    def resolve(self, method: str, path: str) -> ApiRoute | None:
+        normalized = method.upper()
+        exact = self._exact.get((normalized, path))
+        if exact:
+            return exact
+        return next((route for route in self._regex if route.matches(normalized, path)), None)