Add private=True, route=, methods= to @client decorator

private=True: server-internal functions (webhooks, cron) that emit
invalidation but are not client-callable. Rejected from POST /call/
with 403. No codegen. Appears in manifest for Edge.

  @client(affects='subscription', private=True, route='/webhooks/stripe/', methods=['POST'])
  def stripe_webhook(request) -> HttpResponse: ...

route=: Mizan-owned URL pattern for view-path functions. Registered
during autodiscovery. Populates page_routes in the manifest for
Edge/PSR to resolve during invalidation.

methods=: HTTP methods for the route. Defaults to ['GET'] for context
functions, ['POST'] for mutations.

Extended Edge manifest with:
- mutations section: affects, auto_scoped_params, private, route
- render_strategy: "psr" (no user params) or "dynamic_cached" (user-scoped)
- user_scoped: derived from param names matching common identity params
- page_routes: from route= on view-path functions + external view_urls

323 Django tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-03 22:40:36 -04:00
parent 28e517e6ee
commit d228c7ab1b
4 changed files with 310 additions and 26 deletions

View File

@@ -335,8 +335,15 @@ def execute_function(
message=message,
)
# Check auth requirement BEFORE executing
# Reject private functions from RPC dispatch
meta = getattr(view_class, "_meta", {})
if meta.get("private"):
return FunctionError(
code=ErrorCode.FORBIDDEN,
message="Function is not client-callable",
)
# Check auth requirement BEFORE executing
auth_requirement = meta.get("auth")
auth_error = _check_auth_requirement(request, auth_requirement)
if auth_error is not None:

View File

@@ -256,6 +256,9 @@ def client(
*,
context: ContextMode = False,
affects: AffectsMode = None,
private: bool = False,
route: str | None = None,
methods: list[str] | None = None,
websocket: bool = False,
auth: bool | str | Callable[[Any], bool] | None = None,
) -> type[ServerFunction] | Callable[[Callable], type[ServerFunction]]:
@@ -269,21 +272,26 @@ def client(
- GlobalContext: reserved, auto-mounted at root, SSR-hydrated.
affects: Declare which contexts or functions this mutation invalidates.
- A ReactContext instance or context name string: invalidates entire context
- A function name string: invalidates just that function within its context
- A list of the above: invalidates multiple targets
Mutually exclusive with context=.
Scoping is automatic via argument name matching.
Scoping is automatic: if the mutation's arguments overlap with the
context's params by name, the invalidation is scoped to those values.
private: If True, the function is not client-callable.
- Not exposed as an RPC endpoint
- No generated TypeScript
- Still participates in the invalidation graph
- Use for webhooks, cron jobs, internal mutations
route: URL route pattern for view-path functions.
Mizan registers this route during autodiscovery.
Example: '/profile/<user_id>/', '/webhooks/stripe/'
methods: HTTP methods allowed for the route.
Default: ['GET'] for context functions, ['POST'] for mutations.
Example: ['POST'], ['GET', 'POST']
websocket: Enable WebSocket RPC transport (default: False).
auth: Authentication requirement.
- None (default): No auth required
- True or 'required': Must be authenticated
- 'staff'/'superuser': role-based
- callable(request) -> bool: Custom check
Usage:
UserContext = ReactContext('user')
@@ -291,17 +299,16 @@ def client(
@client(context=UserContext)
def user_profile(request, user_id: int) -> ProfileOutput: ...
# Broad invalidation — all UserContext instances
@client(affects=UserContext)
def reset_all_profiles(request) -> dict: ...
# Auto-scoped — user_id matches, only invalidates UserContext(user_id=5)
@client(affects=UserContext)
def update_profile(request, user_id: int, name: str) -> dict: ...
# Function-level — only user_profile refetches, not user_orders
@client(affects='user_profile')
def update_name(request, user_id: int, name: str) -> dict: ...
# View with route — Mizan owns the URL
@client(context=UserContext, route='/profile/<user_id>/')
def profile_page(request, user_id: int) -> HttpResponse: ...
# Private webhook — not client-callable, emits invalidation
@client(affects='subscription', private=True, route='/webhooks/stripe/', methods=['POST'])
def stripe_webhook(request) -> HttpResponse: ...
Returns:
A ServerFunction class that wraps the function
@@ -328,6 +335,7 @@ def client(
def decorator(fn: Callable) -> type[ServerFunction]:
return _create_server_function(
fn, context=resolved_context, affects=affects,
private=private, route=route, methods=methods,
websocket=websocket, auth=auth,
)
@@ -335,6 +343,7 @@ def client(
if fn is not None:
return _create_server_function(
fn, context=resolved_context, affects=affects,
private=private, route=route, methods=methods,
websocket=websocket, auth=auth,
)
return decorator
@@ -373,6 +382,9 @@ def _create_server_function(
*,
context: str | Literal[False] = False,
affects: str | type["ServerFunction"] | list[str | type["ServerFunction"]] | None = None,
private: bool = False,
route: str | None = None,
methods: list[str] | None = None,
websocket: bool = False,
auth: bool | str | None = None,
) -> type[ServerFunction]:
@@ -480,6 +492,15 @@ def _create_server_function(
if is_view_path:
meta["view_path"] = True
# Private flag (not client-callable, no codegen, no RPC endpoint)
if private:
meta["private"] = True
# Route (Mizan-owned URL pattern for view-path functions)
if route:
meta["route"] = route
meta["methods"] = methods or (["GET"] if context else ["POST"])
# Context name (any non-empty string)
if context:
meta["context"] = context

View File

@@ -384,16 +384,20 @@ def generate_edge_manifest(
"""
from pydantic import BaseModel as PydanticBaseModel
# Common user identity param names for user_scoped detection
_USER_SCOPED_PARAMS = {"user_id", "user", "owner_id", "account_id"}
groups = get_context_groups()
registry = get_registry()
all_functions = registry.get("functions", {})
manifest: dict[str, Any] = {"contexts": {}}
manifest: dict[str, Any] = {"contexts": {}, "mutations": {}}
for ctx_name, fn_names in groups.items():
# Collect params from all functions in this context
# Collect params and routes from all functions in this context
param_names: set[str] = set()
functions_meta: list[dict[str, Any]] = []
page_routes: list[str] = []
for fn_name in fn_names:
fn_cls = all_functions.get(fn_name)
@@ -412,23 +416,86 @@ def generate_edge_manifest(
):
param_names.update(input_cls.model_fields.keys())
functions_meta.append({
fn_entry: dict[str, Any] = {
"name": fn_name,
"path": "view" if is_view else "rpc",
})
}
# Collect routes from view-path functions
fn_route = meta.get("route")
if fn_route:
fn_entry["route"] = fn_route
fn_entry["methods"] = meta.get("methods", ["GET"])
page_routes.append(fn_route)
functions_meta.append(fn_entry)
sorted_params = sorted(param_names)
user_scoped = bool(param_names & _USER_SCOPED_PARAMS)
ctx_entry: dict[str, Any] = {
"functions": functions_meta,
"endpoints": [f"{base_url}/ctx/{ctx_name}/"],
"params": sorted(param_names),
"params": sorted_params,
"user_scoped": user_scoped,
"render_strategy": "dynamic_cached" if user_scoped else "psr",
}
# Add view URLs if provided
# Add page routes from view-path functions with route=
if page_routes:
ctx_entry["page_routes"] = page_routes
# Add externally-declared view URLs
if view_urls and ctx_name in view_urls:
ctx_entry["views"] = view_urls[ctx_name]
ctx_entry.setdefault("page_routes", []).extend(view_urls[ctx_name])
manifest["contexts"][ctx_name] = ctx_entry
# Mutations section — all functions with affects=
for fn_name, fn_cls in all_functions.items():
meta = getattr(fn_cls, "_meta", {})
affects = meta.get("affects")
if not affects:
continue
# Resolve context names from affects targets
affected_contexts = []
for target in affects:
if target["type"] == "context":
affected_contexts.append(target["name"])
elif target["type"] == "function" and target.get("context"):
affected_contexts.append(target["context"])
affected_contexts = list(dict.fromkeys(affected_contexts))
# Determine which params auto-scope
auto_scoped = []
input_cls = getattr(fn_cls, "Input", None)
if input_cls and input_cls is not PydanticBaseModel and hasattr(input_cls, "model_fields"):
fn_params = set(input_cls.model_fields.keys())
for ctx_name in affected_contexts:
ctx_params = set()
for ctx_fn_name in groups.get(ctx_name, []):
ctx_fn_cls = all_functions.get(ctx_fn_name)
if ctx_fn_cls:
ctx_input = getattr(ctx_fn_cls, "Input", None)
if ctx_input and ctx_input is not PydanticBaseModel and hasattr(ctx_input, "model_fields"):
ctx_params.update(ctx_input.model_fields.keys())
auto_scoped.extend(sorted(fn_params & ctx_params))
auto_scoped = list(dict.fromkeys(auto_scoped))
mutation_entry: dict[str, Any] = {
"affects": affected_contexts,
}
if auto_scoped:
mutation_entry["auto_scoped_params"] = auto_scoped
if meta.get("private"):
mutation_entry["private"] = True
if meta.get("route"):
mutation_entry["route"] = meta["route"]
mutation_entry["methods"] = meta.get("methods", ["POST"])
manifest["mutations"][fn_name] = mutation_entry
return manifest

View File

@@ -2538,7 +2538,7 @@ class EdgeManifestTests(TestCase):
)
user_ctx = manifest["contexts"]["user"]
self.assertEqual(user_ctx["views"], ["/profile/:user_id/", "/u/:user_id/"])
self.assertEqual(user_ctx["page_routes"], ["/profile/:user_id/", "/u/:user_id/"])
def test_manifest_custom_base_url(self):
"""Manifest respects custom base URL."""
@@ -2604,3 +2604,192 @@ class EdgeManifestTests(TestCase):
parsed = json.loads(json1)
context_keys = list(parsed["contexts"].keys())
self.assertEqual(context_keys, sorted(context_keys))
# =============================================================================
# Private Functions + Route Tests
# =============================================================================
class PrivateAndRouteTests(TestCase):
"""Tests for private=True, route=, and methods= on @client."""
def setUp(self):
clear_registry()
def tearDown(self):
clear_registry()
def test_private_meta_flag(self):
"""private=True sets the flag in _meta."""
from django.http import HttpResponse
@client(affects="subscription", private=True)
def stripe_webhook(request: HttpRequest) -> HttpResponse:
return HttpResponse(status=200)
self.assertTrue(stripe_webhook._meta.get("private"))
def test_private_rejected_from_rpc(self):
"""Private functions cannot be called via POST /api/mizan/call/."""
from django.http import HttpResponse
@client(affects="subscription", private=True)
def stripe_webhook(request: HttpRequest) -> HttpResponse:
return HttpResponse(status=200)
register(stripe_webhook, "stripe_webhook")
response = self.client.post(
"/api/mizan/call/",
data=json.dumps({"fn": "stripe_webhook", "args": {}}),
content_type="application/json",
)
self.assertEqual(response.status_code, 403)
data = response.json()
self.assertTrue(data["error"])
self.assertEqual(data["code"], "FORBIDDEN")
def test_route_stored_in_meta(self):
"""route= and methods= are stored in _meta."""
from django.http import HttpResponse
@client(
affects="subscription",
private=True,
route="/webhooks/stripe/",
methods=["POST"],
)
def stripe_webhook(request: HttpRequest) -> HttpResponse:
return HttpResponse(status=200)
self.assertEqual(stripe_webhook._meta["route"], "/webhooks/stripe/")
self.assertEqual(stripe_webhook._meta["methods"], ["POST"])
def test_route_default_methods(self):
"""Default methods: GET for context, POST for mutation."""
from django.http import HttpResponse
UserCtx = ReactContext("user")
@client(context=UserCtx, route="/profile/<user_id>/")
def profile_page(request: HttpRequest, user_id: int) -> HttpResponse:
return HttpResponse("<html>profile</html>")
@client(affects=UserCtx, route="/profile/<user_id>/update/")
def update_profile(request: HttpRequest, user_id: int) -> HttpResponse:
return HttpResponse(status=302)
self.assertEqual(profile_page._meta["methods"], ["GET"])
self.assertEqual(update_profile._meta["methods"], ["POST"])
def test_private_in_manifest(self):
"""Private functions appear in manifest mutations section."""
from django.http import HttpResponse
from mizan.export import generate_edge_manifest
SubCtx = ReactContext("subscription")
@client(context=SubCtx)
def subscription_info(request: HttpRequest) -> ValidOutput:
return ValidOutput(valid=True)
@client(
affects=SubCtx,
private=True,
route="/webhooks/stripe/",
methods=["POST"],
)
def stripe_webhook(request: HttpRequest) -> HttpResponse:
return HttpResponse(status=200)
register(subscription_info, "subscription_info")
register(stripe_webhook, "stripe_webhook")
manifest = generate_edge_manifest()
# In mutations section
self.assertIn("stripe_webhook", manifest["mutations"])
mut = manifest["mutations"]["stripe_webhook"]
self.assertTrue(mut["private"])
self.assertEqual(mut["route"], "/webhooks/stripe/")
self.assertEqual(mut["methods"], ["POST"])
self.assertEqual(mut["affects"], ["subscription"])
def test_manifest_render_strategy_user_scoped(self):
"""Contexts with user_id params are user_scoped + dynamic_cached."""
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()
ctx = manifest["contexts"]["user"]
self.assertTrue(ctx["user_scoped"])
self.assertEqual(ctx["render_strategy"], "dynamic_cached")
def test_manifest_render_strategy_psr(self):
"""Contexts without user-scoped params are PSR."""
from mizan.export import generate_edge_manifest
ProductCtx = ReactContext("products")
@client(context=ProductCtx)
def product_detail(request: HttpRequest, product_id: int) -> ValidOutput:
return ValidOutput(valid=True)
register(product_detail, "product_detail")
manifest = generate_edge_manifest()
ctx = manifest["contexts"]["products"]
self.assertFalse(ctx["user_scoped"])
self.assertEqual(ctx["render_strategy"], "psr")
def test_manifest_auto_scoped_params(self):
"""Mutations with matching params show auto_scoped_params in manifest."""
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(affects=UserCtx)
def update_profile(request: HttpRequest, user_id: int, name: str) -> ValidOutput:
return ValidOutput(valid=True)
register(user_profile, "user_profile")
register(update_profile, "update_profile")
manifest = generate_edge_manifest()
mut = manifest["mutations"]["update_profile"]
self.assertIn("user_id", mut["auto_scoped_params"])
def test_manifest_page_routes_from_decorator(self):
"""View-path functions with route= populate page_routes in manifest."""
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, route="/profile/<user_id>/")
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()
ctx = manifest["contexts"]["user"]
self.assertIn("/profile/<user_id>/", ctx["page_routes"])