diff --git a/packages/mizan-django/src/mizan/export/__init__.py b/packages/mizan-django/src/mizan/export/__init__.py index d4148c0..a3132f0 100644 --- a/packages/mizan-django/src/mizan/export/__init__.py +++ b/packages/mizan-django/src/mizan/export/__init__.py @@ -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) diff --git a/packages/mizan-django/src/mizan/management/commands/export_edge_manifest.py b/packages/mizan-django/src/mizan/management/commands/export_edge_manifest.py new file mode 100644 index 0000000..ec26399 --- /dev/null +++ b/packages/mizan-django/src/mizan/management/commands/export_edge_manifest.py @@ -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) diff --git a/packages/mizan-django/src/mizan/tests/test_core.py b/packages/mizan-django/src/mizan/tests/test_core.py index a9c0761..3f9d7a2 100644 --- a/packages/mizan-django/src/mizan/tests/test_core.py +++ b/packages/mizan-django/src/mizan/tests/test_core.py @@ -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("profile") + + 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))