rythazhur/mizan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Mizan IR: cut over to KDL, delete OpenAPI envelope

Browse Source 
Replaces the transitional OpenAPI 3.0 + `x-mizan-*` extensions
substrate with the canonical Mizan IR as KDL, per docs/AFI_ARCHITECTURE.md:
"KDL is the contract; everything else (REST envelopes, OpenAPI
documents, framework idioms) is sediment around it."

End-to-end cutover. No transitional path left on main.

Forward direction:
  cores/mizan-python/src/mizan_core/ir.py
    build_ir() walks mizan_core.registry, introspects Pydantic
    models directly (no JSON-Schema indirection), and emits the
    Mizan IR document. The KDL grammar is locked in this file's
    module docstring.

Backends emit KDL:
  backends/mizan-fastapi/src/mizan_fastapi/ir.py
    `python -m mizan_fastapi.ir <module>` — CLI entry point.
  backends/mizan-django/.../management/commands/export_mizan_ir.py
    `manage.py export_mizan_ir` — Django mgmt command.

Codegen consumes KDL:
  protocol/mizan-codegen/Cargo.toml: + kdl = "6"
  protocol/mizan-codegen/src/ir.rs: NamedType { Struct/List/Enum/Alias }
    + TypeShape { Primitive/Ref/List/Optional/Enum/Union } sum types,
    replacing the JsonSchema sprawl. KDL parser walks the
    `kdl::KdlDocument` tree into typed Rust structs.
  protocol/mizan-codegen/src/fetch.rs: subprocess command switches
    to the new IR-export entry points.
  All emit modules (stage1 / react / python / rust / vue / svelte /
    channels) port their type-walkers from JsonSchema to the new
    sum types — case analysis collapses substantially.

Substrate-honesty wins beyond the moat closure:
  - `int | bool` multi-arm unions land as `TypeShape::Union` (was
    silently coerced to "string" before).
  - `<CamelName>Output = list[T]` returns emit as named alias
    types instead of struct-shaped wrappers, so consumer code
    `.map()` works directly on the type.
  - Pydantic field defaults flow through to `default` properties
    in KDL, then back to non-optional shape in every target.

Deleted:
  - backends/mizan-fastapi/src/mizan_fastapi/{cli,schema}.py
  - backends/mizan-django/.../export_mizan_schema.py
  - openapi-bearing half of mizan/export/__init__.py (edge
    manifest generator preserved — separate concern).
  - tests/afi/schema_normalizer.py
  - tests/fixtures/{afi_schema.json, channels_schema.json}
  - tests/fixtures/js_* baseline directories.

Verification:
  - 20 mizan-codegen unit tests green (IR deserialization,
    byte-equivalence parity across stage1/rust/python/react/vue/svelte
    against fresh KDL-driven baselines, channels structural).
  - tests/rust/run_wire_parity.py: 12/12 probes green driving
    the binary end-to-end through KDL.
  - Blazr studio-ui typechecks against the regenerated React client.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ryth Azhur
2026-05-17 19:14:47 -04:00
parent 7fb0c4a400
commit 9900f8a36f
86 changed files with 2231 additions and 2272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

472
backends/mizan-django/src/mizan/export/__init__.py
View File

@@ -1,380 +1,30 @@
"""
mizan OpenAPI Schema Generator
Mizan Edge Manifest Generator.
Generates OpenAPI 3.0 compatible schema from registered server functions.
Uses Django Ninja's battle-tested schema generation for robust Pydantic→OpenAPI conversion.
This schema is consumed by the frontend generator which uses openapi-typescript
for robust type generation.
NOTE: Schema export is only available via management command for security.
HTTP endpoint has been removed to prevent function enumeration.
Generates the Edge manifest — a static JSON mapping contexts to URL
patterns and params, consumed by Mizan Edge at deploy time for CDN
cache invalidation. Independent from the Mizan IR; the IR drives
codegen, the manifest drives CDN purging.
Usage:
python manage.py export_mizan_schema
from mizan.export import generate_edge_manifest, generate_edge_manifest_json
"""
from __future__ import annotations
import json
import re
from typing import TYPE_CHECKING, Any
from typing import Any
# Lazy imports to avoid Django settings access at module load time
# (asgi.py imports mizan before Django is fully configured)
if TYPE_CHECKING:
from django import forms
from ninja import NinjaAPI
from mizan_core.registry import get_registry, get_schema, get_context_groups, get_function
from mizan_core.type_utils import extract_list_element, extract_optional
from mizan_core.registry import get_context_groups, get_registry
__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]]:
"""
Extract field definitions with constraints from a Django Form class.
Returns a list of field metadata suitable for Zod schema generation:
- name: field name
- zodType: base Zod type ("string", "number", "boolean", "array")
- required: whether field is required
- constraints: dict of Zod-compatible constraints
Constraints include:
- min/max: for string length or number range
- email/url: for format validation
- regex: for pattern validation
- choices: for enum validation
"""
try:
# Try to instantiate form to get bound fields
form = form_class()
fields_dict = form.fields
except TypeError:
# Form requires extra args - use base_fields
fields_dict = getattr(form_class, "base_fields", {})
result = []
for name, field in fields_dict.items():
field_meta = _extract_field_constraints(name, field)
result.append(field_meta)
return result
def _extract_field_constraints(name: str, field: "forms.Field") -> dict[str, Any]:
"""
Extract Zod-compatible constraints from a single Django form field.
"""
from django import forms # Lazy import
meta: dict[str, Any] = {
"name": name,
"required": field.required,
"constraints": {},
}
# Determine base Zod type
if isinstance(field, forms.BooleanField):
meta["zodType"] = "boolean"
elif isinstance(field, (forms.IntegerField, forms.FloatField, forms.DecimalField)):
meta["zodType"] = "number"
if isinstance(field, forms.IntegerField):
meta["constraints"]["int"] = True
elif isinstance(field, forms.MultipleChoiceField):
meta["zodType"] = "array"
meta["constraints"]["items"] = "string"
elif isinstance(field, forms.FileField):
meta["zodType"] = "file"
else:
# Default to string (CharField, EmailField, URLField, etc.)
meta["zodType"] = "string"
# Extract string constraints
if hasattr(field, "max_length") and field.max_length is not None:
meta["constraints"]["max"] = field.max_length
if hasattr(field, "min_length") and field.min_length is not None:
meta["constraints"]["min"] = field.min_length
# Extract number constraints
if hasattr(field, "max_value") and field.max_value is not None:
meta["constraints"]["max"] = field.max_value
if hasattr(field, "min_value") and field.min_value is not None:
meta["constraints"]["min"] = field.min_value
# Email/URL format
if isinstance(field, forms.EmailField):
meta["constraints"]["email"] = True
elif isinstance(field, forms.URLField):
meta["constraints"]["url"] = True
# Choices (for enum validation)
if hasattr(field, "choices") and field.choices:
# Extract choice values (not labels)
choices = []
for choice in field.choices:
if isinstance(choice, (list, tuple)) and len(choice) >= 1:
# Skip empty/blank choices
if choice[0] != "":
choices.append(str(choice[0]))
else:
choices.append(str(choice))
if choices:
meta["constraints"]["choices"] = choices
# Regex validators
for validator in field.validators:
if hasattr(validator, "regex"):
# RegexValidator - extract pattern
pattern = validator.regex.pattern
meta["constraints"]["regex"] = pattern
if hasattr(validator, "message"):
meta["constraints"]["regexMessage"] = validator.message
break # Only use first regex validator
return meta
def snake_to_camel(name: str) -> str:
"""Convert snake_case or dotted.name to camelCase.
Examples:
- login -> login
- login.schema -> loginSchema
- activate_totp -> activateTotp
- activate_totp.schema -> activateTotpSchema
"""
# Split on both underscores and dots
components = re.split(r"[._]", name)
return components[0] + "".join(x.title() for x in components[1:])
def _register_schema_endpoint(
api: "NinjaAPI",
path: str,
operation_id: str,
summary: str,
input_cls: type | None,
output_cls: type,
) -> None:
"""
Register a dummy endpoint on the API for schema generation.
Sets __annotations__ directly to avoid closure capture issues
and exec() security concerns.
"""
if input_cls is not None:
def endpoint(request, data):
pass
# Set annotations directly to the actual type objects (not strings)
endpoint.__annotations__ = {"data": input_cls}
else:
def endpoint(request):
pass
# Register with Ninja
api.post(path, response=output_cls, operation_id=operation_id, summary=summary)(
endpoint
)
def generate_openapi_schema() -> dict[str, Any]:
"""
Generate OpenAPI 3.0 schema for all registered mizan functions.
Uses Django Ninja's schema generation internally to ensure proper
Pydantic→OpenAPI conversion (handling $refs, nested types, etc.).
Returns a complete OpenAPI document that can be processed by openapi-typescript.
"""
from ninja import NinjaAPI # Lazy import
from pydantic import BaseModel, RootModel, create_model # Lazy import
registry = get_registry()
functions = registry.get("functions", {})
# Create a temporary Ninja API for schema generation only
# This is NOT exposed as an HTTP endpoint - purely for leveraging Ninja's
# battle-tested Pydantic→OpenAPI conversion
schema_api = NinjaAPI(
title="mizan Server Functions",
version="1.0.0",
description="Auto-generated schema for mizan server functions",
docs_url=None, # No docs endpoint
openapi_url=None, # No openapi endpoint
)
function_metadata: list[dict[str, Any]] = []
# Store dynamically created classes so they persist for schema generation
schema_classes: dict[str, type] = {}
for name, fn_class in functions.items():
camel_name = snake_to_camel(name)
meta = getattr(fn_class, "_meta", {})
# Get Input/Output classes
input_cls = getattr(fn_class, "Input", None)
output_cls = getattr(fn_class, "Output", None) or BaseModel
# Check if input_cls is a valid Pydantic model with fields
has_input = (
input_cls is not None
and input_cls is not BaseModel
and hasattr(input_cls, "model_fields")
and bool(input_cls.model_fields)
)
# Determine type names for metadata
input_type_name = f"{camel_name}Input" if has_input else None
output_type_name = f"{camel_name}Output"
# Strip Optional so the rename gets a concrete base — nullability is
# carried on the response declaration, not the schema class itself.
output_inner, output_nullable = extract_optional(output_cls)
# Create renamed Pydantic classes for cleaner schema names
# Store them in schema_classes so they persist beyond loop scope
if has_input:
schema_classes[input_type_name] = create_model(
input_type_name, __base__=input_cls
)
if extract_list_element(output_inner) is not None:
# list[T] — RootModel makes the rename emit `type: array` rather
# than wrapping the list in a property.
schema_classes[output_type_name] = type(
output_type_name, (RootModel[output_inner],), {},
)
else:
schema_classes[output_type_name] = create_model(
output_type_name, __base__=output_inner
)
response_cls = schema_classes[output_type_name]
if output_nullable:
response_cls = response_cls | None
# Register endpoint using helper to avoid closure capture issues
_register_schema_endpoint(
api=schema_api,
path=f"/mizan/{name}",
operation_id=camel_name,
summary=fn_class.__doc__ or f"Call {name}",
input_cls=schema_classes.get(input_type_name),
output_cls=response_cls,
)
# Collect function metadata for provider generation
fn_meta_entry: dict[str, Any] = {
"name": name,
"camelName": camel_name,
"hasInput": has_input,
"inputType": input_type_name,
"outputType": output_type_name,
"transport": "websocket" if meta.get("websocket") else "http",
"isContext": meta.get("context", False),
# Form metadata
"isForm": meta.get("form", False),
"formName": meta.get("form_name"),
"formRole": meta.get("form_role"), # "schema", "validate", "submit"
}
# Affects metadata (mutation invalidation)
if meta.get("affects"):
fn_meta_entry["affects"] = meta["affects"]
if meta.get("merge"):
fn_meta_entry["merge"] = meta["merge"]
# For form schema functions, extract field definitions for Zod generation
if meta.get("form") and meta.get("form_role") == "schema":
form_class = meta.get("form_class")
if form_class is not None:
try:
fn_meta_entry["formFields"] = _extract_form_fields(form_class)
except Exception as e:
# Don't fail schema generation if field extraction fails
fn_meta_entry["formFields"] = []
fn_meta_entry["formFieldsError"] = str(e)
function_metadata.append(fn_meta_entry)
# Get the OpenAPI schema from Ninja (handles all Pydantic conversion properly)
schema = schema_api.get_openapi_schema(path_prefix="")
# Add custom extension with function metadata for provider generation
schema["x-mizan-functions"] = function_metadata
# Add x-mizan-contexts: grouped context metadata with param elevation
context_groups = get_context_groups()
if context_groups:
contexts_meta: dict[str, Any] = {}
for ctx_name, fn_names in context_groups.items():
# Analyze params across all functions in the context
param_info: dict[str, dict[str, Any]] = {}
for fn_name in fn_names:
fn_cls = get_function(fn_name)
if fn_cls is None:
continue
input_cls = getattr(fn_cls, "Input", None)
if input_cls and input_cls is not BaseModel and hasattr(input_cls, "model_fields"):
for field_name, field_info in input_cls.model_fields.items():
if field_name not in param_info:
annotation = field_info.annotation
# Map Python types to JSON schema types
type_name = "string"
if annotation in (int,):
type_name = "integer"
elif annotation in (float,):
type_name = "number"
elif annotation in (bool,):
type_name = "boolean"
param_info[field_name] = {
"type": type_name,
"sharedBy": [],
}
param_info[field_name]["sharedBy"].append(fn_name)
# A param is required if ALL functions in the context declare it
for p_name, p_meta in param_info.items():
p_meta["required"] = len(p_meta["sharedBy"]) == len(fn_names)
contexts_meta[ctx_name] = {
"functions": fn_names,
"params": param_info,
}
schema["x-mizan-contexts"] = contexts_meta
# Add x-mizan metadata to each operation
for fn_meta in function_metadata:
path = f"/mizan/{fn_meta['name']}"
if path in schema.get("paths", {}):
schema["paths"][path]["post"]["x-mizan"] = {
"transport": fn_meta["transport"],
"isContext": fn_meta["isContext"],
}
return schema
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,
@@ -394,14 +44,10 @@ def generate_edge_manifest(
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
# Common user identity param names for user_scoped detection
_USER_SCOPED_PARAMS = {"user_id", "user", "owner_id", "account_id"}
groups = get_context_groups()
@@ -411,7 +57,6 @@ def generate_edge_manifest(
manifest: dict[str, Any] = {"version": 1, "contexts": {}, "mutations": {}}
for ctx_name, fn_names in groups.items():
# 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] = []
@@ -421,40 +66,31 @@ def generate_edge_manifest(
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())
if input_cls is not None and hasattr(input_cls, "model_fields"):
for param_name in input_cls.model_fields:
param_names.add(param_name)
meta = getattr(fn_cls, "_meta", {})
route = meta.get("route")
view_path = meta.get("view_path")
fn_entry: dict[str, Any] = {
"name": fn_name,
"path": "view" if is_view else "rpc",
"path": "view" if view_path else "rpc",
}
# Collect routes from view-path functions
fn_route = meta.get("route")
if fn_route:
fn_entry["route"] = fn_route
if route:
fn_entry["route"] = route
fn_entry["methods"] = meta.get("methods", ["GET"])
page_routes.append(fn_route)
# Cache protocol metadata
if "rev" in meta:
page_routes.append(route)
if meta.get("rev"):
fn_entry["rev"] = meta["rev"]
if "cache" in meta:
if meta.get("cache") is not None and meta.get("cache") is not True:
fn_entry["cache"] = meta["cache"]
functions_meta.append(fn_entry)
sorted_params = sorted(param_names)
user_scoped = bool(param_names & _USER_SCOPED_PARAMS)
user_scoped = any(p in _USER_SCOPED_PARAMS for p in param_names)
ctx_entry: dict[str, Any] = {
"functions": functions_meta,
@@ -464,69 +100,57 @@ def generate_edge_manifest(
"render_strategy": "dynamic_cached" if user_scoped else "psr",
}
# 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.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:
if not meta.get("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))
affected_contexts = list({a["name"] for a in meta["affects"]})
mutation: dict[str, Any] = {"affects": affected_contexts}
# Determine which params auto-scope
auto_scoped = []
# Auto-scoped params — function params that match context params
input_cls = getattr(fn_cls, "Input", None)
if input_cls and input_cls is not PydanticBaseModel and hasattr(input_cls, "model_fields"):
if input_cls is not None and hasattr(input_cls, "model_fields"):
fn_params = set(input_cls.model_fields.keys())
auto_scoped: list[str] = []
for ctx_name in affected_contexts:
ctx_params = set()
for ctx_fn_name in groups.get(ctx_name, []):
ctx_param_names: set[str] = set()
ctx_fns = groups.get(ctx_name, [])
for ctx_fn_name in ctx_fns:
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))
if ctx_fn_cls is None:
continue
ctx_input = getattr(ctx_fn_cls, "Input", None)
if ctx_input is not None and hasattr(ctx_input, "model_fields"):
ctx_param_names.update(ctx_input.model_fields.keys())
for p in fn_params:
if p in ctx_param_names and p not in auto_scoped:
auto_scoped.append(p)
if auto_scoped:
mutation["auto_scoped_params"] = sorted(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
mutation["private"] = True
if meta.get("route"):
mutation_entry["route"] = meta["route"]
mutation_entry["methods"] = meta.get("methods", ["POST"])
mutation["route"] = meta["route"]
mutation["methods"] = meta.get("methods", ["POST"])
manifest["mutations"][fn_name] = mutation_entry
manifest["mutations"][fn_name] = mutation
return manifest
def generate_edge_manifest_json(
indent: int = 2,
base_url: str = "/api/mizan",
view_urls: dict[str, list[str]] | None = None,
indent: int = 2,
) -> 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)
"""JSON-serialize the Edge manifest."""
return json.dumps(generate_edge_manifest(base_url, view_urls), indent=indent)

28
backends/mizan-django/src/mizan/management/commands/export_mizan_ir.py Normal file
View File

@@ -0,0 +1,28 @@
"""
Mizan IR (KDL) export — Django management command.
Usage:
python manage.py export_mizan_ir
Triggers Mizan client discovery to populate the registry, then writes
the canonical Mizan IR as KDL to stdout. The Rust codegen binary
consumes this directly.
"""
from __future__ import annotations
from django.core.management.base import BaseCommand
from mizan_core.ir import build_ir
class Command(BaseCommand):
help = "Export every registered @client function as Mizan IR (KDL)."
def handle(self, *args, **options) -> None:
# Load every project-side @client function so the registry is
# populated before we emit. Conventionally apps/*/clients.py.
from mizan.setup.discovery import mizan_clients
mizan_clients("apps")
self.stdout.write(build_ir(), ending="")

49
backends/mizan-django/src/mizan/management/commands/export_mizan_schema.py
View File

@@ -1,49 +0,0 @@
"""
Export mizan Schema
Management command to export the mizan OpenAPI schema for TypeScript code generation.
The schema is consumed by openapi-typescript for robust type generation.
Usage:
python manage.py export_mizan_schema # Output to stdout
python manage.py export_mizan_schema --output schema.json # Output to file
"""
import json
from pathlib import Path
from django.core.management.base import BaseCommand
from mizan.export import generate_openapi_schema
class Command(BaseCommand):
help = "Export mizan OpenAPI schema for TypeScript code generation"
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)",
)
def handle(self, *args, **options):
schema = generate_openapi_schema()
indent = options["indent"] if options["indent"] > 0 else None
json_output = json.dumps(schema, indent=indent)
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"Schema written to {output_path}"))
else:
self.stdout.write(json_output)

2
backends/mizan-django/src/mizan/urls.py
View File

@@ -8,7 +8,7 @@ HTTP endpoints:
Security:
- Schema export is NOT exposed over HTTP to prevent API enumeration
- Use the management command instead: python manage.py export_mizan_schema
- Use the management command instead: python manage.py export_mizan_ir
"""
from django.http import JsonResponse

2
backends/mizan-fastapi/src/mizan_fastapi/__init__.py
View File

@@ -35,7 +35,6 @@ from .executor import (
execute_function,
)
from .router import router, mizan_exception_handler, mizan_validation_handler
from .schema import build_schema
__all__ = [
"router",
@@ -43,7 +42,6 @@ __all__ = [
"mizan_validation_handler",
"execute_function",
"compute_invalidation",
"build_schema",
"ErrorCode",
"MizanError",
"NotFound",

45
backends/mizan-fastapi/src/mizan_fastapi/cli.py
View File

@@ -1,45 +0,0 @@
"""
Schema-export CLI for codegen consumption.
Usage:
python -m mizan_fastapi.cli <module>
Imports the named module (whose import side effects must register every
@client function with mizan_core.registry — typically by `@client` plus
`register(...)` calls at module top level), then prints the OpenAPI
schema to stdout as JSON.
Mirrors mizan-django's `manage.py export_mizan_schema` so the codegen
CLI can fetch from either backend the same subprocess way.
"""
from __future__ import annotations
import importlib
import json
import sys
from .schema import build_schema
def main(argv: list[str] | None = None) -> int:
args = list(sys.argv[1:] if argv is None else argv)
if len(args) != 1:
print("usage: python -m mizan_fastapi.cli <module>", file=sys.stderr)
return 2
module_name = args[0]
try:
importlib.import_module(module_name)
except Exception as e:
print(f"failed to import {module_name!r}: {e}", file=sys.stderr)
return 1
schema = build_schema()
json.dump(schema, sys.stdout)
sys.stdout.write("\n")
return 0
if __name__ == "__main__":
sys.exit(main())

39
backends/mizan-fastapi/src/mizan_fastapi/ir.py Normal file
View File

@@ -0,0 +1,39 @@
"""
Mizan IR (KDL) export CLI for FastAPI backends.
Usage:
python -m mizan_fastapi.ir <module>
Imports the named module (whose import side effects must register every
@client function with `mizan_core.registry`), then writes the canonical
Mizan IR as KDL to stdout. The Rust codegen binary consumes this
directly.
"""
from __future__ import annotations
import importlib
import sys
from mizan_core.ir import build_ir
def main(argv: list[str] | None = None) -> int:
args = list(sys.argv[1:] if argv is None else argv)
if len(args) != 1:
print("usage: python -m mizan_fastapi.ir <module>", file=sys.stderr)
return 2
module_name = args[0]
try:
importlib.import_module(module_name)
except Exception as e:
print(f"failed to import {module_name!r}: {e}", file=sys.stderr)
return 1
sys.stdout.write(build_ir())
return 0
if __name__ == "__main__":
sys.exit(main())

235
backends/mizan-fastapi/src/mizan_fastapi/schema.py
View File

@@ -1,235 +0,0 @@
"""
Mizan schema export for FastAPI backends.
Builds an OpenAPI 3.0 document from the registered Mizan functions, mirroring
the shape mizan-django emits via Django Ninja so the codegen consumes either
backend identically.
Usage:
from mizan_fastapi.schema import build_schema
schema = build_schema() # uses globally registered functions
"""
from __future__ import annotations
import re
from typing import Any
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
from pydantic import BaseModel, RootModel, create_model
from mizan_core.registry import get_all_functions, get_context_groups, get_function
from mizan_core.type_utils import extract_list_element, extract_optional
__all__ = ["build_schema", "snake_to_camel"]
# Common user identity param names — mirrors mizan-django's _USER_SCOPED_PARAMS
_USER_SCOPED_PARAMS = {"user_id", "user", "owner_id", "account_id"}
def snake_to_camel(name: str) -> str:
"""Convert snake_case or dotted.name to camelCase. Mirrors mizan-django."""
components = re.split(r"[._]", name)
return components[0] + "".join(c.title() for c in components[1:])
def _has_input(input_cls: Any) -> bool:
return (
input_cls is not None
and input_cls is not BaseModel
and hasattr(input_cls, "model_fields")
and bool(input_cls.model_fields)
)
def _annotation_to_jsonschema_type(annotation: Any) -> str:
if annotation is int:
return "integer"
if annotation is float:
return "number"
if annotation is bool:
return "boolean"
return "string"
def _function_metadata(name: str, fn_class: Any) -> dict[str, Any]:
"""Build one entry of x-mizan-functions. Mirrors Django's shape exactly."""
camel = snake_to_camel(name)
meta = getattr(fn_class, "_meta", {})
input_cls = getattr(fn_class, "Input", None)
has_input = _has_input(input_cls)
output_cls = getattr(fn_class, "Output", None)
_, output_nullable = extract_optional(output_cls) if output_cls is not None else (None, False)
entry: dict[str, Any] = {
"name": name,
"camelName": camel,
"hasInput": has_input,
"inputType": f"{camel}Input" if has_input else None,
"outputType": f"{camel}Output",
# Nullability of the response model — Pydantic `T | None` returns. Carried
# on the function entry rather than the schema class because OpenAPI emits
# `anyOf: [{$ref}, {type:null}]` at the response level, which strict
# deserializers (Rust serde) won't decode as Option<T> without this hint.
"outputNullable": output_nullable,
"transport": "websocket" if meta.get("websocket") else "http",
"isContext": meta.get("context", False),
# Form metadata — always emitted so the schema shape matches Django's,
# even for FastAPI projects that don't use forms (these stay False/None).
"isForm": meta.get("form", False),
"formName": meta.get("form_name"),
"formRole": meta.get("form_role"),
}
if meta.get("affects"):
entry["affects"] = meta["affects"]
if meta.get("merge"):
entry["merge"] = meta["merge"]
return entry
def _context_metadata(context_groups: dict[str, list[str]]) -> dict[str, Any]:
"""Build x-mizan-contexts. Mirrors Django's param-elevation logic."""
out: dict[str, Any] = {}
for ctx_name, fn_names in context_groups.items():
param_info: dict[str, dict[str, Any]] = {}
for fn_name in fn_names:
fn_cls = get_function(fn_name)
if fn_cls is None:
continue
input_cls = getattr(fn_cls, "Input", None)
if not _has_input(input_cls):
continue
for field_name, field_info in input_cls.model_fields.items():
if field_name not in param_info:
param_info[field_name] = {
"type": _annotation_to_jsonschema_type(field_info.annotation),
"sharedBy": [],
}
param_info[field_name]["sharedBy"].append(fn_name)
# A param is required iff every function in the context declares it.
for p_meta in param_info.values():
p_meta["required"] = len(p_meta["sharedBy"]) == len(fn_names)
out[ctx_name] = {
"functions": list(fn_names),
"params": param_info,
}
return out
def build_schema() -> dict[str, Any]:
"""
Build an OpenAPI 3.0 schema for all registered Mizan functions.
Drives FastAPI's native OpenAPI generation by registering a stub endpoint
per function with the function's Input/Output Pydantic models, then
appends the protocol's `x-mizan-functions` and `x-mizan-contexts`
extensions.
Returns a dict in the same shape mizan-django's schema export emits, so
the same codegen pipeline consumes either.
"""
functions = get_all_functions()
context_groups = get_context_groups()
schema_app = FastAPI(
title="mizan Server Functions",
version="1.0.0",
description="Auto-generated schema for mizan server functions",
)
# Per-function endpoints + renamed Pydantic models so component names are
# camelCase + "Input"/"Output" rather than the user's original class names.
schema_classes: dict[str, type[BaseModel]] = {}
function_metadata: list[dict[str, Any]] = []
for name, fn_class in functions.items():
camel = snake_to_camel(name)
input_cls = getattr(fn_class, "Input", None)
output_cls = getattr(fn_class, "Output", None) or BaseModel
has_input = _has_input(input_cls)
input_type_name = f"{camel}Input" if has_input else None
output_type_name = f"{camel}Output"
# Strip Optional so the rename gets a concrete base — nullability is
# carried on the response declaration, not the schema class itself.
output_inner, output_nullable = extract_optional(output_cls)
if has_input:
schema_classes[input_type_name] = create_model(
input_type_name, __base__=input_cls,
)
if extract_list_element(output_inner) is not None:
# list[T] — RootModel makes the rename emit `type: array` rather
# than wrapping the list in a property.
schema_classes[output_type_name] = type(
output_type_name, (RootModel[output_inner],), {},
)
else:
schema_classes[output_type_name] = create_model(
output_type_name, __base__=output_inner,
)
response_model = schema_classes[output_type_name]
if output_nullable:
response_model = response_model | None
# Stub endpoint — only exists so FastAPI walks Pydantic types into
# components.schemas. Never invoked. Annotations are set explicitly
# rather than via closures so forward-ref resolution doesn't trip on
# locally-bound type names.
if has_input:
async def stub(payload):
return None
stub.__annotations__ = {"payload": schema_classes[input_type_name]}
else:
async def stub():
return None
schema_app.post(
f"/mizan/{name}",
response_model=response_model,
operation_id=camel,
summary=fn_class.__doc__ or f"Call {name}",
)(stub)
function_metadata.append(_function_metadata(name, fn_class))
schema = get_openapi(
title=schema_app.title,
version=schema_app.version,
description=schema_app.description,
routes=schema_app.routes,
)
schema["x-mizan-functions"] = function_metadata
if context_groups:
schema["x-mizan-contexts"] = _context_metadata(context_groups)
# Attach x-mizan operation metadata, mirroring Django.
paths = schema.get("paths", {})
for fn_meta in function_metadata:
op = paths.get(f"/mizan/{fn_meta['name']}", {}).get("post")
if op is not None:
op["x-mizan"] = {
"transport": fn_meta["transport"],
"isContext": fn_meta["isContext"],
}
return schema