migration: establish exact preserved app baseline
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
from .context import correlation_id
|
||||
from .errors import normalize_error_payload
|
||||
from .router import AccessRole, ApiRoute, ApiRouteRegistry, RouteRegistryError
|
||||
|
||||
__all__ = [
|
||||
"AccessRole", "ApiRoute", "ApiRouteRegistry", "RouteRegistryError",
|
||||
"correlation_id", "normalize_error_payload",
|
||||
]
|
||||
@@ -0,0 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import uuid
|
||||
|
||||
|
||||
REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9._-]{8,80}")
|
||||
|
||||
|
||||
def correlation_id(supplied: str = "") -> str:
|
||||
value = str(supplied or "").strip()
|
||||
return value if REQUEST_ID_PATTERN.fullmatch(value) else uuid.uuid4().hex
|
||||
@@ -0,0 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
from typing import Any
|
||||
|
||||
|
||||
STATUS_CODES = {
|
||||
HTTPStatus.BAD_REQUEST: "bad_request",
|
||||
HTTPStatus.UNAUTHORIZED: "authentication_required",
|
||||
HTTPStatus.FORBIDDEN: "access_denied",
|
||||
HTTPStatus.NOT_FOUND: "not_found",
|
||||
HTTPStatus.CONFLICT: "conflict",
|
||||
HTTPStatus.INTERNAL_SERVER_ERROR: "internal_error",
|
||||
HTTPStatus.SERVICE_UNAVAILABLE: "service_unavailable",
|
||||
}
|
||||
|
||||
|
||||
def normalize_error_payload(
|
||||
payload: dict[str, Any], status: int | HTTPStatus, request_id: str,
|
||||
) -> dict[str, Any]:
|
||||
if "error" not in payload:
|
||||
return payload
|
||||
status_value = HTTPStatus(int(status))
|
||||
message = str(payload.get("message") or payload.get("error") or status_value.phrase)
|
||||
return {
|
||||
**payload,
|
||||
"error": message,
|
||||
"code": str(payload.get("code") or STATUS_CODES.get(status_value) or "request_failed"),
|
||||
"message": message,
|
||||
"request_id": request_id,
|
||||
}
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user