Extract client/function.py to mizan-core (Tier B)
The @client decorator + ServerFunction base + composition machinery is mostly framework-agnostic. The only Django couplings were typing (HttpRequest in __init__ and submit_handler signatures) and runtime view-path detection (HttpResponseBase isinstance/issubclass checks). Replaced both with backend-extension hooks: - HttpRequest type hints → Any. Type Protocol can be tightened later. - HttpResponseBase view-path detection → set_framework_response_base(cls) hook in mizan_core.client.function. Backends register their framework's response base at import time. is_framework_response(obj_or_cls) handles both instance and subclass checks via the registered base. mizan-django registers HttpResponseBase via mizan/client/__init__.py before any @client-decorated code is loaded. FastAPI would similarly register starlette.responses.Response. Direct consumers updated: - mizan/setup/discovery.py: ServerFunction import path - mizan/forms/__init__.py: ServerFunction + create_form_functions imports mizan/client/__init__.py keeps its public re-export surface stable so 'from mizan.client import client, ServerFunction, …' continues to work for downstream Django consumers. Verified: - mizan-core: 15/15 - mizan-django: 348 pass, 21 skip, 0 fail - mizan-ts edge-compat: 34/34 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
894
cores/mizan-python/src/mizan_core/client/function.py
Normal file
894
cores/mizan-python/src/mizan_core/client/function.py
Normal file
@@ -0,0 +1,894 @@
|
||||
"""
|
||||
mizan Server Functions - Core Primitive
|
||||
|
||||
Server functions are the core primitive. Everything else builds on them.
|
||||
|
||||
Two styles supported:
|
||||
|
||||
1. Function-based (recommended, Django Ninja style):
|
||||
@client("update-profile")
|
||||
def update_profile(request, input: UpdateProfileInput) -> UpdateProfileOutput:
|
||||
return UpdateProfileOutput(success=True)
|
||||
|
||||
2. Class-based (for complex cases):
|
||||
class UpdateProfile(ServerFunction):
|
||||
def call(self, input: UpdateProfileInput) -> UpdateProfileOutput:
|
||||
return UpdateProfileOutput(success=True)
|
||||
register(UpdateProfile, 'update-profile')
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import warnings
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
ClassVar,
|
||||
Generic,
|
||||
Literal,
|
||||
TypeVar,
|
||||
Union,
|
||||
get_args,
|
||||
get_origin,
|
||||
get_type_hints,
|
||||
)
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
# ─── Framework-response-base hook ───────────────────────────────────────────
|
||||
#
|
||||
# View-path detection — distinguishing functions that return data (RPC path)
|
||||
# from functions that return a framework-native response object (view path) —
|
||||
# requires knowing the framework's response base class. Each backend adapter
|
||||
# registers its base class here at import time.
|
||||
#
|
||||
# Django sets this to django.http.HttpResponseBase. FastAPI would set it to
|
||||
# starlette.responses.Response. If unset, all functions are treated as RPC.
|
||||
|
||||
_framework_response_base: type | None = None
|
||||
|
||||
|
||||
def set_framework_response_base(cls: type) -> None:
|
||||
"""Backends register their framework's response base class for view-path detection."""
|
||||
global _framework_response_base
|
||||
_framework_response_base = cls
|
||||
|
||||
|
||||
def is_framework_response(obj_or_cls: Any) -> bool:
|
||||
"""True if obj_or_cls is, or is a subclass of, the registered framework response base."""
|
||||
if _framework_response_base is None:
|
||||
return False
|
||||
if isinstance(obj_or_cls, type):
|
||||
return issubclass(obj_or_cls, _framework_response_base)
|
||||
return isinstance(obj_or_cls, _framework_response_base)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# REACT CONTEXT - Named context marker
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class ReactContext:
|
||||
"""
|
||||
A named context that groups server functions into one provider and one fetch.
|
||||
|
||||
Usage:
|
||||
UserContext = ReactContext('user')
|
||||
|
||||
@client(context=UserContext)
|
||||
def user_profile(request, user_id: int) -> ProfileShape: ...
|
||||
|
||||
@client(context=UserContext)
|
||||
def user_orders(request, user_id: int) -> list[OrderShape]: ...
|
||||
|
||||
@client(affects=UserContext)
|
||||
def edit_profile(request, name: str) -> dict: ...
|
||||
|
||||
@client(affects=[UserContext, OrderContext])
|
||||
def change_plan(request) -> dict: ...
|
||||
"""
|
||||
|
||||
def __init__(self, name: str):
|
||||
if not name or not isinstance(name, str):
|
||||
raise ValueError("ReactContext name must be a non-empty string")
|
||||
self.name = name
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"ReactContext({self.name!r})"
|
||||
|
||||
|
||||
# Built-in global context (auto-mounted at root, SSR-hydrated)
|
||||
GlobalContext = ReactContext("global")
|
||||
|
||||
|
||||
# Context parameter type: a ReactContext instance, a raw string, or False
|
||||
ContextMode = ReactContext | str | Literal[False]
|
||||
|
||||
|
||||
TInput = TypeVar("TInput", bound=BaseModel)
|
||||
TOutput = TypeVar("TOutput", bound=BaseModel)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# SERVER FUNCTION - The Core Primitive
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class ServerFunction(ABC, Generic[TInput, TOutput]):
|
||||
"""
|
||||
Class-based server function (for complex cases).
|
||||
|
||||
For simple functions, use the @client decorator instead.
|
||||
|
||||
Usage:
|
||||
class UpdateProfile(ServerFunction):
|
||||
def call(self, input: UpdateProfileInput) -> UpdateProfileOutput:
|
||||
self.user.name = input.name
|
||||
self.user.save()
|
||||
return UpdateProfileOutput(success=True)
|
||||
|
||||
register(UpdateProfile, 'update-profile')
|
||||
"""
|
||||
|
||||
# Registration name (set by register())
|
||||
name: ClassVar[str]
|
||||
|
||||
# Metadata for code generation
|
||||
_meta: ClassVar[dict[str, Any]] = {}
|
||||
|
||||
# Schema classes (set automatically from type hints or explicitly)
|
||||
Input: ClassVar[type[BaseModel]] = BaseModel
|
||||
Output: ClassVar[type[BaseModel]] = BaseModel
|
||||
|
||||
def __init__(self, request: Any):
|
||||
"""Initialize with the framework's request object (HttpRequest in Django, Request in FastAPI, etc.)."""
|
||||
self.request = request
|
||||
|
||||
@property
|
||||
def user(self):
|
||||
"""Shortcut to request.user."""
|
||||
return self.request.user
|
||||
|
||||
@abstractmethod
|
||||
def call(self, input: TInput) -> TOutput:
|
||||
"""
|
||||
Execute the function.
|
||||
|
||||
Args:
|
||||
input: Validated input data
|
||||
|
||||
Returns:
|
||||
Output instance
|
||||
"""
|
||||
raise NotImplementedError(f"{self.__class__.__name__} must implement call()")
|
||||
|
||||
@classmethod
|
||||
def get_schema_export(cls) -> dict[str, Any]:
|
||||
"""Export schema for TypeScript generation."""
|
||||
export = {
|
||||
"name": getattr(cls, "name", cls.__name__),
|
||||
"type": "function",
|
||||
"meta": getattr(cls, "_meta", {}),
|
||||
}
|
||||
|
||||
# Get Input/Output from class attributes
|
||||
input_cls = getattr(cls, "Input", BaseModel)
|
||||
output_cls = getattr(cls, "Output", BaseModel)
|
||||
|
||||
# Check if Input has fields
|
||||
input_schema = input_cls.model_json_schema()
|
||||
has_input = bool(input_schema.get("properties"))
|
||||
if has_input:
|
||||
export["input"] = input_schema
|
||||
export["has_input"] = has_input
|
||||
|
||||
export["output"] = output_cls.model_json_schema()
|
||||
|
||||
return export
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FUNCTION DECORATOR - Django Ninja Style
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class _FunctionWrapper(ServerFunction):
|
||||
"""Internal wrapper that makes a plain function behave like a ServerFunction."""
|
||||
|
||||
# Will be set per-wrapper instance
|
||||
_wrapped_fn: ClassVar[Callable]
|
||||
_input_cls: ClassVar[type[BaseModel] | None]
|
||||
_output_cls: ClassVar[type[BaseModel]]
|
||||
_param_names: ClassVar[list[str]] = []
|
||||
_is_primitive_output: ClassVar[bool] = False
|
||||
|
||||
def call(self, input):
|
||||
"""Execute the wrapped function, unpacking input into individual args."""
|
||||
if input is not None and self._param_names:
|
||||
# Unpack validated model into keyword arguments
|
||||
kwargs = {name: getattr(input, name) for name in self._param_names}
|
||||
result = self._wrapped_fn(self.request, **kwargs)
|
||||
else:
|
||||
result = self._wrapped_fn(self.request)
|
||||
|
||||
# View path — return a framework-native response directly (no serialization)
|
||||
if is_framework_response(result):
|
||||
return result
|
||||
|
||||
# Wrap primitive returns in the generated output model
|
||||
if self._is_primitive_output:
|
||||
return self._output_cls(result=result)
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def get_schema_export(cls) -> dict[str, Any]:
|
||||
"""Export schema for TypeScript generation."""
|
||||
export = {
|
||||
"name": getattr(cls, "name", cls.__name__),
|
||||
"type": "function",
|
||||
"meta": getattr(cls, "_meta", {}),
|
||||
}
|
||||
|
||||
# Use stored schema classes
|
||||
if cls._input_cls is not None:
|
||||
input_schema = cls._input_cls.model_json_schema()
|
||||
has_input = bool(input_schema.get("properties"))
|
||||
if has_input:
|
||||
export["input"] = input_schema
|
||||
export["has_input"] = has_input
|
||||
else:
|
||||
export["has_input"] = False
|
||||
|
||||
export["output"] = cls._output_cls.model_json_schema()
|
||||
|
||||
return export
|
||||
|
||||
|
||||
# Valid string values for auth parameter
|
||||
_VALID_AUTH_STRINGS = frozenset({"required", "staff", "superuser"})
|
||||
|
||||
|
||||
def _resolve_context(context: ContextMode) -> str | Literal[False]:
|
||||
"""Resolve a context parameter to its name string."""
|
||||
if context is False:
|
||||
return False
|
||||
if isinstance(context, ReactContext):
|
||||
return context.name
|
||||
if isinstance(context, str):
|
||||
if not context.strip():
|
||||
raise ValueError("context must be a non-empty string, ReactContext, or False.")
|
||||
if context == "local":
|
||||
warnings.warn(
|
||||
"context='local' is deprecated. Use ReactContext('name') instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=3,
|
||||
)
|
||||
return context
|
||||
raise ValueError(
|
||||
f"context must be a ReactContext, a string, or False. Got {type(context).__name__}."
|
||||
)
|
||||
|
||||
|
||||
# Affects parameter type
|
||||
AffectsTarget = ReactContext | str | type["ServerFunction"]
|
||||
AffectsMode = AffectsTarget | list[AffectsTarget] | None
|
||||
|
||||
|
||||
def client(
|
||||
fn: Callable = None,
|
||||
*,
|
||||
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,
|
||||
rev: int = 0,
|
||||
cache: int | bool = True,
|
||||
) -> type[ServerFunction] | Callable[[Callable], type[ServerFunction]]:
|
||||
"""
|
||||
Register a function as a server function.
|
||||
|
||||
Args:
|
||||
context: Named context for React state management.
|
||||
- False (default): Not a context, just a callable function.
|
||||
- ReactContext instance: groups functions into a named context.
|
||||
- GlobalContext: reserved, auto-mounted at root, SSR-hydrated.
|
||||
|
||||
affects: Declare which contexts or functions this mutation invalidates.
|
||||
Mutually exclusive with context=.
|
||||
Scoping is automatic via argument name matching.
|
||||
|
||||
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.
|
||||
|
||||
Usage:
|
||||
UserContext = ReactContext('user')
|
||||
|
||||
@client(context=UserContext)
|
||||
def user_profile(request, user_id: int) -> ProfileOutput: ...
|
||||
|
||||
@client(affects=UserContext)
|
||||
def update_profile(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
|
||||
"""
|
||||
# Resolve context to name string
|
||||
resolved_context = _resolve_context(context)
|
||||
|
||||
# Validate affects parameter
|
||||
if affects is not None:
|
||||
if resolved_context is not False:
|
||||
raise ValueError(
|
||||
"context= and affects= are mutually exclusive. "
|
||||
"A function cannot be both a context reader and a mutation."
|
||||
)
|
||||
|
||||
# Validate auth parameter
|
||||
if auth is not None:
|
||||
if isinstance(auth, str) and auth not in _VALID_AUTH_STRINGS:
|
||||
raise ValueError(
|
||||
f"Invalid auth value '{auth}'. "
|
||||
f"Must be one of: {', '.join(sorted(_VALID_AUTH_STRINGS))}, True, or a callable."
|
||||
)
|
||||
|
||||
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, rev=rev, cache=cache,
|
||||
)
|
||||
|
||||
# Support both @client and @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, rev=rev, cache=cache,
|
||||
)
|
||||
return decorator
|
||||
|
||||
|
||||
def _normalize_affects(affects: AffectsMode) -> list[dict[str, str]] | None:
|
||||
"""Normalize the affects parameter into a list of target descriptors."""
|
||||
if affects is None:
|
||||
return None
|
||||
|
||||
items = affects if isinstance(affects, list) else [affects]
|
||||
result = []
|
||||
for item in items:
|
||||
if isinstance(item, ReactContext):
|
||||
result.append({"type": "context", "name": item.name})
|
||||
elif isinstance(item, str):
|
||||
result.append({"type": "context", "name": item})
|
||||
elif isinstance(item, type) and issubclass(item, ServerFunction):
|
||||
fn_meta = getattr(item, "_meta", {})
|
||||
fn_ctx = fn_meta.get("context")
|
||||
result.append({
|
||||
"type": "function",
|
||||
"name": getattr(item, "__name__", str(item)),
|
||||
"context": fn_ctx or None,
|
||||
})
|
||||
else:
|
||||
raise ValueError(
|
||||
f"affects items must be ReactContext instances, context name strings, "
|
||||
f"or @client function references. Got {type(item)}"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _create_server_function(
|
||||
fn: Callable,
|
||||
*,
|
||||
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,
|
||||
rev: int = 0,
|
||||
cache: int | bool = True,
|
||||
) -> type[ServerFunction]:
|
||||
"""Internal helper that creates a ServerFunction from a decorated function."""
|
||||
from pydantic import create_model
|
||||
|
||||
# Use function name directly
|
||||
name = fn.__name__
|
||||
|
||||
# Extract type hints and signature
|
||||
hints = get_type_hints(fn)
|
||||
sig = inspect.signature(fn)
|
||||
params = list(sig.parameters.items())
|
||||
|
||||
# Skip 'request' parameter (first param)
|
||||
input_params = params[1:] if params else []
|
||||
|
||||
# Build input schema from function parameters
|
||||
if input_params:
|
||||
# Build field definitions for create_model
|
||||
# Format: {field_name: (type, default) or (type, ...)}
|
||||
fields = {}
|
||||
for param_name, param in input_params:
|
||||
param_type = hints.get(param_name, Any)
|
||||
|
||||
if param.default is inspect.Parameter.empty:
|
||||
# Required field
|
||||
fields[param_name] = (param_type, ...)
|
||||
else:
|
||||
# Optional field with default
|
||||
fields[param_name] = (param_type, param.default)
|
||||
|
||||
# Create dynamic Pydantic model
|
||||
input_cls = create_model(f"{fn.__name__}_Input", **fields)
|
||||
else:
|
||||
input_cls = None
|
||||
|
||||
# Get output type from return annotation
|
||||
output_type = hints.get("return")
|
||||
if output_type is None:
|
||||
raise TypeError(f"Server function '{name}' must have a return type annotation")
|
||||
|
||||
# Detect view path: function returns a framework-native response type
|
||||
# (e.g. Django HttpResponse, FastAPI Response). View functions often just
|
||||
# have -> HttpResponse with no Pydantic model.
|
||||
is_view_path = is_framework_response(output_type)
|
||||
|
||||
if is_view_path:
|
||||
# View path — no Pydantic output wrapping needed
|
||||
output_cls = BaseModel # placeholder, never used for serialization
|
||||
is_primitive_output = False
|
||||
else:
|
||||
# RPC path — resolve output type
|
||||
import types
|
||||
|
||||
def is_basemodel_type(t: Any) -> bool:
|
||||
"""Check if type is a BaseModel subclass, handling Optional/Union."""
|
||||
if isinstance(t, type) and issubclass(t, BaseModel):
|
||||
return True
|
||||
origin = get_origin(t)
|
||||
if origin is Union or isinstance(t, types.UnionType):
|
||||
args = get_args(t)
|
||||
for arg in args:
|
||||
if (
|
||||
arg is not type(None)
|
||||
and isinstance(arg, type)
|
||||
and issubclass(arg, BaseModel)
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
if is_basemodel_type(output_type):
|
||||
output_cls = output_type
|
||||
is_primitive_output = False
|
||||
else:
|
||||
output_cls = create_model(f"{fn.__name__}_Output", result=(output_type, ...))
|
||||
is_primitive_output = True
|
||||
|
||||
# Store param names for unpacking validated input
|
||||
param_names = [p[0] for p in input_params]
|
||||
|
||||
# Create a unique wrapper class for this function
|
||||
class FunctionWrapper(_FunctionWrapper):
|
||||
_param_names: ClassVar[list[str]] = param_names
|
||||
|
||||
FunctionWrapper.__name__ = fn.__name__
|
||||
FunctionWrapper.__doc__ = fn.__doc__
|
||||
FunctionWrapper.__module__ = fn.__module__ # Critical for discovery
|
||||
FunctionWrapper._wrapped_fn = staticmethod(fn)
|
||||
FunctionWrapper._input_cls = input_cls
|
||||
FunctionWrapper._output_cls = output_cls
|
||||
FunctionWrapper._is_primitive_output = is_primitive_output
|
||||
|
||||
# Set Input/Output class attributes for compatibility
|
||||
if input_cls is not None:
|
||||
FunctionWrapper.Input = input_cls
|
||||
FunctionWrapper.Output = output_cls
|
||||
|
||||
# Build metadata
|
||||
meta = {}
|
||||
|
||||
# View path flag (function returns HttpResponse, no codegen)
|
||||
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
|
||||
|
||||
# Affects: mutation invalidation targets
|
||||
normalized_affects = _normalize_affects(affects)
|
||||
if normalized_affects:
|
||||
meta["affects"] = normalized_affects
|
||||
|
||||
# WebSocket: enable WebSocket transport
|
||||
if websocket:
|
||||
meta["websocket"] = True
|
||||
|
||||
# Auth requirement
|
||||
if auth is not None:
|
||||
if auth is True:
|
||||
meta["auth"] = "required"
|
||||
elif callable(auth):
|
||||
meta["auth"] = auth
|
||||
else:
|
||||
meta["auth"] = auth
|
||||
|
||||
# Revision: bumped by developer when function logic changes.
|
||||
# Part of the HMAC cache key — old entries become unreachable orphans.
|
||||
if rev != 0:
|
||||
meta["rev"] = rev
|
||||
|
||||
# Cache policy: True=forever (default), False=no-store, int=TTL seconds
|
||||
if cache is not True:
|
||||
meta["cache"] = cache
|
||||
|
||||
# Always assign a fresh dict to prevent shared-dict mutation across classes
|
||||
FunctionWrapper._meta = {**meta}
|
||||
|
||||
# Note: Registration happens via discovery (mizan_clients), not here.
|
||||
# This allows the decorator to be used without import-time side effects.
|
||||
|
||||
return FunctionWrapper
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# COMPOSE - Combine multiple contexts into a single provider
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class ComposedContext:
|
||||
"""
|
||||
Marker class for composed contexts.
|
||||
|
||||
Stores metadata about the composition for schema export.
|
||||
"""
|
||||
|
||||
name: str
|
||||
_meta: dict[str, Any]
|
||||
_children: list[type[ServerFunction] | "ComposedContext"]
|
||||
_leaves: list[type[ServerFunction]]
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
name: str,
|
||||
children: list,
|
||||
leaves: list,
|
||||
on_server: bool,
|
||||
websocket: bool,
|
||||
):
|
||||
self.name = name
|
||||
self._children = children
|
||||
self._leaves = leaves
|
||||
self._meta = {
|
||||
"compose": True,
|
||||
"on_server": on_server,
|
||||
"websocket": websocket,
|
||||
"children": [c.name for c in children],
|
||||
"leaves": [leaf.name for leaf in leaves],
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def get_schema_export(cls) -> dict[str, Any]:
|
||||
"""Export schema for TypeScript generation."""
|
||||
return {
|
||||
"name": cls.name,
|
||||
"type": "compose",
|
||||
"meta": cls._meta,
|
||||
"children": cls._meta.get("children", []),
|
||||
"leaves": cls._meta.get("leaves", []),
|
||||
}
|
||||
|
||||
|
||||
def _get_leaves(item) -> list[type[ServerFunction]]:
|
||||
"""Recursively collect all leaf contexts from a context or composition."""
|
||||
if isinstance(item, type) and issubclass(item, ServerFunction):
|
||||
return [item]
|
||||
elif isinstance(item, ComposedContext):
|
||||
return item._leaves.copy()
|
||||
elif hasattr(item, "_leaves"):
|
||||
# Duck typing for composed contexts
|
||||
return item._leaves.copy()
|
||||
else:
|
||||
raise TypeError(f"Expected ServerFunction or ComposedContext, got {type(item)}")
|
||||
|
||||
|
||||
def _is_context_enabled(item) -> bool:
|
||||
"""Check if an item is a context-enabled function or composition."""
|
||||
if isinstance(item, ComposedContext) or hasattr(item, "_leaves"):
|
||||
return True
|
||||
if isinstance(item, type) and issubclass(item, ServerFunction):
|
||||
meta = getattr(item, "_meta", {})
|
||||
return bool(meta.get("context"))
|
||||
return False
|
||||
|
||||
|
||||
def compose(
|
||||
*children,
|
||||
on_server: bool = False,
|
||||
websocket: bool = False,
|
||||
):
|
||||
"""
|
||||
Compose multiple contexts into a single provider.
|
||||
|
||||
Args:
|
||||
*children: Context functions (@client with a context name)
|
||||
or other @compose functions. All must be unique after flattening.
|
||||
|
||||
on_server: Bundle all calls into a single server request (default: False).
|
||||
- False: Frontend makes individual calls (mixed HTTP/WS OK)
|
||||
- True: Single bundled call. Requires transport consistency:
|
||||
all children must be HTTP-only XOR all must be websocket=True.
|
||||
|
||||
websocket: Transport for bundled call when on_server=True (default: False).
|
||||
- False: Bundled call over HTTP. All children must be HTTP-only.
|
||||
- True: Bundled call over WebSocket. All children must have websocket=True.
|
||||
|
||||
Usage:
|
||||
@client(context='local')
|
||||
def user_profile(request, user_id: int) -> ProfileOutput: ...
|
||||
|
||||
@client(context='local')
|
||||
def user_posts(request, user_id: int) -> PostsOutput: ...
|
||||
|
||||
@compose(user_profile, user_posts)
|
||||
def user_page():
|
||||
pass
|
||||
|
||||
# Frontend generates:
|
||||
# <UserPageProvider user_id={123}>
|
||||
# <App />
|
||||
# </UserPageProvider>
|
||||
|
||||
Nesting:
|
||||
@compose(ctx_a, ctx_b)
|
||||
def ab(): pass
|
||||
|
||||
@compose(ab, ctx_c) # Flattens to [ctx_a, ctx_b, ctx_c]
|
||||
def abc(): pass
|
||||
|
||||
Returns:
|
||||
A ComposedContext that can be used in other compositions.
|
||||
"""
|
||||
|
||||
def decorator(fn: Callable) -> ComposedContext:
|
||||
from mizan_core.registry import register_compose
|
||||
|
||||
name = fn.__name__
|
||||
|
||||
# Validate: all children must be context-enabled
|
||||
for i, child in enumerate(children):
|
||||
if not _is_context_enabled(child):
|
||||
child_name = getattr(
|
||||
child, "name", getattr(child, "__name__", str(child))
|
||||
)
|
||||
raise ValueError(
|
||||
f"@compose argument {i} ({child_name}) is not context-enabled. "
|
||||
f"All children must have @client(context=...) or be @compose."
|
||||
)
|
||||
|
||||
# Flatten to collect all leaves
|
||||
leaves = []
|
||||
for child in children:
|
||||
leaves.extend(_get_leaves(child))
|
||||
|
||||
# Validate: no duplicate leaves (by identity)
|
||||
seen = set()
|
||||
for leaf in leaves:
|
||||
if id(leaf) in seen:
|
||||
raise ValueError(
|
||||
f"Duplicate context '{leaf.name}' in @compose({name}). "
|
||||
f"Each context can only appear once. Use named kwargs for reuse (future feature)."
|
||||
)
|
||||
seen.add(id(leaf))
|
||||
|
||||
# Validate transport consistency when on_server=True
|
||||
if on_server:
|
||||
has_websocket = [
|
||||
getattr(leaf, "_meta", {}).get("websocket", False) for leaf in leaves
|
||||
]
|
||||
|
||||
if websocket:
|
||||
# All must have websocket=True
|
||||
if not all(has_websocket):
|
||||
non_ws = [
|
||||
leaf.name for leaf, ws in zip(leaves, has_websocket) if not ws
|
||||
]
|
||||
raise ValueError(
|
||||
f"@compose({name}, on_server=True, websocket=True) requires all children "
|
||||
f"to have websocket=True. These are HTTP-only: {non_ws}"
|
||||
)
|
||||
else:
|
||||
# All must be HTTP-only
|
||||
if any(has_websocket):
|
||||
ws_enabled = [
|
||||
leaf.name for leaf, ws in zip(leaves, has_websocket) if ws
|
||||
]
|
||||
raise ValueError(
|
||||
f"@compose({name}, on_server=True, websocket=False) requires all children "
|
||||
f"to be HTTP-only. These have websocket=True: {ws_enabled}"
|
||||
)
|
||||
|
||||
# Create composed context
|
||||
composed = ComposedContext(
|
||||
name=name,
|
||||
children=list(children),
|
||||
leaves=leaves,
|
||||
on_server=on_server,
|
||||
websocket=websocket,
|
||||
)
|
||||
|
||||
# Make it a class-like object for consistency
|
||||
composed.__name__ = name
|
||||
composed.__doc__ = fn.__doc__
|
||||
|
||||
# Register the composition
|
||||
register_compose(composed, name)
|
||||
|
||||
return composed
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# FORM HELPERS - Output types used by form server functions
|
||||
# =============================================================================
|
||||
|
||||
|
||||
class FormValidationOutput(BaseModel):
|
||||
"""Standard output for form validation."""
|
||||
|
||||
valid: bool
|
||||
errors: dict[str, list[str]]
|
||||
|
||||
|
||||
class FormSchemaField(BaseModel):
|
||||
"""Schema for a single form field."""
|
||||
|
||||
name: str
|
||||
type: str
|
||||
required: bool
|
||||
label: str
|
||||
help_text: str | None = None
|
||||
choices: list[tuple[str, str]] | None = None
|
||||
initial: Any = None
|
||||
|
||||
|
||||
class FormSchemaOutput(BaseModel):
|
||||
"""Standard output for form schema."""
|
||||
|
||||
fields: list[FormSchemaField]
|
||||
|
||||
|
||||
def create_form_functions(
|
||||
form_class: type,
|
||||
name: str,
|
||||
submit_handler: Callable[[Any, dict], BaseModel] | None = None,
|
||||
) -> tuple[type[ServerFunction], type[ServerFunction], type[ServerFunction] | None]:
|
||||
"""
|
||||
Generate server functions for a Django Form.
|
||||
|
||||
Args:
|
||||
form_class: Django Form class
|
||||
name: Base name for the functions
|
||||
submit_handler: Optional handler for form submission
|
||||
|
||||
Returns:
|
||||
Tuple of (SchemaFunction, ValidateFunction, SubmitFunction or None)
|
||||
|
||||
Usage:
|
||||
SchemaFn, ValidateFn, SubmitFn = create_form_functions(
|
||||
ContactForm,
|
||||
'contact',
|
||||
submit_handler=lambda req, data: ContactSubmitOutput(success=True),
|
||||
)
|
||||
register(SchemaFn, 'contact-schema')
|
||||
register(ValidateFn, 'contact-validate')
|
||||
register(SubmitFn, 'contact-submit')
|
||||
|
||||
Or use the helper:
|
||||
register_form(ContactForm, 'contact', submit_handler=...)
|
||||
"""
|
||||
from mizan.forms.schema_utils import build_form_schema
|
||||
|
||||
# Schema function - returns field definitions
|
||||
class FormSchema(ServerFunction):
|
||||
class Output(FormSchemaOutput):
|
||||
pass
|
||||
|
||||
def call(self, input):
|
||||
schema = build_form_schema(form_class)
|
||||
fields = [
|
||||
FormSchemaField(
|
||||
name=field.name,
|
||||
type=field.type,
|
||||
required=field.required,
|
||||
label=field.label or field.name,
|
||||
help_text=field.help_text or None,
|
||||
choices=[(c.value, c.label) for c in field.choices]
|
||||
if field.choices
|
||||
else None,
|
||||
initial=field.initial,
|
||||
)
|
||||
for field in schema.fields
|
||||
]
|
||||
return self.Output(fields=fields)
|
||||
|
||||
FormSchema.__name__ = f"{name.title().replace('-', '')}Schema"
|
||||
FormSchema._meta = {"form": True, "form_name": name, "form_role": "schema"}
|
||||
|
||||
# Validation function
|
||||
class FormDataInput(BaseModel):
|
||||
data: dict[str, Any]
|
||||
|
||||
class FormValidate(ServerFunction):
|
||||
Input = FormDataInput
|
||||
|
||||
class Output(FormValidationOutput):
|
||||
pass
|
||||
|
||||
def call(self, input):
|
||||
form = form_class(data=input.data)
|
||||
if form.is_valid():
|
||||
return self.Output(valid=True, errors={})
|
||||
return self.Output(valid=False, errors=dict(form.errors))
|
||||
|
||||
FormValidate.__name__ = f"{name.title().replace('-', '')}Validate"
|
||||
FormValidate._meta = {"form": True, "form_name": name, "form_role": "validate"}
|
||||
|
||||
# Submit function (optional)
|
||||
FormSubmit = None
|
||||
if submit_handler:
|
||||
|
||||
class FormSubmit(ServerFunction):
|
||||
Input = FormDataInput
|
||||
|
||||
def call(self, input):
|
||||
# Validate first
|
||||
form = form_class(data=input.data)
|
||||
if not form.is_valid():
|
||||
raise ValueError("Form validation failed")
|
||||
# Call handler
|
||||
return submit_handler(self.request, form.cleaned_data)
|
||||
|
||||
FormSubmit.__name__ = f"{name.title().replace('-', '')}Submit"
|
||||
FormSubmit._meta = {"form": True, "form_name": name, "form_role": "submit"}
|
||||
|
||||
return FormSchema, FormValidate, FormSubmit
|
||||
Reference in New Issue
Block a user