Edge manifest: static JSON for CDN cache invalidation routing

generate_edge_manifest() compiles the decorator registry into a
static JSON artifact that Edge reads at deploy time:

{
  "contexts": {
    "user": {
      "functions": [
        {"name": "user_profile", "path": "rpc"},
        {"name": "profile_page", "path": "view"}
      ],
      "endpoints": ["/api/mizan/ctx/user/"],
      "params": ["user_id"],
      "views": ["/profile/:user_id/"]
    }
  }
}

When Edge receives X-Mizan-Invalidate: user;user_id=5, it:
1. Looks up "user" in the manifest
2. Resolves URL patterns: /profile/:user_id/ → /profile/5/
3. Purges /profile/5/ and /api/mizan/ctx/user/?user_id=5

Features:
- Distinguishes view-path vs RPC-path functions
- Accepts optional view_urls mapping from developer
- Custom base URL support
- Deterministic JSON output (sorted keys)
- Management command: python manage.py export_edge_manifest

314 Django tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-03 13:10:41 -04:00
parent b4c7e783bd
commit 28e517e6ee
3 changed files with 301 additions and 1 deletions

View File

@@ -29,7 +29,13 @@ if TYPE_CHECKING:
from mizan.setup.registry import get_registry, get_schema, get_context_groups, get_function
__all__ = ["get_schema", "generate_openapi_schema", "generate_openapi_json"]
__all__ = [
"get_schema",
"generate_openapi_schema",
"generate_openapi_json",
"generate_edge_manifest",
"generate_edge_manifest_json",
]
def _extract_form_fields(form_class: type) -> list[dict[str, Any]]:
@@ -350,3 +356,87 @@ def generate_openapi_json(indent: int = 2) -> str:
"""Generate OpenAPI schema as formatted JSON string."""
schema = generate_openapi_schema()
return json.dumps(schema, indent=indent)
def generate_edge_manifest(
base_url: str = "/api/mizan",
view_urls: dict[str, list[str]] | None = None,
) -> dict[str, Any]:
"""
Generate the Edge manifest — a static JSON mapping contexts to URL
patterns and params for CDN cache purging.
The manifest is consumed by Mizan Edge at deploy time. When Edge
receives X-Mizan-Invalidate: user;user_id=5, it:
1. Looks up 'user' in the manifest
2. Resolves URL patterns with params: /profile/:user_id/ → /profile/5/
3. Purges the resolved URLs + the context API endpoint
Args:
base_url: The Mizan API mount point (default: /api/mizan)
view_urls: Optional mapping of context names to URL patterns for
view-path functions. These are URLs that Edge should
also purge when a context is invalidated.
Example: {"user": ["/profile/:user_id/"]}
Returns:
Manifest dict suitable for JSON serialization.
"""
from pydantic import BaseModel as PydanticBaseModel
groups = get_context_groups()
registry = get_registry()
all_functions = registry.get("functions", {})
manifest: dict[str, Any] = {"contexts": {}}
for ctx_name, fn_names in groups.items():
# Collect params from all functions in this context
param_names: set[str] = set()
functions_meta: list[dict[str, Any]] = []
for fn_name in fn_names:
fn_cls = all_functions.get(fn_name)
if fn_cls is None:
continue
meta = getattr(fn_cls, "_meta", {})
is_view = meta.get("view_path", False)
# Collect param names from Input schema
input_cls = getattr(fn_cls, "Input", None)
if (
input_cls
and input_cls is not PydanticBaseModel
and hasattr(input_cls, "model_fields")
):
param_names.update(input_cls.model_fields.keys())
functions_meta.append({
"name": fn_name,
"path": "view" if is_view else "rpc",
})
ctx_entry: dict[str, Any] = {
"functions": functions_meta,
"endpoints": [f"{base_url}/ctx/{ctx_name}/"],
"params": sorted(param_names),
}
# Add view URLs if provided
if view_urls and ctx_name in view_urls:
ctx_entry["views"] = view_urls[ctx_name]
manifest["contexts"][ctx_name] = ctx_entry
return manifest
def generate_edge_manifest_json(
indent: int = 2,
base_url: str = "/api/mizan",
view_urls: dict[str, list[str]] | None = None,
) -> str:
"""Generate Edge manifest as formatted JSON string."""
manifest = generate_edge_manifest(base_url=base_url, view_urls=view_urls)
return json.dumps(manifest, indent=indent, sort_keys=True)

View File

@@ -0,0 +1,56 @@
"""
Export Edge Manifest
Generates the static JSON manifest that Mizan Edge reads at deploy time
to configure CDN cache rules and invalidation routing.
Usage:
python manage.py export_edge_manifest
python manage.py export_edge_manifest --output mizan-manifest.json
python manage.py export_edge_manifest --base-url /api/mizan
"""
import json
from pathlib import Path
from django.core.management.base import BaseCommand
from mizan.export import generate_edge_manifest
class Command(BaseCommand):
help = "Export Edge manifest for CDN cache invalidation"
def add_arguments(self, parser):
parser.add_argument(
"--output",
"-o",
type=str,
default=None,
help="Output file path. If not specified, outputs to stdout.",
)
parser.add_argument(
"--indent",
type=int,
default=2,
help="JSON indentation level (0 for compact output)",
)
parser.add_argument(
"--base-url",
type=str,
default="/api/mizan",
help="Mizan API mount point (default: /api/mizan)",
)
def handle(self, *args, **options):
manifest = generate_edge_manifest(base_url=options["base_url"])
indent = options["indent"] if options["indent"] > 0 else None
json_output = json.dumps(manifest, indent=indent, sort_keys=True)
if options["output"]:
output_path = Path(options["output"])
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json_output)
self.stdout.write(self.style.SUCCESS(f"Manifest written to {output_path}"))
else:
self.stdout.write(json_output)

View File

@@ -2450,3 +2450,157 @@ class EdgeCompatibilityTests(TestCase):
data = response.json()
self.assertNotIn("invalidate", data)
self.assertNotIn("X-Mizan-Invalidate", response)
# =============================================================================
# Edge Manifest Tests
# =============================================================================
class EdgeManifestTests(TestCase):
"""Tests for the Edge manifest generator."""
def setUp(self):
clear_registry()
def tearDown(self):
clear_registry()
def test_basic_manifest(self):
"""Manifest lists contexts with their functions, endpoints, and params."""
from mizan.export import generate_edge_manifest
UserCtx = ReactContext("user")
@client(context=UserCtx)
def user_profile(request: HttpRequest, user_id: int) -> ValidOutput:
return ValidOutput(valid=True)
@client(context=UserCtx)
def user_orders(request: HttpRequest, user_id: int, page: int = 1) -> ValidOutput:
return ValidOutput(valid=True)
register(user_profile, "user_profile")
register(user_orders, "user_orders")
manifest = generate_edge_manifest()
self.assertIn("contexts", manifest)
self.assertIn("user", manifest["contexts"])
user_ctx = manifest["contexts"]["user"]
self.assertEqual(user_ctx["endpoints"], ["/api/mizan/ctx/user/"])
self.assertIn("user_id", user_ctx["params"])
self.assertIn("page", user_ctx["params"])
self.assertEqual(len(user_ctx["functions"]), 2)
def test_manifest_distinguishes_view_and_rpc(self):
"""Manifest marks functions as view or rpc path."""
from django.http import HttpResponse
from mizan.export import generate_edge_manifest
UserCtx = ReactContext("user")
@client(context=UserCtx)
def user_profile(request: HttpRequest, user_id: int) -> ValidOutput:
return ValidOutput(valid=True)
@client(context=UserCtx)
def profile_page(request: HttpRequest, user_id: int) -> HttpResponse:
return HttpResponse("<html>profile</html>")
register(user_profile, "user_profile")
register(profile_page, "profile_page")
manifest = generate_edge_manifest()
functions = manifest["contexts"]["user"]["functions"]
rpc_fn = next(f for f in functions if f["name"] == "user_profile")
view_fn = next(f for f in functions if f["name"] == "profile_page")
self.assertEqual(rpc_fn["path"], "rpc")
self.assertEqual(view_fn["path"], "view")
def test_manifest_with_view_urls(self):
"""Manifest includes view URLs when provided."""
from mizan.export import generate_edge_manifest
UserCtx = ReactContext("user")
@client(context=UserCtx)
def user_profile(request: HttpRequest, user_id: int) -> ValidOutput:
return ValidOutput(valid=True)
register(user_profile, "user_profile")
manifest = generate_edge_manifest(
view_urls={"user": ["/profile/:user_id/", "/u/:user_id/"]}
)
user_ctx = manifest["contexts"]["user"]
self.assertEqual(user_ctx["views"], ["/profile/:user_id/", "/u/:user_id/"])
def test_manifest_custom_base_url(self):
"""Manifest respects custom base URL."""
from mizan.export import generate_edge_manifest
@client(context=ReactContext("data"))
def some_data(request: HttpRequest) -> ValidOutput:
return ValidOutput(valid=True)
register(some_data, "some_data")
manifest = generate_edge_manifest(base_url="/v2/api")
self.assertEqual(
manifest["contexts"]["data"]["endpoints"],
["/v2/api/ctx/data/"],
)
def test_manifest_multiple_contexts(self):
"""Manifest handles multiple contexts correctly."""
from mizan.export import generate_edge_manifest
UserCtx = ReactContext("user")
NotifCtx = ReactContext("notifications")
@client(context=UserCtx)
def user_profile(request: HttpRequest, user_id: int) -> ValidOutput:
return ValidOutput(valid=True)
@client(context=NotifCtx)
def notifications(request: HttpRequest) -> ValidOutput:
return ValidOutput(valid=True)
register(user_profile, "user_profile")
register(notifications, "notifications")
manifest = generate_edge_manifest()
self.assertIn("user", manifest["contexts"])
self.assertIn("notifications", manifest["contexts"])
self.assertEqual(manifest["contexts"]["user"]["params"], ["user_id"])
self.assertEqual(manifest["contexts"]["notifications"]["params"], [])
def test_manifest_json_deterministic(self):
"""Manifest JSON output is deterministic (sorted keys)."""
from mizan.export import generate_edge_manifest_json
@client(context=ReactContext("b_ctx"))
def b_func(request: HttpRequest) -> ValidOutput:
return ValidOutput(valid=True)
@client(context=ReactContext("a_ctx"))
def a_func(request: HttpRequest) -> ValidOutput:
return ValidOutput(valid=True)
register(b_func, "b_func")
register(a_func, "a_func")
json1 = generate_edge_manifest_json()
json2 = generate_edge_manifest_json()
self.assertEqual(json1, json2)
# Keys should be sorted
parsed = json.loads(json1)
context_keys = list(parsed["contexts"].keys())
self.assertEqual(context_keys, sorted(context_keys))