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

Split the registry — function/composition core + backend extensions

Browse Source 
The original registry tangled function, channel, composition, and form
registration in a single file with polymorphic register() dispatch.
That predates the household discipline; it was the design that was
supposed to ship but didn't. Re-implementing the original intent.

cores/mizan-python/src/mizan_core/registry.py (new):
- _functions, _compositions dicts
- register() — ServerFunction-only, no polymorphic dispatch
- register_as(), register_compose()
- register_extension(name, extension) — hook interface
- get_function/get_compose/get_all_functions/get_all_compositions
- get_contexts, get_context_groups
- get_registry, get_schema — aggregate extension contributions
- validate_registry, clear_registry — cascade-clear extensions

RegistryExtension Protocol:
- schema() returns the extension's schema subdict (keyed under its name)
- clear() resets extension state (called by clear_registry)

mizan-django/src/mizan/channels/__init__.py:
- _ChannelsExtension wraps the channel _registry, plugs into core via
  register_extension('channels', ...). Schema output preserves the
  same shape codegen consumed before (snake_case keys, type+bidirectional).

mizan-django/src/mizan/forms/__init__.py:
- register_form() and get_forms() helpers moved here (were in setup/registry.py)
- Both use mizan_core.registry under the hood. Forms don't need a
  separate extension because form sub-functions register as regular
  ServerFunctions with meta.form set.

mizan-django/src/mizan/setup/registry.py: deleted.
mizan-django/src/mizan/setup/__init__.py: re-exports the registry helpers
from mizan_core.registry / mizan.channels / mizan.forms — the Django
adapter's curated public API surface stays stable for users.

Consumers updated: ~10 files imported `from mizan.setup.registry`;
all switched to direct imports from mizan_core.registry, mizan.channels,
or mizan.forms as appropriate. ChannelTests in test_core.py rewritten
to use mizan.channels.register directly (no more polymorphic
@register_as on ReactChannel subclasses).

Verified:
- mizan-core: 15/15
- mizan-django: 348 pass, 21 skip, 0 fail
- mizan-ts edge-compat: 34/34 (cross-language pin holds)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ryth Azhur
2026-05-06 15:21:16 -04:00
parent 9150cdc5ee
commit 76fce2dc85
17 changed files with 363 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backends/mizan-django/README.md
View File

@@ -26,7 +26,7 @@ application = wrap_asgi(get_asgi_application())
```python
from mizan.client import client
from mizan.setup.registry import register
from mizan_core.registry import register
from pydantic import BaseModel
class Output(BaseModel):

41
backends/mizan-django/src/mizan/channels/__init__.py
View File

@@ -523,6 +523,47 @@ def __getattr__(name):
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
# =============================================================================
# Core Registry Extension
# =============================================================================
class _ChannelsExtension:
"""
Plugs the channel registry into mizan_core.registry as the 'channels'
extension. Schema output goes under schema['channels'] in the unified
registry export consumed by codegen.
"""
def all(self) -> dict:
return dict(_registry)
def schema(self) -> dict:
out: dict[str, Any] = {}
for name, channel_class in _registry.items():
channel_schema: dict[str, Any] = {
"name": name,
"type": "channel",
"bidirectional": False,
}
if getattr(channel_class, "Params", None):
channel_schema["params"] = channel_class.Params.model_json_schema()
if getattr(channel_class, "ReactMessage", None):
channel_schema["react_message"] = channel_class.ReactMessage.model_json_schema()
channel_schema["bidirectional"] = True
if getattr(channel_class, "DjangoMessage", None):
channel_schema["django_message"] = channel_class.DjangoMessage.model_json_schema()
out[name] = channel_schema
return out
def clear(self) -> None:
_registry.clear()
from mizan_core.registry import register_extension as _register_extension
_register_extension("channels", _ChannelsExtension())
# =============================================================================
# Exports
# =============================================================================

2
backends/mizan-django/src/mizan/channels/connection.py
View File

@@ -400,7 +400,7 @@ class DjangoReactConsumer(AsyncJsonWebsocketConsumer):
- User context from WebSocket session is passed to function
"""
from mizan.client.executor import execute_function, FunctionError
from mizan.setup.registry import get_function
from mizan_core.registry import get_function
request_id = content.get("id")
fn_name = content.get("fn")

2
backends/mizan-django/src/mizan/client/executor.py
View File

@@ -28,7 +28,7 @@ from django.views.decorators.csrf import csrf_protect
from pydantic import BaseModel, ValidationError
from mizan.cache import get_cache, cache_get, cache_put, cache_purge
from mizan.setup.registry import get_function, get_context_groups
from mizan_core.registry import get_function, get_context_groups
from mizan.setup.settings import get_settings
if TYPE_CHECKING:

2
backends/mizan-django/src/mizan/client/function.py
View File

@@ -665,7 +665,7 @@ def compose(
"""
def decorator(fn: Callable) -> ComposedContext:
from mizan.setup.registry import register_compose
from mizan_core.registry import register_compose
name = fn.__name__

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

@@ -26,7 +26,7 @@ if TYPE_CHECKING:
from django import forms
from ninja import NinjaAPI
from mizan.setup.registry import get_registry, get_schema, get_context_groups, get_function
from mizan_core.registry import get_registry, get_schema, get_context_groups, get_function
__all__ = [

49
backends/mizan-django/src/mizan/forms/__init__.py
View File

@@ -292,7 +292,7 @@ def _register_form_as_server_functions(form_class: type) -> None:
from .schemas import FormSchema, FormSubmitFail, FormSubmitPass, FormValidation
from .schema_utils import build_form_schema
from .validation_utils import validate_form_instance
from mizan.setup.registry import register
from mizan_core.registry import register
from mizan.client.function import ServerFunction
config: mizanFormMeta = form_class.mizan
@@ -484,7 +484,7 @@ def _register_formset_functions(
from .schema_utils import build_form_schema
from .validation_utils import build_formset_validation
from .formset_utils import forms_to_formset_post_data
from mizan.setup.registry import register
from mizan_core.registry import register
from mizan.client.function import ServerFunction
formset_class = formset_factory(form_class)
@@ -630,3 +630,48 @@ def _register_formset_functions(
FormsetSubmitFunction.__name__ = f"{form_name}_formset_submit"
FormsetSubmitFunction.Output = FormsetSubmitPass
register(FormsetSubmitFunction, f"{form_name}.formset.submit")
# ─── Public helpers ─────────────────────────────────────────────────────────
def register_form(
form_class: type,
name: str,
submit_handler: Any = None,
) -> None:
"""
Register a Django Form class as Mizan server functions.
Creates and registers `{name}.schema`, `{name}.validate`, and
`{name}.submit` (if a submit_handler is provided).
"""
from mizan.client.function import create_form_functions
from mizan_core.registry import register
schema_fn, validate_fn, submit_fn = create_form_functions(
form_class, name, submit_handler
)
register(schema_fn, f"{name}.schema")
register(validate_fn, f"{name}.validate")
if submit_fn:
register(submit_fn, f"{name}.submit")
def get_forms() -> dict[str, list]:
"""
Group registered form-related functions by their form name.
Returns a mapping like:
{"contact": [ContactSchema, ContactValidate, ContactSubmit], ...}
"""
from mizan_core.registry import get_all_functions
forms: dict[str, list] = {}
for name, cls in get_all_functions().items():
meta = getattr(cls, "_meta", {})
if not meta.get("form"):
continue
form_name = meta.get("form_name")
forms.setdefault(form_name, []).append(cls)
return forms

33
backends/mizan-django/src/mizan/setup/__init__.py
View File

@@ -1,36 +1,40 @@
"""
mizan.setup - Integration and registration utilities.
mizan.setup - Django integration helpers.
This subpackage contains everything developers need to integrate mizan:
- Registry for server functions and channels
- Auto-discovery for apps
- Configuration settings
Usage:
from mizan.setup import mizan_clients, register, get_function
The function/composition registry now lives in `mizan_core.registry`.
Channels register themselves through the channel-specific registry in
`mizan.channels`. Forms register through `mizan.forms`. This module
re-exports the helpers that Django mizan users typically reach for, so
`from mizan.setup import register, get_function, mizan_clients, …` keeps
working as a single curated surface.
"""
from .registry import (
from mizan_core.registry import (
register,
register_as,
register_form,
register_compose,
get_function,
get_channel,
get_compose,
get_view,
get_all_functions,
get_all_channels,
get_all_compositions,
get_registry,
get_schema,
get_contexts,
get_context_groups,
get_forms,
validate_registry,
clear_registry,
)
from mizan.channels import (
get_channel,
get_registered_channels as get_all_channels,
)
from mizan.forms import (
register_form,
get_forms,
)
from .discovery import (
mizan_clients,
mizan_module,
@@ -52,7 +56,6 @@ __all__ = [
"get_function",
"get_channel",
"get_compose",
"get_view",
"get_all_functions",
"get_all_channels",
"get_all_compositions",

2
backends/mizan-django/src/mizan/setup/discovery.py
View File

@@ -18,7 +18,7 @@ from typing import Any
from mizan._vendor.app_visitor import DjangoAppVisitor, get_members
from .registry import register, get_function
from mizan_core.registry import register, get_function
from mizan.client.function import ServerFunction

373
backends/mizan-django/src/mizan/setup/registry.py
View File

@@ -1,373 +0,0 @@
"""
mizan Registry
Central registration for server functions, channels, and compositions.
All items are identified by name.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Callable
if TYPE_CHECKING:
from mizan.client.function import ServerFunction, ComposedContext
from mizan.channels import ReactChannel
# Global registries - all use name as key
_functions: dict[str, type["ServerFunction"]] = {}
_channels: dict[str, type["ReactChannel"]] = {}
_compositions: dict[str, "ComposedContext"] = {}
def register(
view_class: type["ServerFunction"] | type["ReactChannel"],
name: str,
) -> type["ServerFunction"] | type["ReactChannel"]:
"""
Register a server function or channel.
Args:
view_class: ServerFunction or ReactChannel subclass
name: Registration name (used for API calls and code generation)
Returns:
The view class (allows use as part of decorator chain)
"""
from mizan.client.function import ServerFunction
from mizan.channels import ReactChannel
view_class.name = name
if issubclass(view_class, ReactChannel):
if name in _channels:
# Allow re-registration of the same class (idempotent for reloads)
if _channels[name] is not view_class:
raise ValueError(
f"Channel '{name}' already registered by {_channels[name].__name__}"
)
return view_class
_channels[name] = view_class
elif issubclass(view_class, ServerFunction):
if name in _functions:
# Allow re-registration of the same class (idempotent for reloads)
existing = _functions[name]
if existing.__name__ == view_class.__name__:
# Same function being re-registered (reload scenario)
_functions[name] = view_class
return view_class
raise ValueError(
f"Function '{name}' already registered by {existing.__name__}"
)
_functions[name] = view_class
else:
raise TypeError(f"{view_class} must be a ServerFunction or ReactChannel")
return view_class
def register_as(name: str):
"""
Decorator for registering a server function or channel.
Usage:
@register_as('update-profile')
class UpdateProfile(ServerFunction):
...
"""
def decorator(view_class):
return register(view_class, name)
return decorator
def register_form(
form_class: type,
name: str,
submit_handler: Callable | None = None,
) -> None:
"""
Register a Django Form as server functions.
Creates and registers:
- {name}.schema: Returns form field definitions
- {name}.validate: Validates form data
- {name}.submit: Submits form (if submit_handler provided)
Usage:
register_form(ContactForm, 'contact', submit_handler=handle_contact)
"""
from mizan.client.function import create_form_functions
schema_fn, validate_fn, submit_fn = create_form_functions(
form_class, name, submit_handler
)
register(schema_fn, f"{name}.schema")
register(validate_fn, f"{name}.validate")
if submit_fn:
register(submit_fn, f"{name}.submit")
def register_compose(
composed: "ComposedContext",
name: str,
) -> "ComposedContext":
"""
Register a composed context.
Args:
composed: ComposedContext instance
name: Registration name
Returns:
The composed context
"""
if name in _compositions:
existing = _compositions[name]
if existing.name == composed.name:
# Same composition being re-registered (reload scenario)
_compositions[name] = composed
return composed
raise ValueError(f"Composition '{name}' already registered by {existing.name}")
_compositions[name] = composed
return composed
def get_function(name: str) -> type["ServerFunction"] | None:
"""Get a registered server function by name."""
return _functions.get(name)
def get_channel(name: str) -> type["ReactChannel"] | None:
"""Get a registered channel by name."""
return _channels.get(name)
def get_compose(name: str) -> "ComposedContext | None":
"""Get a registered composition by name."""
return _compositions.get(name)
def get_view(name: str) -> type["ServerFunction"] | type["ReactChannel"] | None:
"""Get any registered view by name (function or channel)."""
return _functions.get(name) or _channels.get(name)
def get_all_functions() -> dict[str, type["ServerFunction"]]:
"""Get all registered functions."""
return _functions.copy()
def get_all_channels() -> dict[str, type["ReactChannel"]]:
"""Get all registered channels."""
return _channels.copy()
def get_all_compositions() -> dict[str, "ComposedContext"]:
"""Get all registered compositions."""
return _compositions.copy()
def get_registry() -> dict[str, dict[str, Any]]:
"""
Get the full registry organized by type.
Returns:
{
"functions": { name: class, ... },
"channels": { name: class, ... },
"compositions": { name: ComposedContext, ... },
}
"""
return {
"functions": _functions.copy(),
"channels": _channels.copy(),
"compositions": _compositions.copy(),
}
def get_schema() -> dict[str, Any]:
"""
Export the full schema for TypeScript generation.
Returns:
{
"functions": {
"update_profile": {
"name": "update_profile",
"type": "function",
"meta": { "context": "global", ... },
"input": { ... },
"output": { ... },
},
...
},
"channels": {
"chat": {
"name": "chat",
"type": "channel",
"params": { ... },
"django_message": { ... },
...
},
...
},
"compositions": {
"user_page": {
"name": "user_page",
"type": "compose",
"meta": { "on_server": false, ... },
"children": ["user_profile", "user_posts"],
"leaves": ["user_profile", "user_posts"],
},
...
},
}
"""
functions = {}
for name, cls in _functions.items():
schema = cls.get_schema_export()
functions[name] = schema
compositions = {}
for name, composed in _compositions.items():
compositions[name] = {
"name": composed.name,
"type": "compose",
"meta": composed._meta,
"children": composed._meta.get("children", []),
"leaves": composed._meta.get("leaves", []),
}
# Build channel schemas from our registry
# Only include keys when they have values (test expects absent keys, not None)
channels_schema = {}
for name, channel_class in _channels.items():
channel_schema: dict[str, Any] = {
"name": name,
"type": "channel",
"bidirectional": False,
}
# Extract Params schema (only if defined)
if hasattr(channel_class, "Params") and channel_class.Params:
channel_schema["params"] = channel_class.Params.model_json_schema()
# Extract ReactMessage schema (only if defined - indicates bidirectional)
if hasattr(channel_class, "ReactMessage") and channel_class.ReactMessage:
channel_schema[
"react_message"
] = channel_class.ReactMessage.model_json_schema()
channel_schema["bidirectional"] = True
# Extract DjangoMessage schema (only if defined)
if hasattr(channel_class, "DjangoMessage") and channel_class.DjangoMessage:
channel_schema[
"django_message"
] = channel_class.DjangoMessage.model_json_schema()
channels_schema[name] = channel_schema
return {
"functions": functions,
"channels": channels_schema,
"compositions": compositions,
}
def get_contexts() -> dict[str, type["ServerFunction"]]:
"""
Get all server functions marked as contexts.
These are functions with meta.context = True, used for SSR hydration.
"""
contexts = {}
for name, cls in _functions.items():
if getattr(cls, "_meta", {}).get("context"):
contexts[name] = cls
return contexts
def get_context_groups() -> dict[str, list[str]]:
"""
Group function names by their context string.
Returns:
{"global": ["current_user"], "user": ["user_profile", "user_orders"]}
"""
groups: dict[str, list[str]] = {}
for name, cls in _functions.items():
ctx = getattr(cls, "_meta", {}).get("context")
if ctx:
groups.setdefault(ctx, []).append(name)
return groups
def get_forms() -> dict[str, list[type["ServerFunction"]]]:
"""
Get all server functions that are form-related, grouped by form name.
Returns:
{
"contact": [ContactSchema, ContactValidate, ContactSubmit],
...
}
"""
forms: dict[str, list] = {}
for name, cls in _functions.items():
meta = getattr(cls, "_meta", {})
if meta.get("form"):
form_name = meta.get("form_name")
if form_name not in forms:
forms[form_name] = []
forms[form_name].append(cls)
return forms
def validate_registry() -> list[str]:
"""
Validate that all affects targets resolve to known contexts or functions.
Called automatically after discovery. Emits warnings for unresolved targets
(e.g., typos in string-based affects declarations).
Returns a list of warning messages (empty if everything resolves).
"""
import warnings
issues: list[str] = []
groups = get_context_groups()
all_fn_names = set(_functions.keys())
for fn_name, fn_cls in _functions.items():
meta = getattr(fn_cls, "_meta", {})
affects = meta.get("affects")
if not affects:
continue
for target in affects:
target_name = target.get("name", "")
target_type = target.get("type", "")
if target_type == "context" and target_name not in groups:
issues.append(
f"@client function '{fn_name}' declares affects='{target_name}', "
f"but no context named '{target_name}' is registered."
)
elif target_type == "function" and target_name not in all_fn_names:
issues.append(
f"@client function '{fn_name}' targets function '{target_name}', "
f"but no function named '{target_name}' is registered."
)
for msg in issues:
warnings.warn(msg, stacklevel=2)
return issues
def clear_registry() -> None:
"""Clear all registrations. Primarily for testing."""
_functions.clear()
_channels.clear()
_compositions.clear()

2
backends/mizan-django/src/mizan/tests/test_auth.py
View File

@@ -32,7 +32,7 @@ from mizan.client.executor import (
ErrorCode,
)
from mizan.client import client
from mizan.setup.registry import clear_registry, register
from mizan_core.registry import clear_registry, register
from pydantic import BaseModel

4
backends/mizan-django/src/mizan/tests/test_benchmarks.py
View File

@@ -27,7 +27,7 @@ from django.test import RequestFactory, TestCase, TransactionTestCase, override_
from pydantic import BaseModel
from mizan.client.executor import FunctionResult, execute_function, function_call_view
from mizan.setup.registry import clear_registry
from mizan_core.registry import clear_registry
from mizan.client import client
@@ -66,7 +66,7 @@ class StatsOutput(BaseModel):
def setup_benchmark_functions():
"""Register benchmark server functions."""
from mizan.setup.registry import register
from mizan_core.registry import register
clear_registry()

6
backends/mizan-django/src/mizan/tests/test_channels.py
View File

@@ -928,13 +928,13 @@ class WebSocketRPCTests(TestCase):
def setUp(self):
# Clear mizan registry
from mizan.setup.registry import clear_registry
from mizan_core.registry import clear_registry
clear_registry()
# Register test functions
from mizan.client import client
from mizan.setup.registry import register
from mizan_core.registry import register
from pydantic import BaseModel
class EchoOutput(BaseModel):
@@ -964,7 +964,7 @@ class WebSocketRPCTests(TestCase):
register(rpc_auth_required, "rpc_auth_required")
def tearDown(self):
from mizan.setup.registry import clear_registry
from mizan_core.registry import clear_registry
clear_registry()

23
backends/mizan-django/src/mizan/tests/test_core.py
View File

@@ -17,15 +17,15 @@ from mizan.client.executor import (
execute_function,
execute_context,
)
from mizan.setup.registry import (
from mizan_core.registry import (
clear_registry,
register,
register_as,
register_form,
get_schema,
get_contexts,
get_function,
)
from mizan.forms import register_form
from mizan.client import ServerFunction, client, ReactContext, GlobalContext
from mizan.channels import ReactChannel
@@ -597,7 +597,7 @@ class ContextTests(TestCase):
def test_context_groups(self):
"""Test get_context_groups() groups functions by context name."""
from mizan.setup.registry import get_context_groups
from mizan_core.registry import get_context_groups
UserCtx = ReactContext("user")
@@ -1147,8 +1147,8 @@ class ChannelTests(TestCase):
def test_register_channel(self):
"""Test channel registration."""
from mizan.channels import register as register_channel, get_channel
@register_as("test-channel")
class TestChannel(ReactChannel):
class DjangoMessage(BaseModel):
text: str
@@ -1156,15 +1156,13 @@ class ChannelTests(TestCase):
def authorize(self, params=None):
return True
from mizan.setup.registry import get_channel
channel = get_channel("test-channel")
self.assertEqual(channel, TestChannel)
register_channel(TestChannel, "test-channel")
self.assertEqual(get_channel("test-channel"), TestChannel)
def test_channel_schema_export(self):
"""Test channel schema export."""
from mizan.channels import register as register_channel
@register_as("chat")
class ChatChannel(ReactChannel):
class Params(BaseModel):
room: int
@@ -1179,6 +1177,8 @@ class ChannelTests(TestCase):
def authorize(self, params):
return True
register_channel(ChatChannel, "chat")
schema = get_schema()
self.assertIn("channels", schema)
@@ -1193,8 +1193,8 @@ class ChannelTests(TestCase):
def test_server_push_only_channel(self):
"""Test channel without ReactMessage (server-push only)."""
from mizan.channels import register as register_channel
@register_as("notifications")
class NotificationsChannel(ReactChannel):
class DjangoMessage(BaseModel):
title: str
@@ -1202,6 +1202,7 @@ class ChannelTests(TestCase):
def authorize(self, params=None):
return True
register_channel(NotificationsChannel, "notifications")
schema = get_schema()
notif_schema = schema["channels"]["notifications"]
@@ -2059,7 +2060,7 @@ class ReturnTypeBranchingTests(TestCase):
register(profile_page, "profile_page")
# It's in the context groups (for invalidation graph)
from mizan.setup.registry import get_context_groups
from mizan_core.registry import get_context_groups
groups = get_context_groups()
self.assertIn("user", groups)
self.assertIn("profile_page", groups["user"])

4
backends/mizan-django/src/mizan/tests/test_pentest.py
View File

@@ -42,7 +42,7 @@ from mizan.client.executor import (
FunctionResult,
execute_function,
)
from mizan.setup.registry import clear_registry, get_function, register
from mizan_core.registry import clear_registry, get_function, register
from mizan.client import ServerFunction, client
@@ -1179,7 +1179,7 @@ class RegistrationSecurityTests(TestCase):
But a DIFFERENT function cannot take over an existing name.
"""
from mizan.client import ServerFunction
from mizan.setup.registry import register
from mizan_core.registry import register
# Register first function
class OriginalFunc(ServerFunction):

2
backends/mizan-django/src/mizan/tests/test_security.py
View File

@@ -29,7 +29,7 @@ from mizan.client.executor import (
execute_function,
function_call_view,
)
from mizan.setup.registry import clear_registry, register, register_as, get_function
from mizan_core.registry import clear_registry, register, register_as, get_function
from mizan.client import ServerFunction, client
from mizan.channels import ReactChannel

230
cores/mizan-python/src/mizan_core/registry.py Normal file
View File

@@ -0,0 +1,230 @@
"""
Mizan core registry — function and composition registration with an
extension hook for backend-specific registries (channels, forms, etc.)
to plug into.
This is the framework-agnostic registry. Backends own their own
type-specific registries (channels in Django Channels, forms in Django
Forms, websockets in FastAPI, etc.) and register them as extensions
here so the unified schema export can include them.
"""
from __future__ import annotations
from typing import Any, Callable, Protocol
# ─── Core registries ────────────────────────────────────────────────────────
_functions: dict[str, Any] = {}
_compositions: dict[str, Any] = {}
# ─── Extension hook ─────────────────────────────────────────────────────────
class RegistryExtension(Protocol):
"""
Backend-specific registries plug into core via this Protocol.
Each extension owns its own registry of backend-shaped registrations
(channels, forms, websocket consumers, etc.) and contributes a schema
subdict to the unified schema export.
"""
def schema(self) -> dict[str, Any]: ...
def clear(self) -> None: ...
_extensions: dict[str, RegistryExtension] = {}
def register_extension(name: str, extension: RegistryExtension) -> None:
"""
Register a backend extension. The extension contributes to schema
output under its name (e.g. 'channels', 'forms').
"""
_extensions[name] = extension
# ─── Function registration ──────────────────────────────────────────────────
def register(view_class: Any, name: str) -> Any:
"""
Register a server function class. Used by the @client decorator.
Idempotent for the same class (supports module reloads).
"""
view_class.name = name
if name in _functions:
existing = _functions[name]
if existing.__name__ == view_class.__name__:
_functions[name] = view_class
return view_class
raise ValueError(
f"Function '{name}' already registered by {existing.__name__}"
)
_functions[name] = view_class
return view_class
def register_as(name: str) -> Callable[[Any], Any]:
"""Decorator form of register()."""
def decorator(view_class: Any) -> Any:
return register(view_class, name)
return decorator
def register_compose(composed: Any, name: str) -> Any:
"""Register a composed context."""
if name in _compositions:
existing = _compositions[name]
if existing.name == composed.name:
_compositions[name] = composed
return composed
raise ValueError(
f"Composition '{name}' already registered by {existing.name}"
)
_compositions[name] = composed
return composed
# ─── Lookups ────────────────────────────────────────────────────────────────
def get_function(name: str) -> Any | None:
"""Get a registered server function by name."""
return _functions.get(name)
def get_compose(name: str) -> Any | None:
"""Get a registered composition by name."""
return _compositions.get(name)
def get_all_functions() -> dict[str, Any]:
"""Get all registered functions."""
return _functions.copy()
def get_all_compositions() -> dict[str, Any]:
"""Get all registered compositions."""
return _compositions.copy()
def get_contexts() -> dict[str, Any]:
"""Get all server functions marked as contexts (meta.context truthy)."""
return {
name: cls
for name, cls in _functions.items()
if getattr(cls, "_meta", {}).get("context")
}
def get_context_groups() -> dict[str, list[str]]:
"""Group function names by their context string."""
groups: dict[str, list[str]] = {}
for name, cls in _functions.items():
ctx = getattr(cls, "_meta", {}).get("context")
if ctx:
groups.setdefault(ctx, []).append(name)
return groups
def get_registry() -> dict[str, Any]:
"""
Full registry organized by type, including extension contributions.
Returns:
{
"functions": {...},
"compositions": {...},
"<extension>": {...}, # one per registered extension
}
"""
out: dict[str, Any] = {
"functions": _functions.copy(),
"compositions": _compositions.copy(),
}
for name, ext in _extensions.items():
# Extensions optionally expose their backing dict via .all()
# (Protocol doesn't require it; only schema() and clear() are mandatory)
if hasattr(ext, "all"):
out[name] = ext.all()
return out
def get_schema() -> dict[str, Any]:
"""
Export the unified schema for codegen consumption.
Aggregates function and composition schemas plus contributions from
each registered backend extension under its name.
"""
schema: dict[str, Any] = {
"functions": {
name: cls.get_schema_export() for name, cls in _functions.items()
},
"compositions": {
name: {
"name": composed.name,
"type": "compose",
"meta": composed._meta,
"children": composed._meta.get("children", []),
"leaves": composed._meta.get("leaves", []),
}
for name, composed in _compositions.items()
},
}
for name, ext in _extensions.items():
schema[name] = ext.schema()
return schema
# ─── Validation ─────────────────────────────────────────────────────────────
def validate_registry() -> list[str]:
"""
Check that all `affects` targets resolve to known contexts or functions.
Emits warnings for unresolved targets (e.g. typos in string-based affects).
Returns the list of warning messages (empty if all resolve).
"""
import warnings
issues: list[str] = []
groups = get_context_groups()
all_fn_names = set(_functions.keys())
for fn_name, fn_cls in _functions.items():
meta = getattr(fn_cls, "_meta", {})
affects = meta.get("affects")
if not affects:
continue
for target in affects:
target_name = target.get("name", "")
target_type = target.get("type", "")
if target_type == "context" and target_name not in groups:
issues.append(
f"@client function '{fn_name}' declares affects='{target_name}', "
f"but no context named '{target_name}' is registered."
)
elif target_type == "function" and target_name not in all_fn_names:
issues.append(
f"@client function '{fn_name}' targets function '{target_name}', "
f"but no function named '{target_name}' is registered."
)
for msg in issues:
warnings.warn(msg, stacklevel=2)
return issues
# ─── Clear (testing) ────────────────────────────────────────────────────────
def clear_registry() -> None:
"""Clear all registrations, including extension state. For testing."""
_functions.clear()
_compositions.clear()
for ext in _extensions.values():
ext.clear()