Python API Reference

This page is generated from docstrings of the public API exported by autocrud.

autocrud

autocrud

Attributes

__all__ module-attribute

__all__ = [
    "AutoCRUD",
    "BackgroundTaskAccepted",
    "BlobUploadSession",
    "DisplayName",
    "DuplicateResourceError",
    "IConstraintChecker",
    "IValidator",
    "JobRedirectInfo",
    "LoadStats",
    "MissingOperationContextError",
    "OnDelete",
    "OnDuplicate",
    "Ref",
    "RefRevision",
    "ResourceOps",
    "RefType",
    "RevisionNotMigratedError",
    "Schema",
    "SearchedResource",
    "Unique",
    "UniqueConstraintError",
    "ValidationError",
    "crud",
    "struct_to_pydantic",
]

__version__ module-attribute

__version__ = '0.8.3a13'

Classes

AutoCRUD

High-level entry point for registering resource models and generating CRUD routes.

AutoCRUD manages a set of per-resource ResourceManagers and applies a set of route templates to a FastAPI APIRouter (or FastAPI app) to generate endpoints.

Typical setup:

from fastapi import FastAPI
from autocrud import crud  # global instance

app = FastAPI()

# configure once at startup (optional)
crud.configure(model_naming="kebab")

# register models/schemas
crud.add_model(User)
crud.add_model(Post)

# generate routes
crud.apply(app)

Notes: - Call configure() / add_model() during application startup, before serving requests. - apply() installs route templates, custom create/update actions, ref routes, and backup routes. - openapi() customizes OpenAPI schema to include AutoCRUD-specific schemas and extensions.

PARAMETER	DESCRIPTION
model_naming	How model names are converted to resource names (URL paths). Either one of: "same", "pascal", "camel", "snake", "kebab", or a callable (type) -> str. TYPE: Literal['same', 'pascal', 'camel', 'snake', 'kebab'] | Callable[[type], str] DEFAULT: 'kebab'
route_templates	Route templates to apply. When None or a dict, default templates are used and can be configured via {TemplateClass: kwargs}. TYPE: list[IRouteTemplate] | dict[type, dict[str, Any]] | None DEFAULT: None
storage_factory	Default storage factory for models that don't specify storage. TYPE: IStorageFactory | None DEFAULT: None
message_queue_factory	Default message queue factory used for Job models (when enabled). TYPE: IMessageQueueFactory | None DEFAULT: None
admin	If provided and permission_checker is not set, enables RBAC with admin as root user. TYPE: str | None DEFAULT: None
permission_checker	Permission checker used by default for models that don't override it. TYPE: IPermissionChecker | None DEFAULT: None
dependency_provider	Dependency injection provider passed to route templates (when using defaults). TYPE: DependencyProvider | None DEFAULT: None
event_handlers	Global event handlers used by default for models that don't override it. TYPE: Sequence[IEventHandler] | None DEFAULT: None
encoding	Default encoding for stored payloads (e.g. json/msgpack). TYPE: Encoding DEFAULT: json
default_user	Default user (or factory) used when user is not specified. When set, the DependencyProvider's default get_user returns this value instead of "anonymous". A custom get_user on the provider always takes priority. TYPE: str | Callable[[], str] | UnsetType DEFAULT: UNSET
default_now	Default timestamp function used when time is not specified. TYPE: Callable[[], datetime] | UnsetType DEFAULT: UNSET
strict_operation_context	When True, all write operations (create, update, delete, etc.) will raise :class:MissingOperationContextError if required context fields (user, now) are not fully resolved from any source (explicit kwargs, using() scope, or manager defaults). Defaults to False. TYPE: bool DEFAULT: False

See also

• Schema: declare schema/validation/migration for a resource.
• Ref, RefRevision: reference types used across APIs and OpenAPI schema.
• dump(), load(): export/import utilities for backups.
• Routes: docs/howto/routes.md
• Behavior & lifecycle: docs/reference/behavior.md
• Performance notes: docs/guides/performance.md

Source code in autocrud/crud/core.py

class AutoCRUD:
    """High-level entry point for registering resource models and generating CRUD routes.

    AutoCRUD manages a set of per-resource `ResourceManager`s and applies a set of
    route templates to a FastAPI `APIRouter` (or `FastAPI` app) to generate endpoints.

    Typical setup:

    ```python
    from fastapi import FastAPI
    from autocrud import crud  # global instance

    app = FastAPI()

    # configure once at startup (optional)
    crud.configure(model_naming="kebab")

    # register models/schemas
    crud.add_model(User)
    crud.add_model(Post)

    # generate routes
    crud.apply(app)
    ```

    Notes:
    - Call `configure()` / `add_model()` during application startup, before serving requests.
    - `apply()` installs route templates, custom create/update actions, ref routes, and backup routes.
    - `openapi()` customizes OpenAPI schema to include AutoCRUD-specific schemas and extensions.

    Args:
        model_naming:
            How model names are converted to resource names (URL paths). Either one of:
            `"same"`, `"pascal"`, `"camel"`, `"snake"`, `"kebab"`, or a callable `(type) -> str`.
        route_templates:
            Route templates to apply. When `None` or a `dict`, default templates are used and
            can be configured via `{TemplateClass: kwargs}`.
        storage_factory:
            Default storage factory for models that don't specify `storage`.
        message_queue_factory:
            Default message queue factory used for Job models (when enabled).
        admin:
            If provided and `permission_checker` is not set, enables RBAC with `admin` as root user.
        permission_checker:
            Permission checker used by default for models that don't override it.
        dependency_provider:
            Dependency injection provider passed to route templates (when using defaults).
        event_handlers:
            Global event handlers used by default for models that don't override it.
        encoding:
            Default encoding for stored payloads (e.g. json/msgpack).
        default_user:
            Default user (or factory) used when user is not specified.
            When set, the ``DependencyProvider``'s default ``get_user``
            returns this value instead of ``"anonymous"``.  A custom
            ``get_user`` on the provider always takes priority.
        default_now:
            Default timestamp function used when time is not specified.
        strict_operation_context:
            When ``True``, all write operations (create, update, delete, etc.)
            will raise :class:`MissingOperationContextError` if required
            context fields (``user``, ``now``) are not fully resolved from
            any source (explicit kwargs, ``using()`` scope, or manager
            defaults).  Defaults to ``False``.

    See also:
        - `Schema`: declare schema/validation/migration for a resource.
        - `Ref`, `RefRevision`: reference types used across APIs and OpenAPI schema.
        - `dump()`, `load()`: export/import utilities for backups.
        - Routes: docs/howto/routes.md
        - Behavior & lifecycle: docs/reference/behavior.md
        - Performance notes: docs/guides/performance.md
    """

    def __init__(
        self,
        *,
        model_naming: Literal["same", "pascal", "camel", "snake", "kebab"]
        | Callable[[type], str] = "kebab",
        route_templates: list[IRouteTemplate]
        | dict[type, dict[str, Any]]
        | None = None,
        storage_factory: IStorageFactory | None = None,
        message_queue_factory: IMessageQueueFactory | None = None,
        admin: str | None = None,
        permission_checker: IPermissionChecker | None = None,
        dependency_provider: DependencyProvider | None = None,
        event_handlers: Sequence[IEventHandler] | None = None,
        encoding: Encoding = Encoding.json,
        default_user: str | Callable[[], str] | UnsetType = UNSET,
        default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
        strict_operation_context: bool = False,
    ):
        # Initialize empty collections
        self.resource_managers: OrderedDict[str, IResourceManager] = OrderedDict()
        self.message_queues: OrderedDict[str, IMessageQueue] = OrderedDict()
        self.model_names: dict[type[T], str | None] = {}
        self.relationships: list[_RefInfo] = []

        # Initialize attributes with defaults before applying configuration
        self.storage_factory = MemoryStorageFactory()
        self.blob_store = MemoryBlobStore()
        self.model_naming = "kebab"
        self.message_queue_factory = None
        self.route_templates: list[IRouteTemplate] = []
        self.permission_checker = AllowAll()
        self.event_handlers = None
        self.default_encoding = Encoding.json
        self.default_user = UNSET
        self.default_now = UNSET
        self.strict_operation_context = False
        self._pending_create_actions: list[_PendingCreateAction] = []
        self._pending_update_actions: list[_PendingUpdateAction] = []

        # Apply configuration using shared logic
        self._apply_configuration(
            model_naming=model_naming,
            route_templates=route_templates,
            storage_factory=storage_factory,
            message_queue_factory=message_queue_factory,
            admin=admin,
            permission_checker=permission_checker,
            dependency_provider=dependency_provider,
            event_handlers=event_handlers,
            encoding=encoding,
            default_user=default_user,
            default_now=default_now,
            strict_operation_context=strict_operation_context,
        )

    def _apply_configuration(
        self,
        *,
        model_naming: Literal["same", "pascal", "camel", "snake", "kebab"]
        | Callable[[type], str]
        | UnsetType = UNSET,
        route_templates: list[IRouteTemplate]
        | dict[type, dict[str, Any]]
        | None
        | UnsetType = UNSET,
        storage_factory: IStorageFactory | None | UnsetType = UNSET,
        message_queue_factory: IMessageQueueFactory | None | UnsetType = UNSET,
        admin: str | None | UnsetType = UNSET,
        permission_checker: IPermissionChecker | None | UnsetType = UNSET,
        dependency_provider: DependencyProvider | None | UnsetType = UNSET,
        event_handlers: Sequence[IEventHandler] | None | UnsetType = UNSET,
        encoding: Encoding | UnsetType = UNSET,
        default_user: str | Callable[[], str] | UnsetType = UNSET,
        default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
        strict_operation_context: bool | UnsetType = UNSET,
    ) -> None:
        """Apply configuration settings to the AutoCRUD instance.

        This internal method contains the shared logic for both __init__ and configure.
        It handles UNSET values to allow partial updates in configure() while still
        working with direct values in __init__().
        """
        # Update model_naming
        if model_naming is not UNSET:
            self.model_naming = model_naming

        # Update storage_factory and blob_store
        if storage_factory is not UNSET:
            if storage_factory is None:
                self.storage_factory = MemoryStorageFactory()
            else:
                self.storage_factory = storage_factory

            # Recreate blob_store based on new storage_factory
            if isinstance(self.storage_factory, DiskStorageFactory):
                self.blob_store = DiskBlobStore(self.storage_factory.rootdir / "_blobs")
            elif hasattr(self.storage_factory, "build_blob_store"):
                self.blob_store = self.storage_factory.build_blob_store()
            else:
                self.blob_store = MemoryBlobStore()

        # Update message_queue_factory
        if message_queue_factory is not UNSET:
            if message_queue_factory is None:
                from autocrud.message_queue.simple import SimpleMessageQueueFactory

                self.message_queue_factory = SimpleMessageQueueFactory()
            else:
                self.message_queue_factory = message_queue_factory

        # Update route_templates
        # If dependency_provider or default_user is changed, we need to
        # rebuild route_templates so the DependencyProvider picks up the
        # correct default user.
        rebuild_templates = route_templates is not UNSET or (
            (dependency_provider is not UNSET or default_user is not UNSET)
            and route_templates is UNSET
        )

        if rebuild_templates:
            self.route_templates = []
            if (
                route_templates is UNSET
                or route_templates is None
                or isinstance(route_templates, dict)
            ):
                route_templates_dict = (
                    route_templates if isinstance(route_templates, dict) else {}
                )
                dep_provider = (
                    dependency_provider if dependency_provider is not UNSET else None
                )

                # Propagate default_user to the DependencyProvider so that
                # route handlers receive the configured user instead of
                # "anonymous" when no custom get_user is set.
                effective_default_user = (
                    default_user if default_user is not UNSET else self.default_user
                )
                if effective_default_user is not UNSET:
                    base_dp = dep_provider or DependencyProvider()
                    dep_provider = base_dp.with_default_user(effective_default_user)

                for rt in [
                    CreateRouteTemplate,
                    ListRouteTemplate,
                    ReadRouteTemplate,
                    UpdateRouteTemplate,
                    PatchRouteTemplate,
                    SwitchRevisionRouteTemplate,
                    RerunRouteTemplate,
                    JobLogsRouteTemplate,
                    DeleteRouteTemplate,
                    PermanentlyDeleteRouteTemplate,
                    RestoreRouteTemplate,
                    BatchDeleteRouteTemplate,
                    BatchRestoreRouteTemplate,
                    ExportRouteTemplate,
                    ImportRouteTemplate,
                    BlobRouteTemplate,
                ]:
                    more_kwargs = route_templates_dict.get(rt, {})
                    more_kwargs.setdefault("dependency_provider", dep_provider)
                    self.route_templates.append(rt(**more_kwargs))
            else:
                self.route_templates = route_templates

        # Update permission_checker
        if permission_checker is not UNSET:
            if permission_checker is None:
                # Determine based on admin setting
                if admin is not UNSET:
                    if not admin:
                        self.permission_checker = AllowAll()
                    else:
                        self.permission_checker = RBACPermissionChecker(
                            storage_factory=self.storage_factory,
                            root_user=admin,
                        )
                else:
                    # Default when permission_checker=None but admin not provided
                    self.permission_checker = AllowAll()
            else:
                self.permission_checker = permission_checker
        elif admin is not UNSET:
            # admin changed but permission_checker not explicitly set
            if not admin:
                self.permission_checker = AllowAll()
            else:
                self.permission_checker = RBACPermissionChecker(
                    storage_factory=self.storage_factory,
                    root_user=admin,
                )

        # Update event_handlers
        if event_handlers is not UNSET:
            self.event_handlers = event_handlers

        # Update encoding
        if encoding is not UNSET:
            self.default_encoding = encoding

        # Update default_user
        if default_user is not UNSET:
            self.default_user = default_user

        # Update default_now
        if default_now is not UNSET:
            self.default_now = default_now

        # Update strict_operation_context
        if strict_operation_context is not UNSET:
            self.strict_operation_context = strict_operation_context

    def configure(
        self,
        *,
        model_naming: Literal["same", "pascal", "camel", "snake", "kebab"]
        | Callable[[type], str]
        | UnsetType = UNSET,
        route_templates: list[IRouteTemplate]
        | dict[type, dict[str, Any]]
        | UnsetType = UNSET,
        storage_factory: IStorageFactory | UnsetType = UNSET,
        message_queue_factory: IMessageQueueFactory | UnsetType = UNSET,
        admin: str | None | UnsetType = UNSET,
        permission_checker: IPermissionChecker | UnsetType = UNSET,
        dependency_provider: DependencyProvider | UnsetType = UNSET,
        event_handlers: Sequence[IEventHandler] | UnsetType = UNSET,
        encoding: Encoding | UnsetType = UNSET,
        default_user: str | Callable[[], str] | UnsetType = UNSET,
        default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
        strict_operation_context: bool | UnsetType = UNSET,
    ) -> None:
        """Configure the AutoCRUD instance dynamically.

        This method allows you to reconfigure an existing AutoCRUD instance,
        useful for the global instance pattern where you import a pre-created
        instance and configure it later in your application startup.

        Warning:
            This method should only be called during application initialization,
            before any models are registered or routes are applied. Calling this
            after models have been registered may lead to inconsistent behavior.

        Args:
            model_naming: Controls how model names are converted to URL paths.
            route_templates: Custom list of route templates or configuration dict.
            storage_factory: Storage backend to use for all models.
            message_queue_factory: Message queue factory for async job processing.
            admin: Admin user for RBAC permission system.
            permission_checker: Custom permission checker implementation.
            dependency_provider: Dependency injection provider for routes.
            event_handlers: List of event handlers for lifecycle hooks.
            encoding: Default encoding format (json/msgpack).
            default_user: Default user for operations when not specified.  When set,
                the ``DependencyProvider``'s default ``get_user`` will return this
                value instead of ``"anonymous"``.  A custom ``get_user`` on the
                provider always takes priority.
            default_now: Default timestamp function for operations.
            strict_operation_context: When ``True``, write operations on all
                registered models will raise
                :class:`MissingOperationContextError` if ``user`` and ``now``
                are not resolved from any source (explicit kwargs,
                ``using()`` scope, or manager defaults).

        Example:
            ```python
            from autocrud import crud
            from autocrud.resource_manager.storage_factory import DiskStorageFactory

            # Configure the global instance
            crud.configure(
                storage_factory=DiskStorageFactory("./data"),
                model_naming="snake",
                admin="root@example.com",
            )

            # Now register models
            crud.add_model(User)
            ```
        """
        if self.resource_managers:
            logger.warning(
                "configure() called after models have been registered. "
                "This may lead to inconsistent behavior."
            )

        # Apply configuration using shared logic
        self._apply_configuration(
            model_naming=model_naming,
            route_templates=route_templates,
            storage_factory=storage_factory,
            message_queue_factory=message_queue_factory,
            admin=admin,
            permission_checker=permission_checker,
            dependency_provider=dependency_provider,
            event_handlers=event_handlers,
            encoding=encoding,
            default_user=default_user,
            default_now=default_now,
            strict_operation_context=strict_operation_context,
        )

    def get_resource_manager(self, model: type[T] | str) -> IResourceManager[T]:
        """Get the resource manager for a registered model.

        This method allows you to access the underlying ResourceManager for a specific model.
        The ResourceManager provides low-level access to storage, events, and other
        internal components for that model.

        Args:
            model: The model class or its registered resource name.

        Returns:
            The IResourceManager instance associated with the model.

        Raises:
            KeyError: If the model is not registered.
            ValueError: If the model class is registered with multiple names (ambiguous).

        Example:
            ```python
            # Get by model class
            manager = autocrud.get_resource_manager(User)

            # Get by resource name
            manager = autocrud.get_resource_manager("users")

            # Access underlying storage
            storage = manager.storage
            ```
        """
        if isinstance(model, str):
            return self.resource_managers[model]
        model_name = self.model_names[model]
        if model_name is None:
            raise ValueError(
                f"Model {get_type_name(model) or repr(model)} is registered with multiple names."
            )
        return self.resource_managers[model_name]

    def _is_job_subclass(self, model: type) -> bool:
        """Check if a model is a subclass of Job.

        Args:
            model: The model class to check.

        Returns:
            True if the model is a Job subclass, False otherwise.
        """
        return is_generic_subclass(model, Job)

    def _resource_name(self, model: type[T]) -> str:
        """Convert model class name to resource name using the configured naming convention.

        This internal method handles the conversion of Python class names to URL-friendly
        resource names based on the model_naming configuration.

        Args:
            model: The model class whose name should be converted.

        Returns:
            The converted resource name string that will be used in URLs.

        Examples:
            With model_naming="kebab":
            - UserProfile -> "user-profile"
            - BlogPost -> "blog-post"

            With model_naming="snake":
            - UserProfile -> "user_profile"
            - BlogPost -> "blog_post"

            With custom function:
            - Can implement any custom naming logic
        """
        if callable(self.model_naming):
            return self.model_naming(model)
        original_name = get_type_name(model)
        if original_name is None:
            raise ValueError(
                f"Cannot automatically infer a resource name for type {model!r}. "
                f"Please provide a name explicitly via "
                f"add_model(..., name='your_name')."
            )

        # 使用 NameConverter 進行轉換
        return NameConverter(original_name).to(self.model_naming)

    def add_route_template(self, template: IRouteTemplate) -> None:
        """Add a custom route template to extend the API with additional endpoints.

        Route templates define how to generate specific API endpoints for models.
        By adding custom templates, you can extend the default CRUD functionality
        with specialized endpoints for your use cases.

        If a template of the **same type** already exists (e.g. added by the
        default ``configure()``), it is **replaced** rather than duplicated.
        This prevents ``Duplicate Operation ID`` warnings for templates that
        mount global routes such as ``BlobRouteTemplate`` and
        ``GraphQLRouteTemplate``.

        Args:
            template: A custom route template implementing IRouteTemplate interface.

        Example:
            ```python
            class CustomSearchTemplate(BaseRouteTemplate):
                def apply(self, model_name, resource_manager, router):
                    @router.get(f"/{model_name}/search")
                    async def search_resources(query: str):
                        # Custom search logic
                        pass


            autocrud = AutoCRUD()
            autocrud.add_route_template(CustomSearchTemplate())
            autocrud.add_model(User)
            ```

        Note:
            Templates are sorted by their order property before being applied.
            Add templates before calling add_model() or apply() for best results.
        """
        # Replace any existing template of the same type to avoid duplicates.
        # This is important for templates that mount global routes (e.g.
        # BlobRouteTemplate, GraphQLRouteTemplate) — having two instances
        # would register the same path twice, producing a FastAPI
        # "Duplicate Operation ID" warning.
        template_type = type(template)
        self.route_templates = [
            t for t in self.route_templates if type(t) is not template_type
        ]
        self.route_templates.append(template)

    def create_action(
        self,
        resource_name: str,
        *,
        path: str | None = None,
        label: str | None = None,
        async_mode: Literal["job", "background"] | None = None,
        job_name: str | None = None,
    ) -> Callable:
        """Decorator to register a custom create action for a resource.

        The decorated function is a standard FastAPI endpoint handler — all input
        parsing (``Body``, ``Query``, ``Path``, ``Depends``, etc.) is handled by
        FastAPI.  If the handler returns a resource-type object, AutoCRUD will
        automatically call ``resource_manager.create()`` and respond with
        ``RevisionInfo``.  If it returns ``None``, no automatic creation occurs.

        When ``async_mode='job'`` is set, the framework automatically:

        1. Generates a ``Job`` model with the handler's body type as payload.
        2. Registers the Job model with a message queue.
        3. On POST, creates a Job instance (PENDING) and enqueues it.
        4. Returns HTTP 202 with :class:`~autocrud.types.JobRedirectInfo`.
        5. In the background, executes the handler with the payload.
        6. If the handler returns a resource object, auto-creates it and
           stores the ``RevisionInfo`` as the Job's artifact.

        When ``async_mode='background'`` is set, the framework:

        1. On POST, schedules the handler via FastAPI ``BackgroundTasks``.
        2. Returns HTTP 202 with :class:`~autocrud.types.BackgroundTaskAccepted`
           immediately.
        3. The handler runs in the background; if it returns a resource object,
           ``resource_manager.create()`` is called automatically.
        4. No Job model is created — the task is fire-and-forget.
        5. Errors are logged but not surfaced to the client.

        This mode is suitable for tasks that take a few seconds to complete
        and do not require progress tracking.

        Args:
            resource_name: The name of the resource this action belongs to.
            path: URL path suffix (e.g. ``"import-from-url"``).  If ``None``,
                inferred from the function name (underscores → hyphens).
            label: Human-friendly label shown in the UI.  If ``None``,
                inferred from *path* (hyphens → spaces, title-cased).
            async_mode: Execution mode for the action.  ``None`` (default)
                executes synchronously.  ``'job'`` executes asynchronously
                via the message queue system.  ``'background'`` executes
                asynchronously via FastAPI ``BackgroundTasks``
                (fire-and-forget, no Job tracking).
            job_name: Custom resource name for the auto-generated Job model
                (e.g. ``"my-custom-job"``).  If ``None``, derived automatically
                from *path* and *resource_name*.  Only meaningful when
                ``async_mode='job'``.

        Returns:
            A decorator that registers the handler and returns it unchanged.

        Example:
            ```python
            class ImportFromUrl(Struct):
                url: str


            @crud.create_action("article", label="Import from URL")
            async def import_from_url(body: ImportFromUrl = Body(...)):
                content = await fetch_and_parse(body.url)
                return Article(content=content)  # auto-created


            class GenerateRequest(Struct):
                prompt: str


            @crud.create_action("article", async_mode="job", label="Generate")
            def generate_article(payload: GenerateRequest = Body(...)) -> Article:
                content = call_llm(payload.prompt)  # long-running
                return Article(content=content)  # auto-created in background
            ```

        Note:
            This decorator is lazy — it stores metadata without registering any
            route.  Routes are created when ``apply()`` is called, so the
            decorator can be used before or after ``add_model()``.
        """

        def decorator(func: Callable) -> Callable:
            action_path = path or func.__name__.replace("_", "-")
            action_label = label or action_path.replace("-", " ").title()
            self._pending_create_actions.append(
                _PendingCreateAction(
                    resource_name=resource_name,
                    path=action_path,
                    label=action_label,
                    handler=func,
                    async_mode=async_mode,
                    job_name=job_name,
                )
            )
            return func

        return decorator

    def update_action(
        self,
        resource_name: str,
        *,
        path: str | None = None,
        label: str | None = None,
        mode: Literal["update", "modify"] = "update",
        existing_param: str = "existing",
        info_param: str = "info",
        meta_param: str = "meta",
        async_mode: Literal["job", "background"] | None = None,
        job_name: str | None = None,
    ) -> Callable:
        """Decorator to register a custom update action for a resource.

        The decorated function receives the existing resource data (auto-injected)
        and any custom input parameters.  If the handler returns a resource-type
        object, AutoCRUD will automatically call ``resource_manager.update()`` (or
        ``resource_manager.modify()`` when ``mode='modify'``) and respond with
        ``RevisionInfo``.  If it returns ``None``, no update occurs.

        The existing resource data is automatically fetched via
        ``resource_manager.get(resource_id)`` and injected into the handler
        parameter named by *existing_param* (default ``"existing"``).

        Similarly, the handler may declare parameters named *info_param*
        (default ``"info"``) and *meta_param* (default ``"meta"``) to
        receive the existing resource's ``RevisionInfo`` and ``ResourceMeta``
        respectively.  Like *existing_param*, these are detected by
        **parameter name** and only injected when the handler declares them.

        When ``async_mode='job'`` is set, the framework automatically:

        1. Generates a ``Job`` model with the handler's body type as payload
           (plus an auto-injected ``resource_id`` field).
        2. Registers the Job model with a message queue.
        3. On POST, creates a Job instance (PENDING) and enqueues it.
        4. Returns HTTP 202 with :class:`~autocrud.types.JobRedirectInfo`.
        5. In the background, fetches existing resource (lazy), executes
           the handler with the payload and existing data.
        6. If the handler returns a resource object, auto-updates it and
           stores the ``RevisionInfo`` as the Job's artifact.

        When ``async_mode='background'`` is set, the framework:

        1. On POST, schedules the handler via FastAPI ``BackgroundTasks``.
        2. Returns HTTP 202 with :class:`~autocrud.types.BackgroundTaskAccepted`
           immediately.
        3. The handler runs in the background; if it returns a resource object,
           ``resource_manager.update()`` (or ``modify()``) is called
           automatically.
        4. No Job model is created — the task is fire-and-forget.
        5. Errors are logged but not surfaced to the client.

        Args:
            resource_name: The name of the resource this action belongs to.
            path: URL path suffix (e.g. ``"level-up"``).  If ``None``,
                inferred from the function name (underscores → hyphens).
            label: Human-friendly label shown in the UI.  If ``None``,
                inferred from *path* (hyphens → spaces, title-cased).
            mode: Update mode.  ``"update"`` (default) creates a new
                revision.  ``"modify"`` performs an in-place edit (only
                valid for draft-status resources).
            existing_param: The handler parameter name into which the
                existing resource data will be injected.  Defaults to
                ``"existing"``.
            info_param: The handler parameter name into which the
                existing resource's ``RevisionInfo`` will be injected.
                Defaults to ``"info"``.
            meta_param: The handler parameter name into which the
                existing resource's ``ResourceMeta`` will be injected.
                Defaults to ``"meta"``.
            async_mode: Execution mode for the action.  ``None`` (default)
                executes synchronously.  ``'job'`` executes asynchronously
                via the message queue system.  ``'background'`` executes
                asynchronously via FastAPI ``BackgroundTasks``
                (fire-and-forget, no Job tracking).
            job_name: Custom resource name for the auto-generated Job model
                (e.g. ``"my-custom-job"``).  If ``None``, derived automatically
                from *path* and *resource_name*.  Only meaningful when
                ``async_mode='job'``.

        Returns:
            A decorator that registers the handler and returns it unchanged.

        Example:
            ```python
            class LevelUpInput(Struct):
                levels: int = 1


            @crud.update_action("character", label="Level Up")
            def level_up(
                existing: Character,
                body: LevelUpInput = Body(...),
            ) -> Character:
                return Character(
                    name=existing.name,
                    level=existing.level + body.levels,
                )


            @crud.update_action(
                "character",
                label="Train",
                async_mode="job",
            )
            def train(
                existing: Character,
                body: LevelUpInput = Body(...),
            ) -> Character:
                import time

                time.sleep(10)  # long-running training
                return Character(
                    name=existing.name,
                    level=existing.level + body.levels,
                )


            @crud.update_action(
                "character",
                label="Background Heal",
                async_mode="background",
            )
            def bg_heal(existing: Character) -> Character:
                import time

                time.sleep(5)
                return Character(name=existing.name, level=existing.level + 1)
            ```

        Note:
            This decorator is lazy — it stores metadata without registering any
            route.  Routes are created when ``apply()`` is called.
            The route is ``POST /{resource_name}/{resource_id}/{action_path}``.
        """

        def decorator(func: Callable) -> Callable:
            action_path = path or func.__name__.replace("_", "-")
            action_label = label or action_path.replace("-", " ").title()
            self._pending_update_actions.append(
                _PendingUpdateAction(
                    resource_name=resource_name,
                    path=action_path,
                    label=action_label,
                    handler=func,
                    mode=mode,
                    existing_param=existing_param,
                    info_param=info_param,
                    meta_param=meta_param,
                    async_mode=async_mode,
                    job_name=job_name,
                )
            )
            return func

        return decorator

    def add_model(
        self,
        model: "type[T] | Schema[T]",
        *,
        name: str | None = None,
        id_generator: Callable[[], str] | None = None,
        storage: IStorage | None = None,
        migration: "IMigration | Schema | None" = None,
        indexed_fields: list[str | tuple[str, type] | IndexableField] | None = None,
        event_handlers: Sequence[IEventHandler] | None = None,
        permission_checker: IPermissionChecker | None = None,
        encoding: Encoding | None = None,
        default_status: RevisionStatus | None = None,
        default_user: str | Callable[[], str] | UnsetType = UNSET,
        default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
        message_queue_factory: IMessageQueueFactory | None | UnsetType = UNSET,
        job_handler: Callable[[Resource[Job[T]]], None] | None = None,
        job_handler_factory: Callable[[], Callable[[Resource[Job[T]]], None]]
        | None = None,
        validator: "Callable[[T], None] | IValidator | type | None" = None,
        constraint_checkers: "Sequence[IConstraintChecker | Callable[[ResourceManager], IConstraintChecker]] | None" = None,
    ) -> None:
        """Register a resource model (or `Schema`) and create its `ResourceManager`.

        After a model is registered, calling `apply(router)` will generate FastAPI routes for it
        using the configured route templates.

        You can register either:
        - a plain model type: `add_model(User)`
        - a `Schema`: `add_model(Schema(User, version=...))`

        Args:
            model:
                Resource type or `Schema`. Supported types depend on your project setup, commonly
                msgspec `Struct`. Pydantic `BaseModel` is supported and will be converted to a struct.
            name:
                Resource name (used as route base path). If `None`, derived from the model type and
                `model_naming`.
            id_generator:
                Custom ID generator for created resources. If `None`, the default generator is used
                by `ResourceManager`.
            storage:
                Storage instance for this resource. If `None`, a storage is created via
                `self.storage_factory.build(model_name)`.
            migration:
                Schema/migration configuration.
                - If `model` is a `Schema`, `migration` must be `None`.
                - If `migration` is a `Schema`, it is used as the resolved schema for this model.
                - Passing `IMigration` is supported but **deprecated** (converted via `Schema.from_legacy`).
            indexed_fields:
                Fields to index for search/query. Each element can be:
                - `IndexableField`
                - `str` (field path)
                - `(field_path: str, field_type: type)` tuple
            event_handlers:
                Per-model event handlers. If `self.event_handlers` is configured globally, it takes
                precedence; otherwise these handlers are used.
            permission_checker:
                Per-model permission checker. If `self.permission_checker` is configured globally, it
                takes precedence; otherwise this checker is used.
            encoding:
                Encoding for stored payloads. If `None`, uses `self.default_encoding`.
            default_status:
                Default revision status for this resource (if supported by the revision model).
            default_user:
                Per-model default user (or factory). If `UNSET`, falls back to `self.default_user`
                when configured.
            default_now:
                Per-model default timestamp function. If `UNSET`, falls back to `self.default_now`
                when configured.
            message_queue_factory:
                Overrides message queue behavior for Job models:
                - `UNSET`: use `self.message_queue_factory`
                - `None`: explicitly disable queue
                - factory instance: use the provided factory
            job_handler:
                Handler for Job resources (when the model is detected as a Job subclass).
            job_handler_factory:
                Lazy factory producing a job handler. If provided, it is wrapped as a lazy handler.
            validator:
                Validation hook(s). When the model is a Pydantic `BaseModel` and no validator is set
                on the resolved schema, the Pydantic model is used as validator by default.
            constraint_checkers:
                Extra constraint checkers for this resource. Each element can be an instance or a
                factory callable that receives the `ResourceManager` and returns a checker.

        Behavior:
            - If `model` is a `Schema`, it must declare `resource_type`; schema-level migration/validator
            should be provided on the `Schema` itself.
            - If the model is a Pydantic type, it is converted to a struct for storage and the Pydantic
            model can be used for validation.
            - Ref relationships are collected from `Ref` / `RefRevision` annotations for later route and
            referential integrity setup.
            - Ref fields (resource_id refs only) are auto-indexed for searchability.
            - For Job models with a message queue enabled, `status` and `retries` are auto-indexed
            (if not already present in `indexed_fields`).

        Raises:
            ValueError:
                - if the resource name already exists
                - if `Schema` is passed as first argument but `migration`/`validator` is also provided
                - if `Ref(..., on_delete=set_null)` is used on a non-optional field
            TypeError:
                - if `indexed_fields` contains an invalid item

        Examples:
            Basic registration:

            ```python
            from autocrud import AutoCRUD

            autocrud = AutoCRUD()
            autocrud.add_model(User)
            ```

            Custom resource name:

            ```python
            autocrud.add_model(User, name="people")
            ```

            Provide explicit storage:

            ```python
            # storage is per-model; if you want a default for all models, pass `storage_factory=...`
            # when constructing AutoCRUD / calling configure().
            model_name = "people"
            st = autocrud.storage_factory.build(model_name)
            autocrud.add_model(User, name=model_name, storage=st)
            ```

            Using Schema as the first argument:

            ```python
            schema = Schema(User, version="v1")
            autocrud.add_model(schema)
            ```
        """
        _indexed_fields: list[IndexableField] = []
        for field in indexed_fields or []:
            if isinstance(field, IndexableField):
                _indexed_fields.append(field)
            elif (
                isinstance(field, tuple)
                and len(field) == 2
                and isinstance(field[0], str)
            ):
                field = IndexableField(field_path=field[0], field_type=field[1])
                _indexed_fields.append(field)
            elif isinstance(field, str):
                field = IndexableField(field_path=field, field_type=UNSET)
                _indexed_fields.append(field)
            else:
                raise TypeError(
                    "Invalid indexed field, should be IndexableField or tuple[field_name, field_type]",
                )

        # ── Resolve Schema vs type argument ────────────────────────
        import warnings

        resolved_schema: Schema | None = None
        if isinstance(model, Schema):
            # Schema passed as first argument
            if migration is not None:
                raise ValueError(
                    "Cannot specify 'migration' when passing Schema as the first argument. "
                    "Define migration steps on the Schema instead."
                )
            if validator is not None:
                raise ValueError(
                    "Cannot specify 'validator' when passing Schema as the first argument. "
                    "Pass validator to Schema(..., validator=...) instead."
                )
            resolved_schema = model
            model = resolved_schema.resource_type  # type: ignore[assignment]
            if model is None:
                raise ValueError(
                    "Schema passed as first argument must have a resource_type."
                )
        else:
            # model is a plain type
            if isinstance(migration, Schema):
                resolved_schema = migration
            elif isinstance(migration, IMigration):
                warnings.warn(
                    "Passing IMigration to migration= is deprecated. "
                    "Use Schema(resource_type, version).step(...) instead.",
                    DeprecationWarning,
                    stacklevel=2,
                )
                resolved_schema = Schema.from_legacy(migration)
            # else migration is None → no schema

        model_name = name or self._resource_name(model)

        # Handle Pydantic BaseModel as model type:
        # auto-generate struct and use Pydantic for validation
        pydantic_model = None
        if is_pydantic_model(model):
            pydantic_model = model
            model = pydantic_to_struct(pydantic_model)
            if validator is None and (
                resolved_schema is None or not resolved_schema.has_validator
            ):
                validator = pydantic_model

        if model_name in self.resource_managers:
            raise ValueError(f"Model name {model_name} already exists.")
        if model in self.model_names:
            self.model_names[model] = None
            logger.warning(
                f"Model {get_type_name(model) or repr(model)} is already registered with a different name. "
                f"This resource manager will not be accessible by its type.",
            )
        else:
            self.model_names[model] = model_name
        if storage is None:
            storage = self.storage_factory.build(model_name)
        if encoding is None:
            encoding = self.default_encoding
        other_options = {}
        if default_status is not None:
            other_options["default_status"] = default_status
        if default_user is not UNSET:
            other_options["default_user"] = default_user
        elif self.default_user is not UNSET:
            other_options["default_user"] = self.default_user
        if default_now is not UNSET:
            other_options["default_now"] = default_now
        elif self.default_now is not UNSET:
            other_options["default_now"] = self.default_now
        # Auto-detect Job subclass and create message queue
        if self._is_job_subclass(model) and (
            job_handler is not None or job_handler_factory is not None
        ):
            # Determine which factory to use
            if message_queue_factory is UNSET:
                mq_factory = self.message_queue_factory
            elif message_queue_factory is None:
                mq_factory = None  # Explicitly disabled
            else:
                mq_factory = message_queue_factory

            if mq_factory is not None:
                real_handler = job_handler
                if job_handler_factory is not None:
                    real_handler = LazyJobHandler(job_handler_factory)

                # Create message queue with job handler
                other_options["message_queue"] = mq_factory.build(real_handler)

                # Check if status is already in indexed fields
                if not any(field.field_path == "status" for field in _indexed_fields):
                    _indexed_fields.append(
                        IndexableField(field_path="status", field_type=TaskStatus)
                    )

                # Check if retries is already in indexed fields
                if not any(field.field_path == "retries" for field in _indexed_fields):
                    _indexed_fields.append(
                        IndexableField(field_path="retries", field_type=int)
                    )

        resource_manager = ResourceManager(
            model,
            storage=storage,
            blob_store=self.blob_store,
            id_generator=id_generator,
            migration=resolved_schema or migration,
            indexed_fields=_indexed_fields,
            event_handlers=self.event_handlers or event_handlers,
            permission_checker=self.permission_checker or permission_checker,
            encoding=encoding,
            name=model_name,
            validator=validator,
            pydantic_type=pydantic_model,
            constraint_checkers=constraint_checkers,
            strict_operation_context=self.strict_operation_context,
            **other_options,
        )
        self.resource_managers[model_name] = resource_manager

        # Scan Ref / RefRevision annotations and collect relationships
        refs = extract_refs(model, model_name)
        self.relationships.extend(refs)
        # Validate set_null requires nullable field
        for ref_info in refs:
            if ref_info.on_delete == OnDelete.set_null and not ref_info.nullable:
                raise ValueError(
                    f"Ref on '{get_type_name(model) or repr(model)}.{ref_info.source_field}' uses "
                    f"on_delete=set_null but the field is not Optional. "
                    f"Use Annotated[str | None, Ref(...)] instead."
                )

        # Auto-index Ref fields (resource_id refs only) for searchability
        for ref_info in refs:
            if ref_info.ref_type == "resource_id":
                # Use list[str] for list refs, str for scalar refs
                field_type = list[str] if ref_info.is_list else str
                resource_manager.add_indexed_field(
                    IndexableField(
                        field_path=ref_info.source_field,
                        field_type=field_type,
                    )
                )

    @staticmethod
    def _get_unique_fields(rm: ResourceManager) -> list[str]:
        """Extract unique field names from the RM's registered constraint checkers."""
        from autocrud.resource_manager.constraint_handler import (
            ConstraintEventHandler,
        )
        from autocrud.resource_manager.unique_handler import (
            UniqueConstraintChecker,
        )

        for h in rm.event_handlers:
            handler = None
            if isinstance(h, ConstraintEventHandler):
                handler = h
            if handler is not None:
                for c in handler.checkers:
                    if isinstance(c, UniqueConstraintChecker):
                        return c.unique_fields
        return []

    def openapi(self, app: FastAPI, structs: list[type] = None) -> None:
        """Generate and register the OpenAPI schema for the FastAPI application.

        This method customizes the OpenAPI schema generation to include all the
        AutoCRUD-specific types, models, and response schemas. It ensures that
        the generated API documentation (Swagger UI / ReDoc) correctly reflects
        the structure of your resources and their endpoints.

        Args:
            app: The FastAPI application instance.
            structs: Optional list of additional msgspec Structs to include in the schema.

        Note:
            When :meth:`apply` is called with a ``FastAPI`` instance as the
            first argument, this method is called automatically at the end of
            ``apply()``.  You only need to call it manually if you passed a
            bare ``APIRouter`` to ``apply()`` or need to customise the
            ``structs`` parameter separately.
        """

        # Handle root_path by setting servers if not already set
        structs = structs or []
        servers = app.servers
        if app.root_path and not servers:
            servers = [{"url": app.root_path}]

        app.openapi_schema = get_openapi(
            title=app.title,
            version=app.version,
            openapi_version=app.openapi_version,
            summary=app.summary,
            description=app.description,
            terms_of_service=app.terms_of_service,
            contact=app.contact,
            license_info=app.license_info,
            routes=app.routes,
            webhooks=app.webhooks.routes,
            tags=app.openapi_tags,
            servers=servers,
            separate_input_output_schemas=app.separate_input_output_schemas,
        )
        app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
            [
                ResourceMeta,
                RevisionInfo,
                RevisionListResponse,
                *[rm.resource_type for rm in self.resource_managers.values()],
                *[
                    FullResourceResponse[rm.resource_type]
                    for rm in self.resource_managers.values()
                ],
                RFC6902_Add,
                RFC6902_Remove,
                RFC6902_Replace,
                RFC6902_Move,
                RFC6902_Test,
                RFC6902_Copy,
                RFC6902,
                *structs,
            ],
        )[1]

        # Include MigrateProgress and MigrateResult when MigrateRouteTemplate is active
        if any(isinstance(rt, MigrateRouteTemplate) for rt in self.route_templates):
            app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
                [MigrateProgress, MigrateResult],
            )[1]

        # Include custom create action body schemas in components
        action_body_structs = []
        for action in self._pending_create_actions:
            if action.resource_name not in self.resource_managers:
                warnings.warn(
                    f"Resource '{action.resource_name}' not found in resource managers. "
                    f"Skipping action '{action.handler.__name__}'.",
                    stacklevel=2,
                )
                continue
            action_body_structs.extend(self._collect_action_body_structs(action))
        # Also include custom update action body schemas
        for action in self._pending_update_actions:
            if action.resource_name not in self.resource_managers:
                continue
            action_body_structs.extend(
                self._collect_action_body_structs(
                    action,
                    skip_params={
                        action.existing_param,
                        action.info_param,
                        action.meta_param,
                    },
                )
            )
        if action_body_structs:
            app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
                action_body_structs,
            )[1]

        # Inject x-ref-* / x-ref-revision-* metadata into schema properties
        self._inject_ref_metadata(app.openapi_schema)

        # Inject x-autocrud-custom-create-actions top-level extension
        self._inject_custom_create_actions(app.openapi_schema)

        # Inject x-autocrud-custom-update-actions top-level extension
        self._inject_custom_update_actions(app.openapi_schema)

        # Inject x-autocrud-async-create-jobs mapping (job resource → parent)
        self._inject_async_create_jobs(app.openapi_schema)

        # Inject x-autocrud-async-update-jobs mapping (job resource → parent)
        self._inject_async_update_jobs(app.openapi_schema)

        # Inject x-autocrud-indexed-fields mapping (resource → indexed field paths)
        self._inject_indexed_fields(app.openapi_schema)

        # Promote inline $defs (from Pydantic-generated schemas) into
        # components/schemas and rewrite $ref paths so swagger-parser can
        # resolve them.  Must run before _resolve_missing_schema_refs.
        self._promote_defs_to_components(app.openapi_schema)

        # Resolve dangling $ref pointers caused by module-qualified name
        # divergence (e.g. route $ref → "Skill" but components only has
        # "__main___Skill" due to pydantic_to_struct round-trip duplication).
        self._resolve_missing_schema_refs(app.openapi_schema)

    @staticmethod
    def _promote_defs_to_components(schema: dict) -> None:
        """Hoist inline ``$defs`` from component schemas into ``components/schemas``.

        When FastAPI embeds a Pydantic model's JSON Schema inside a ``Body``
        component, the resulting schema may contain a property-level ``$defs``
        block with ``$ref: "#/$defs/X"`` pointers.  These absolute JSON
        pointers target the **document root** which has no ``$defs`` key,
        causing swagger-parser / json-schema-ref-parser to fail.

        This method:

        1. Walks all component schemas looking for nested ``$defs`` dicts.
        2. Promotes each definition to ``#/components/schemas/<name>``
           (skipping if a name already exists to avoid overwriting).
        3. Rewrites ``$ref: "#/$defs/X"`` → ``"#/components/schemas/X"``
           within the same schema tree (including ``discriminator.mapping``).
        4. Removes the now-unnecessary ``$defs`` keys.

        Args:
            schema: The full OpenAPI schema dict (mutated in-place).
        """
        components = schema.get("components", {}).get("schemas", {})
        if not components:
            return

        defs_prefix = "#/$defs/"
        comp_prefix = "#/components/schemas/"

        # Collect all $defs entries that need promotion
        # Each entry: (parent_dict_containing_$defs_key, defs_dict)
        defs_to_promote: list[tuple[dict, dict]] = []

        def _find_defs(obj: Any) -> None:
            """Recursively find dicts containing a ``$defs`` key."""
            if isinstance(obj, dict):
                if "$defs" in obj and isinstance(obj["$defs"], dict):
                    defs_to_promote.append((obj, obj["$defs"]))
                for v in obj.values():
                    if isinstance(v, (dict, list)):
                        _find_defs(v)
            elif isinstance(obj, list):
                for item in obj:
                    _find_defs(item)

        # Search within all existing component schemas
        for comp_value in list(components.values()):
            _find_defs(comp_value)

        if not defs_to_promote:
            return

        # Build the rename map: $defs name → components name
        rename_map: dict[str, str] = {}
        for _parent, defs_dict in defs_to_promote:
            for def_name, def_schema in defs_dict.items():
                if def_name not in components:
                    components[def_name] = def_schema
                    rename_map[def_name] = def_name
                else:
                    # Name collision — keep existing, still rewrite refs
                    rename_map[def_name] = def_name

        def _rewrite_defs_refs(obj: Any) -> Any:
            """Rewrite ``#/$defs/X`` refs to ``#/components/schemas/X``
            and strip ``$defs`` keys."""
            if isinstance(obj, dict):
                out: dict[str, Any] = {}
                for k, v in obj.items():
                    # Drop the $defs block — its contents are now in components
                    if k == "$defs" and isinstance(v, dict):
                        continue
                    if k == "$ref" and isinstance(v, str) and v.startswith(defs_prefix):
                        def_name = v[len(defs_prefix) :]
                        target = rename_map.get(def_name, def_name)
                        out[k] = comp_prefix + target
                    elif k == "mapping" and isinstance(v, dict):
                        out[k] = {
                            mk: (
                                comp_prefix
                                + rename_map.get(
                                    mv[len(defs_prefix) :], mv[len(defs_prefix) :]
                                )
                                if isinstance(mv, str) and mv.startswith(defs_prefix)
                                else mv
                            )
                            for mk, mv in v.items()
                        }
                    else:
                        out[k] = _rewrite_defs_refs(v)
                return out
            if isinstance(obj, list):
                return [_rewrite_defs_refs(item) for item in obj]
            return obj

        # Rewrite refs in every component schema that had $defs
        for comp_name in list(components.keys()):
            components[comp_name] = _rewrite_defs_refs(components[comp_name])

    @staticmethod
    def _resolve_missing_schema_refs(schema: dict) -> None:
        """Add alias entries for dangling ``$ref`` pointers in the OpenAPI schema.

        When ``msgspec.json.schema_components()`` is called with two types that
        share the same ``__name__`` but differ in ``__module__`` (e.g. the
        original ``Skill`` from ``__main__`` and a round-tripped version from
        ``pydantic_converter``), msgspec disambiguates by emitting
        module-qualified names like ``__main___Skill``.

        Meanwhile, per-route ``jsonschema_to_json_schema_extra(Skill)`` only
        sees **one** Skill type and uses the simple name ``Skill`` in its
        ``$ref``.  This mismatch leaves dangling ``$ref`` pointers.

        This method scans the entire schema for ``$ref`` targets, identifies
        those that are missing from ``components.schemas``, and creates alias
        entries by copying the schema from a matching module-prefixed
        candidate.  The ``__main__`` variant is preferred over others.

        Args:
            schema: The full OpenAPI schema dict (mutated in-place).
        """
        import json
        import re

        components = schema.get("components", {}).get("schemas", {})
        if not components:
            return

        # Collect all $ref target names from the entire schema
        schema_json = json.dumps(schema)
        all_ref_names: set[str] = set(
            re.findall(r'"\$ref":\s*"#/components/schemas/([^"]+)"', schema_json)
        )

        missing = all_ref_names - set(components.keys())
        if not missing:
            return

        for simple_name in missing:
            # Find component keys that end with _<simple_name> (module prefix)
            candidates: list[str] = []
            for comp_name in components:
                if comp_name.endswith(f"_{simple_name}") and comp_name != simple_name:
                    candidates.append(comp_name)

            if not candidates:
                continue

            # Prefer __main__ variant (user's original type) over others
            chosen = candidates[0]
            for c in candidates:
                if c.startswith("__main__"):
                    chosen = c
                    break

            # Create an alias: copy the chosen component under the simple name
            components[simple_name] = components[chosen].copy()

    def _inject_ref_metadata(self, schema: dict) -> None:
        """Post-process OpenAPI schema to inject ``x-ref-*`` extensions.

        This scans all registered resource Structs — and their nested Struct
        fields (e.g. ``Job[PayloadType]``) — for ``Ref`` / ``RefRevision``
        annotations and writes the corresponding ``x-ref-resource``,
        ``x-ref-type``, and ``x-ref-on-delete`` extensions into the matching
        schema properties so the web generator can discover relationships.

        It also injects ``x-display-name-field`` and ``x-unique`` extensions.

        When ``struct_to_pydantic()`` causes module-qualified name dedup
        (e.g. only ``__main___Skill`` exists instead of ``Skill``), the
        internal ``_find_component`` helper falls back to module-prefixed
        variants so metadata is still injected correctly.
        """
        components = schema.get("components", {}).get("schemas", {})
        all_refs: list[_RefInfo] = []

        # Collect (schema_name, refs) pairs for both top-level and nested Structs
        processed_structs: set[type] = set()

        def _find_component(simple_name: str) -> dict | None:
            """Look up a component by simple name, falling back to a
            module-qualified variant when name dedup has eliminated the
            simple entry (e.g. only ``__main___Skill`` exists).

            Prefers the ``__main__`` variant when multiple candidates exist.
            """
            comp = components.get(simple_name)
            if comp is not None:
                return comp
            candidates = [
                k
                for k in components
                if k.endswith(f"_{simple_name}") and k != simple_name
            ]
            if not candidates:
                return None
            chosen = candidates[0]
            for c in candidates:
                if c.startswith("__main__"):
                    chosen = c
                    break
            return components.get(chosen)

        def _inject_into_component(comp_name: str, refs: list[_RefInfo]) -> None:
            comp = _find_component(comp_name)
            if not comp or "properties" not in comp:
                return
            for ref_info in refs:
                prop = comp["properties"].get(ref_info.source_field)
                if not prop:
                    continue
                ext: dict[str, str] = {
                    "x-ref-resource": ref_info.target,
                    "x-ref-type": ref_info.ref_type,
                }
                if ref_info.ref_type == "resource_id":
                    ext["x-ref-on-delete"] = ref_info.on_delete.value
                prop.update(ext)

        def _process_single_struct(
            struct_type: type,
            model_name: str,
            *,
            inject_unique: bool = False,
            rm: Any = None,
        ) -> None:
            """Inject ref / display-name / unique metadata for a single Struct
            type into its OpenAPI component, and recurse into nested Structs.
            """
            struct_name = get_type_name(struct_type)
            if struct_name is None:
                return

            refs = extract_refs(struct_type, model_name)
            all_refs.extend(refs)
            _inject_into_component(struct_name, refs)

            # DisplayName annotation
            from autocrud.types import extract_display_name

            dn_field = extract_display_name(struct_type)
            if dn_field is not None:
                comp = _find_component(struct_name)
                if comp is not None:
                    comp["x-display-name-field"] = dn_field

            # Unique field annotations (only for top-level resource types)
            if inject_unique and rm is not None:
                unique_fields = self._get_unique_fields(rm)
                if unique_fields:
                    comp = _find_component(struct_name)
                    if comp is not None:
                        props = comp.get("properties", {})
                        for uf in unique_fields:
                            prop = props.get(uf)
                            if prop is not None:
                                prop["x-unique"] = True

            # Nested Struct types (e.g. Job[Payload])
            nested = collect_nested_struct_types(struct_type, set())
            for nested_struct in nested:
                if nested_struct in processed_structs:
                    continue
                processed_structs.add(nested_struct)
                nested_refs = extract_refs(nested_struct, model_name)
                all_refs.extend(nested_refs)
                nested_name = get_type_name(nested_struct)
                if nested_name is not None:
                    _inject_into_component(nested_name, nested_refs)

        for model_name, rm in self.resource_managers.items():
            processed_structs.add(rm.resource_type)

            if is_union_type(rm.resource_type):
                # Union type (e.g. Cat | Dog) — process each member type
                member_types = get_union_args(rm.resource_type) or ()
                for member_type in member_types:
                    if member_type in processed_structs:
                        continue
                    processed_structs.add(member_type)
                    _process_single_struct(
                        member_type, model_name, inject_unique=False, rm=rm
                    )
                continue

            # Regular (non-union) resource Struct
            _process_single_struct(
                rm.resource_type, model_name, inject_unique=True, rm=rm
            )

        # Also process custom create action body schemas so that Ref /
        # RefRevision annotations in action body Structs get x-ref-*
        # extensions injected into their OpenAPI components.
        for action in self._pending_create_actions:
            if action.resource_name not in self.resource_managers:
                continue
            for body_struct in self._collect_action_body_structs(action):
                if body_struct in processed_structs:
                    continue
                processed_structs.add(body_struct)
                _process_single_struct(
                    body_struct,
                    action.resource_name,
                    inject_unique=False,
                    rm=None,
                )

        # Also process custom update action body schemas
        for action in self._pending_update_actions:
            if action.resource_name not in self.resource_managers:
                continue
            for body_struct in self._collect_action_body_structs(
                action,
                skip_params={
                    action.existing_param,
                    action.info_param,
                    action.meta_param,
                },
            ):
                if body_struct in processed_structs:
                    continue
                processed_structs.add(body_struct)
                _process_single_struct(
                    body_struct,
                    action.resource_name,
                    inject_unique=False,
                    rm=None,
                )

        # Also inject a top-level x-autocrud-relationships extension
        if all_refs:
            schema["x-autocrud-relationships"] = [
                {
                    "source": r.source,
                    "sourceField": r.source_field,
                    "target": r.target,
                    "refType": r.ref_type,
                    "onDelete": r.on_delete.value,
                    "nullable": r.nullable,
                }
                for r in all_refs
            ]

    @staticmethod
    def _get_body_schema_name(
        handler: Any, *, skip_params: set[str] | None = None
    ) -> str | None:
        """Extract the body parameter's schema name from a handler signature.

        Scans the handler's parameters for the first ``msgspec.Struct`` or
        Pydantic ``BaseModel`` type annotation and returns its ``__name__``.

        Args:
            handler: The handler function to inspect.
            skip_params: Parameter names to skip (e.g. the ``existing_param``
                in update actions whose type matches the resource struct).
        """
        import msgspec

        _skip = skip_params or set()
        sig = inspect.signature(handler)
        for param in sig.parameters.values():
            if param.name in _skip:
                continue
            ann = param.annotation
            if ann is inspect.Parameter.empty:
                continue
            # Unwrap Annotated[T, ...] → T
            ann, _ = unwrap_annotated(ann)
            if isinstance(ann, type) and issubclass(ann, msgspec.Struct):
                return ann.__name__
            # Pydantic BaseModel check
            if isinstance(ann, type):
                try:
                    from pydantic import BaseModel

                    if issubclass(ann, BaseModel):
                        return ann.__name__
                except ImportError:
                    pass
        return None

    @staticmethod
    def _collect_action_body_structs(
        action: Any, *, skip_params: set[str] | None = None
    ) -> list[type]:
        """Return all ``msgspec.Struct`` types found in *action* handler params.

        Used by both ``_customize_openapi()`` (to register component schemas)
        and ``_inject_ref_metadata()`` (to inject ``x-ref-*`` extensions).

        Args:
            action: The pending action dataclass.
            skip_params: Parameter names to skip (e.g. the ``existing_param``
                in update actions whose type matches the resource struct).
        """
        import msgspec

        _skip = skip_params or set()
        structs: list[type] = []
        sig = inspect.signature(action.handler)
        for param in sig.parameters.values():
            if param.name in _skip:
                continue
            ann = param.annotation
            if ann is inspect.Parameter.empty:
                continue
            ann, _ = unwrap_annotated(ann)
            if isinstance(ann, type) and issubclass(ann, msgspec.Struct):
                structs.append(ann)
        return structs

    @staticmethod
    def _extract_handler_ref_map(handler: Any) -> dict[str, dict[str, str]]:
        """Scan *handler* parameter annotations for ``Ref`` / ``RefRevision``
        markers and return a mapping of ``{param_name: x-ref-* extensions}``.

        This enables path / query / inline-body parameters annotated with
        ``Annotated[str, Ref(...)]`` to carry ``x-ref-resource``, ``x-ref-type``,
        and ``x-ref-on-delete`` metadata into the OpenAPI extension so the web
        generator can render them as RefSelect / RefRevisionSelect.
        """
        ref_map: dict[str, dict[str, str]] = {}
        sig = inspect.signature(handler)
        for param in sig.parameters.values():
            ann = param.annotation
            if ann is inspect.Parameter.empty:
                continue
            _, metadata = unwrap_annotated(ann)
            for meta in metadata:
                if isinstance(meta, Ref):
                    ext: dict[str, str] = {
                        "x-ref-resource": meta.resource,
                        "x-ref-type": meta.ref_type.value,
                    }
                    if meta.ref_type == RefType.resource_id:
                        ext["x-ref-on-delete"] = meta.on_delete.value
                    ref_map[param.name] = ext
                    break
                if isinstance(meta, RefRevision):
                    ref_map[param.name] = {
                        "x-ref-resource": meta.resource,
                        "x-ref-type": "revision_id",
                    }
                    break
        return ref_map

    def _inject_custom_create_actions(self, schema: dict) -> None:
        """Inject ``x-autocrud-custom-create-actions`` top-level extension.

        Groups all registered create actions by resource name and writes a
        lookup table into the OpenAPI schema so the web generator can
        discover custom create actions for each resource.
        """
        if not self._pending_create_actions:
            return

        from collections import defaultdict

        actions_by_resource: dict[str, list[dict]] = defaultdict(list)
        for action in self._pending_create_actions:
            if action.resource_name not in self.resource_managers:
                continue
            # Strip leading slash from action.path to avoid double-slash
            # when the user writes path="/{name}/new".
            action_path_segment = action.path.lstrip("/")
            info: dict[str, str] = {
                "path": f"/{action.resource_name}/{action_path_segment}",
                "label": action.label,
                "operationId": action.handler.__name__,
            }
            body_schema = self._get_body_schema_name(action.handler)
            if body_schema:
                info["bodySchema"] = body_schema
            # Expose path / query parameters from the generated spec
            # so the frontend generator can produce form fields for them.
            paths = schema.get("paths", {})
            operation_path = f"/{action.resource_name}/{action_path_segment}"
            # Try exact match first; fall back to suffix match when the
            # routes live under a prefixed APIRouter (e.g. /api/v1/...).
            path_item = paths.get(operation_path, {})
            if not path_item:
                suffix = operation_path
                for spec_path, spec_item in paths.items():
                    if spec_path.endswith(suffix) and "post" in spec_item:
                        path_item = spec_item
                        break
            operation = path_item.get("post", {})
            parameters = operation.get("parameters", [])
            pp = [
                {
                    "name": p["name"],
                    "required": p.get("required", True),
                    "schema": p.get("schema", {}),
                }
                for p in parameters
                if p.get("in") == "path"
            ]
            qp = [
                {
                    "name": p["name"],
                    "required": p.get("required", False),
                    "schema": p.get("schema", {}),
                }
                for p in parameters
                if p.get("in") == "query"
            ]
            # Inject x-ref-* metadata from handler annotations into
            # path / query param schemas so the frontend generator can
            # render RefSelect / RefRevisionSelect for these params.
            ref_map = self._extract_handler_ref_map(action.handler)
            for param_list in (pp, qp):
                for p in param_list:
                    ref_ext = ref_map.get(p["name"])
                    if ref_ext:
                        p["schema"].update(ref_ext)
            if pp:
                info["pathParams"] = pp
            if qp:
                info["queryParams"] = qp
            # Extract inline body params and file params from the request body
            # schema.  This works both when bodySchema is set (mixed case with
            # additional Body(embed=True) / UploadFile params) and when there is
            # no body schema (pure compositional case).
            # When UploadFile is present, FastAPI uses multipart/form-data
            # instead of application/json, so we check both content types.
            content = operation.get("requestBody", {}).get("content", {})
            rb = content.get("application/json", {}).get("schema", {})
            # Fall back to multipart/form-data (when UploadFile is present)
            if not rb:
                rb = content.get("multipart/form-data", {}).get("schema", {})
            # Resolve $ref to components/schemas
            if "$ref" in rb:
                ref_name = rb["$ref"].split("/")[-1]
                rb = schema.get("components", {}).get("schemas", {}).get(ref_name, {})
            props: dict = rb.get("properties", {})
            required_list: list = rb.get("required", [])
            # When bodySchema is set, identify the property that IS the body
            # schema (via $ref or allOf.$ref) so we can exclude it from inline
            # params and avoid duplication.
            body_schema_prop_names: set[str] = set()
            if body_schema:
                for pname, pschema in props.items():
                    ref_target = pschema.get("$ref", "")
                    if not ref_target and "allOf" in pschema:
                        for item in pschema["allOf"]:
                            if "$ref" in item:
                                ref_target = item["$ref"]
                                break
                    if ref_target and ref_target.split("/")[-1] == body_schema:
                        body_schema_prop_names.add(pname)
                    # Also match inline schemas whose title equals the
                    # body schema name (happens when Pydantic params are
                    # replaced with untyped Body for multipart/form-data).
                    elif (
                        not ref_target
                        and pschema.get("title") == body_schema
                        and pschema.get("type") == "object"
                    ):
                        body_schema_prop_names.add(pname)
            # Record the handler parameter name for the body schema so
            # the frontend generator can build the correct FormData /
            # JSON body key when mixing body-schema + other param types.
            if body_schema_prop_names:
                info["bodySchemaParamName"] = next(iter(body_schema_prop_names))
            # Separate file params (format=binary) from inline body params
            file_params: list[dict] = []
            inline_params: list[dict] = []
            for pname, pschema in props.items():
                if pname in body_schema_prop_names:
                    continue  # Skip the body schema field itself
                if pschema.get("format") == "binary":
                    file_params.append(
                        {
                            "name": pname,
                            "required": pname in required_list,
                            "schema": {
                                "type": pschema.get("type", "string"),
                                "format": "binary",
                            },
                        }
                    )
                else:
                    inline_params.append(
                        {
                            "name": pname,
                            "required": pname in required_list,
                            "schema": pschema,
                        }
                    )
            # Inject x-ref-* into inline body param schemas
            for p in inline_params:
                ref_ext = ref_map.get(p["name"])
                if ref_ext:
                    p["schema"].update(ref_ext)
            if inline_params:
                info["inlineBodyParams"] = inline_params
            if file_params:
                info["fileParams"] = file_params
            # Async create-action metadata
            if action.async_mode is not None:
                info["asyncMode"] = action.async_mode
                if action.async_mode == "job":
                    from autocrud.crud.async_job_builder import (
                        derive_job_resource_name,
                    )

                    info["jobResourceName"] = (
                        action.job_name
                        or derive_job_resource_name(action.path, action.resource_name)
                    )
            # Warn when two actions for the same resource share the same label —
            # duplicate labels cause frontend key collisions and confuse users.
            existing_labels = {
                a["label"] for a in actions_by_resource[action.resource_name]
            }
            if action.label in existing_labels:
                warnings.warn(
                    f"Resource '{action.resource_name}' already has a create action "
                    f"with label '{action.label}' "
                    f"(duplicate handler: '{action.handler.__name__}'). "
                    f"Duplicate labels will cause frontend key collisions.",
                    stacklevel=2,
                )
            actions_by_resource[action.resource_name].append(info)

        if actions_by_resource:
            schema["x-autocrud-custom-create-actions"] = dict(actions_by_resource)

    def _inject_custom_update_actions(self, schema: dict) -> None:
        """Inject ``x-autocrud-custom-update-actions`` top-level extension.

        Groups all registered update actions by resource name and writes a
        lookup table into the OpenAPI schema so the web generator can
        discover custom update actions for each resource.
        """
        if not self._pending_update_actions:
            return

        from collections import defaultdict

        actions_by_resource: dict[str, list[dict]] = defaultdict(list)
        for action in self._pending_update_actions:
            if action.resource_name not in self.resource_managers:
                continue
            action_path_segment = action.path.lstrip("/")
            info: dict[str, Any] = {
                "path": f"/{action.resource_name}/{{resource_id}}/{action_path_segment}",
                "label": action.label,
                "operationId": action.handler.__name__,
                "mode": action.mode,
            }
            body_schema = self._get_body_schema_name(
                action.handler,
                skip_params={
                    action.existing_param,
                    action.info_param,
                    action.meta_param,
                },
            )
            if body_schema:
                info["bodySchema"] = body_schema
            # Expose path / query parameters from the generated spec
            paths = schema.get("paths", {})
            operation_path = (
                f"/{action.resource_name}/{{resource_id}}/{action_path_segment}"
            )
            path_item = paths.get(operation_path, {})
            if not path_item:
                suffix = operation_path
                for spec_path, spec_item in paths.items():
                    if spec_path.endswith(suffix) and "post" in spec_item:
                        path_item = spec_item
                        break
            operation = path_item.get("post", {})
            parameters = operation.get("parameters", [])
            pp = [
                {
                    "name": p["name"],
                    "required": p.get("required", True),
                    "schema": p.get("schema", {}),
                }
                for p in parameters
                if p.get("in") == "path" and p["name"] != "resource_id"
            ]
            qp = [
                {
                    "name": p["name"],
                    "required": p.get("required", False),
                    "schema": p.get("schema", {}),
                }
                for p in parameters
                if p.get("in") == "query"
            ]
            # Inject x-ref-* metadata
            ref_map = self._extract_handler_ref_map(action.handler)
            for param_list in (pp, qp):
                for p in param_list:
                    ref_ext = ref_map.get(p["name"])
                    if ref_ext:
                        p["schema"].update(ref_ext)
            if pp:
                info["pathParams"] = pp
            if qp:
                info["queryParams"] = qp
            # Extract inline body params and file params
            content = operation.get("requestBody", {}).get("content", {})
            rb = content.get("application/json", {}).get("schema", {})
            if not rb:
                rb = content.get("multipart/form-data", {}).get("schema", {})
            if "$ref" in rb:
                ref_name = rb["$ref"].split("/")[-1]
                rb = schema.get("components", {}).get("schemas", {}).get(ref_name, {})
            props: dict = rb.get("properties", {})
            required_list: list = rb.get("required", [])
            body_schema_prop_names: set[str] = set()
            if body_schema:
                for pname, pschema in props.items():
                    ref_target = pschema.get("$ref", "")
                    if not ref_target and "allOf" in pschema:
                        for item in pschema["allOf"]:
                            if "$ref" in item:
                                ref_target = item["$ref"]
                                break
                    if ref_target and ref_target.split("/")[-1] == body_schema:
                        body_schema_prop_names.add(pname)
                    elif (
                        not ref_target
                        and pschema.get("title") == body_schema
                        and pschema.get("type") == "object"
                    ):
                        body_schema_prop_names.add(pname)
            if body_schema_prop_names:
                info["bodySchemaParamName"] = next(iter(body_schema_prop_names))
            file_params: list[dict] = []
            inline_params: list[dict] = []
            for pname, pschema in props.items():
                if pname in body_schema_prop_names:
                    continue
                if pschema.get("format") == "binary":
                    file_params.append(
                        {
                            "name": pname,
                            "required": pname in required_list,
                            "schema": {
                                "type": pschema.get("type", "string"),
                                "format": "binary",
                            },
                        }
                    )
                else:
                    inline_params.append(
                        {
                            "name": pname,
                            "required": pname in required_list,
                            "schema": pschema,
                        }
                    )
            for p in inline_params:
                ref_ext = ref_map.get(p["name"])
                if ref_ext:
                    p["schema"].update(ref_ext)
            if inline_params:
                info["inlineBodyParams"] = inline_params
            if file_params:
                info["fileParams"] = file_params
            # Async update-action metadata
            if action.async_mode is not None:
                info["asyncMode"] = action.async_mode
                if action.async_mode == "job":
                    from autocrud.crud.async_job_builder import (
                        derive_job_resource_name,
                    )

                    info["jobResourceName"] = (
                        action.job_name
                        or derive_job_resource_name(action.path, action.resource_name)
                    )
            # Warn on duplicate labels
            existing_labels = {
                a["label"] for a in actions_by_resource[action.resource_name]
            }
            if action.label in existing_labels:
                warnings.warn(
                    f"Resource '{action.resource_name}' already has an update action "
                    f"with label '{action.label}' "
                    f"(duplicate handler: '{action.handler.__name__}'). "
                    f"Duplicate labels will cause frontend key collisions.",
                    stacklevel=2,
                )
            actions_by_resource[action.resource_name].append(info)

        if actions_by_resource:
            schema["x-autocrud-custom-update-actions"] = dict(actions_by_resource)

    def _inject_async_create_jobs(self, schema: dict) -> None:
        """Inject ``x-autocrud-async-create-jobs`` top-level extension.

        Maps each auto-generated job resource name to its parent resource
        name so the frontend generator can build sidebar accordions and
        navigation links.
        """
        if not self._async_job_registry:
            return

        mapping: dict[str, str] = {}
        for (
            job_resource_name,
            _job_model,
            target_rm,
            _auto_payload_type,
            _param_conversions,
        ) in self._async_job_registry.values():
            mapping[job_resource_name] = target_rm.resource_name

        if mapping:
            schema["x-autocrud-async-create-jobs"] = mapping

    def _inject_async_update_jobs(self, schema: dict) -> None:
        """Inject ``x-autocrud-async-update-jobs`` top-level extension.

        Maps each auto-generated update-job resource name to its parent
        resource name so the frontend generator can build pending-job
        accordions on the detail page.
        """
        if not self._async_update_job_registry:
            return

        mapping: dict[str, str] = {}
        for (
            job_resource_name,
            _job_model,
            target_rm,
            _auto_payload_type,
            _param_conversions,
            _update_mode,
            _existing_param,
            _info_param,
            _meta_param,
        ) in self._async_update_job_registry.values():
            mapping[job_resource_name] = target_rm.resource_name

        if mapping:
            schema["x-autocrud-async-update-jobs"] = mapping

    def _inject_indexed_fields(self, schema: dict) -> None:
        """Inject ``x-autocrud-indexed-fields`` top-level extension.

        Maps each resource name to its list of indexed field paths so the
        web generator can populate ``indexedFields`` in the resource config.
        This enables the frontend to distinguish server-sortable/filterable
        fields from client-only fields.
        """
        mapping: dict[str, list[str]] = {}
        for name, rm in self.resource_managers.items():
            indexed = rm.indexed_fields
            if indexed:
                mapping[name] = [f.field_path for f in indexed]

        if mapping:
            schema["x-autocrud-indexed-fields"] = mapping

    def _install_ref_integrity_handlers(self) -> None:
        """Install event handlers for referential integrity (cascade / set_null).

        For each registered resource that is a *target* of a ``Ref`` with
        ``on_delete`` of ``cascade`` or ``set_null``, this method registers a
        ``_RefIntegrityHandler`` on the target's ``ResourceManager`` so that
        when the target is deleted the referencing resources are automatically
        updated.
        """
        registered = set(self.resource_managers.keys())

        # Build a mapping: target_resource -> list of actionable refs
        from collections import defaultdict

        target_refs: dict[str, list[_RefInfo]] = defaultdict(list)
        for ref_info in self.relationships:
            if (
                ref_info.on_delete != OnDelete.dangling
                and ref_info.target in registered
                and ref_info.source in registered
            ):
                target_refs[ref_info.target].append(ref_info)

        for target_name, refs in target_refs.items():
            handler = _RefIntegrityHandler(
                refs=refs,
                resource_managers=self.resource_managers,
            )
            target_rm = self.resource_managers[target_name]
            target_rm.event_handlers.append(handler)

    def apply(
        self,
        app: FastAPI | APIRouter,
        *,
        router: APIRouter | None = None,
        structs: list[type] | None = None,
        auto_include: bool = True,
    ) -> APIRouter:
        """Apply all route templates to generate API endpoints.

        This method generates all the CRUD endpoints for all registered models.
        When ``app`` is a :class:`~fastapi.FastAPI` instance, the OpenAPI schema
        is automatically customised via :meth:`openapi` after route generation.

        Args:
            app: The FastAPI application or an APIRouter to attach routes to.
                When a ``FastAPI`` instance is provided, :meth:`openapi` is
                called automatically after route generation.
            router: Optional sub-router.  When provided, routes are generated
                on this router instead of directly on ``app``.  If
                ``auto_include`` is ``True`` and ``app`` is a ``FastAPI``
                instance, the router is automatically included on ``app``
                via ``app.include_router(router)`` before OpenAPI generation.
            structs: Additional ``msgspec.Struct`` types to include in the
                OpenAPI ``components/schemas``.  Forwarded to :meth:`openapi`.
            auto_include: When ``True`` (the default) and both ``app`` is a
                ``FastAPI`` instance and ``router`` is provided, automatically
                call ``app.include_router(router)`` so that the sub-router's
                routes are reachable and visible in the OpenAPI schema.
                Set to ``False`` if you have already called
                ``app.include_router(router)`` yourself.

        Returns:
            The router that routes were generated on — either ``router``
            (if provided) or ``app``.

        Example:
            ```python
            from fastapi import FastAPI, APIRouter
            from autocrud import AutoCRUD

            app = FastAPI()
            autocrud = AutoCRUD()
            autocrud.add_model(User)
            autocrud.add_model(Post)

            # 1. Simplest — routes on app, auto OpenAPI
            autocrud.apply(app)

            # 2. With a sub-router — auto include + auto OpenAPI
            api_router = APIRouter(prefix="/api/v1")
            autocrud.apply(app, router=api_router)

            # 3. Manual include (e.g. already included elsewhere)
            api_router = APIRouter(prefix="/api/v1")
            autocrud.apply(app, router=api_router, auto_include=False)
            app.include_router(api_router)
            autocrud.openapi(app)

            # 4. Pure APIRouter (no FastAPI, no OpenAPI)
            api_router = APIRouter(prefix="/api/v1")
            autocrud.apply(api_router)
            ```

        Note:
            - Call this method after adding all models and custom route templates.
            - When ``app`` is a bare ``APIRouter``, OpenAPI customisation is
              skipped (``APIRouter`` has no OpenAPI schema).
            - ``structs`` is ignored when ``app`` is not a ``FastAPI`` instance.
        """
        # Determine the target router for route generation
        target = router if router is not None else app

        # Validate all Ref targets point to registered resources
        registered = set(self.resource_managers.keys())
        for ref_info in self.relationships:
            if ref_info.target not in registered:
                logger.warning(
                    f"Ref on '{ref_info.source}.{ref_info.source_field}' targets "
                    f"resource '{ref_info.target}' which is not registered. "
                    f"The reference will be dangling at runtime."
                )

        # Install referential integrity event handlers
        self._install_ref_integrity_handlers()

        # Auto-register Job models for async create actions BEFORE applying
        # route templates so the Jobs get their own CRUD endpoints.
        self._register_async_job_models()

        # Auto-register Job models for async update actions.
        self._register_async_update_job_models()

        self.route_templates.sort()
        for model_name, resource_manager in self.resource_managers.items():
            for route_template in self.route_templates:
                try:
                    route_template.apply(model_name, resource_manager, target)
                except Exception:
                    pass

        # Register custom create action routes
        self._apply_create_actions(target)

        # Register custom update action routes
        self._apply_update_actions(target)

        # Add ref-specific routes (referrers + relationships)
        self._apply_ref_routes(target)

        # Global backup / restore endpoints
        self._apply_backup_routes(target)

        # Auto include_router + auto openapi when app is a FastAPI instance
        is_fastapi = isinstance(app, FastAPI)
        if is_fastapi:
            if router is not None and auto_include:
                app.include_router(router)
            # Only generate OpenAPI when routes are actually on the app.
            # When router is provided but auto_include is False, the routes
            # live on the sub-router and are not yet reachable from app.routes,
            # so skip openapi and let the user call it manually.
            if router is None or auto_include:
                self.openapi(app, structs or [])

        return target

    @staticmethod
    def _reconstruct_params(
        kwargs: dict, conversions: dict[str, tuple[str, type]]
    ) -> dict:
        """Reconstruct original param types from serialised surrogates.

        Called inside the job handler after unpacking the auto-payload.

        Conversion kinds:

        * ``'upload_file'`` — :class:`UploadFilePayload` →
          ``starlette.datastructures.UploadFile``
        * ``'pydantic'`` — msgspec Struct → original Pydantic ``BaseModel``
        * ``'to_str'`` — ``str`` → attempt ``original_type(str_value)``
        """
        import io

        import msgspec as _ms

        from autocrud.crud.async_job_builder import UploadFilePayload

        for field_name, (conv_kind, orig_type) in conversions.items():
            if field_name not in kwargs:
                continue
            val = kwargs[field_name]

            if conv_kind == "upload_file" and isinstance(val, UploadFilePayload):
                from starlette.datastructures import Headers
                from starlette.datastructures import UploadFile as _StarletteUpload

                binary = val.binary
                data = (
                    binary.data if not isinstance(binary.data, _ms.UnsetType) else b""
                )
                ct = (
                    binary.content_type
                    if not isinstance(binary.content_type, _ms.UnsetType)
                    else "application/octet-stream"
                )
                sz = binary.size if not isinstance(binary.size, _ms.UnsetType) else None
                kwargs[field_name] = _StarletteUpload(
                    file=io.BytesIO(data),
                    filename=val.filename,
                    size=sz,
                    headers=Headers({"content-type": ct}),
                )
            elif conv_kind == "pydantic":
                kwargs[field_name] = orig_type.model_validate(_ms.to_builtins(val))
            elif conv_kind == "to_str":
                try:
                    kwargs[field_name] = orig_type(val)
                except Exception:
                    pass  # keep as str if reconstruction fails

        return kwargs

    def _register_async_job_models(self) -> None:
        """Auto-register Job models for ``async_mode='job'`` create actions.

        For each pending create action with ``async_mode='job'``:

        1. Inspects the handler to find the body parameter's Struct type.
        2. Creates a dynamic ``Job[PayloadType, dict]`` subclass.
        3. Wraps the user handler to auto-create the target resource.
        4. Registers the Job model via ``add_model()`` with a message queue.

        The mapping ``_async_job_registry`` is populated so that
        ``_apply_create_actions()`` can create the correct endpoint handlers.
        """
        import msgspec as _msgspec

        from autocrud.crud.async_job_builder import (
            build_async_job_model,
            build_auto_payload_struct,
            derive_job_resource_name,
            resolve_payload_field_type,
        )

        # Registry: action handler id → (job_resource_name, job_model, target_rm)
        self._async_job_registry: dict[int, tuple[str, type, ResourceManager]] = {}

        for action in self._pending_create_actions:
            if action.async_mode != "job":
                continue

            target_rm = self.resource_managers.get(action.resource_name)
            if target_rm is None:
                logger.warning(
                    f"async create_action '{action.path}' targets resource "
                    f"'{action.resource_name}' which is not registered. Skipping."
                )
                continue

            # Discover the handler's body parameter type (first Struct param)
            sig = inspect.signature(action.handler)
            payload_type = None
            auto_payload_type = None
            for name, param in sig.parameters.items():
                ann = param.annotation
                if ann is inspect.Parameter.empty:
                    continue
                raw_ann, _ = unwrap_annotated(ann)
                if isinstance(raw_ann, type) and issubclass(raw_ann, _msgspec.Struct):
                    payload_type = raw_ann
                    break

            # If no explicit Struct param, auto-generate a payload Struct
            # from the handler's parameters.  Non-msgspec types (UploadFile,
            # Pydantic BaseModel, pydantic_core.Url, etc.) are converted to
            # serialisable equivalents via resolve_payload_field_type().
            if payload_type is None:
                param_fields: list[tuple[str, type]] = []
                param_conversions: dict[str, tuple[str, type]] = {}

                for pname, param in sig.parameters.items():
                    ann = param.annotation
                    if ann is inspect.Parameter.empty:
                        continue
                    raw_ann, _ = unwrap_annotated(ann)

                    ser_type, conv_kind = resolve_payload_field_type(raw_ann)
                    param_fields.append((pname, ser_type))
                    if conv_kind is not None:
                        param_conversions[pname] = (conv_kind, raw_ann)

                if not param_fields:
                    logger.warning(
                        f"async create_action '{action.path}' handler has no "
                        f"parameters. Cannot generate Job model. "
                        f"Falling back to sync."
                    )
                    action.async_mode = None
                    continue

                # Build a dynamic payload Struct from the scalar params
                auto_payload_type = build_auto_payload_struct(
                    action_name=action.path,
                    resource_name=action.resource_name,
                    param_fields=param_fields,
                )
                payload_type = auto_payload_type
            else:
                param_conversions = {}

            # Build dynamic Job model
            job_model = build_async_job_model(
                action_name=action.path,
                resource_name=action.resource_name,
                payload_type=payload_type,
            )
            job_resource_name = action.job_name or derive_job_resource_name(
                action.path, action.resource_name
            )

            # Build the job handler that wraps the user's create-action handler
            original_handler = action.handler
            target_resource_manager = target_rm
            _is_auto_payload = auto_payload_type is not None

            def _make_job_handler(
                handler,
                trm,
                *,
                auto_payload=False,
                param_conversions=None,
                resource_managers=None,
                job_resource_name=None,
            ):
                """Create a closure to capture the right handler and target RM.

                Args:
                    handler: The user's create-action function.
                    trm: The target resource's ResourceManager.
                    auto_payload: When ``True`` the payload is an auto-generated
                        Struct whose fields mirror the handler's parameters.
                        The handler is called with ``**kwargs`` instead of a
                        single positional Struct argument.
                    param_conversions: Mapping of field name →
                        ``(conv_kind, original_type)`` for fields that need
                        reconstruction from their serialisable surrogates.
                    resource_managers: The ``AutoCRUD.resource_managers`` dict.
                        Used at job-execution time to restore ``Binary`` data
                        from the blob store before reconstructing params.
                    job_resource_name: Name of the job resource, used together
                        with *resource_managers* to locate the job RM.
                """
                _is_async = inspect.iscoroutinefunction(handler)
                _conversions = param_conversions or {}
                _has_binary_conv = any(
                    k == "upload_file" for k, _ in _conversions.values()
                )
                _rms = resource_managers
                _jrn = job_resource_name

                def job_handler(resource, job_context=None):
                    payload = resource.data.payload
                    if auto_payload:
                        # Restore Binary.data from blob store when the
                        # payload contains UploadFilePayload fields whose
                        # Binary.data was stripped during storage.
                        if _has_binary_conv and _rms and _jrn:
                            jrm = _rms.get(_jrn)
                            if jrm is not None:
                                restored = jrm.restore_binary(resource.data)
                                payload = restored.payload

                        kwargs = {
                            f: getattr(payload, f) for f in payload.__struct_fields__
                        }
                        # Reconstruct original types from serialised surrogates
                        if _conversions:
                            kwargs = AutoCRUD._reconstruct_params(kwargs, _conversions)
                        raw_result = handler(**kwargs)
                    else:
                        raw_result = handler(payload)

                    # Handle async handlers called from the sync MQ consumer
                    if _is_async:
                        result = asyncio.run(raw_result)
                    else:
                        result = raw_result

                    if result is not None:
                        # Propagate the user who created the Job to the
                        # target resource so created_by is traceable.
                        _job_user = resource.info.created_by or "system"
                        with trm.using(_job_user, dt.datetime.now()):
                            info = trm.create(result)
                        # Store RevisionInfo as artifact (dict form for
                        # serialisability)
                        artifact = {
                            "resource_id": info.resource_id,
                            "revision_id": info.revision_id,
                            "resource_name": trm.resource_name,
                        }
                        if job_context is not None:
                            job_context.set_artifact(artifact)
                        else:
                            resource.data.artifact = artifact

                return job_handler

            wrapped_handler = _make_job_handler(
                original_handler,
                target_resource_manager,
                auto_payload=_is_auto_payload,
                param_conversions=param_conversions if _is_auto_payload else None,
                resource_managers=self.resource_managers,
                job_resource_name=job_resource_name,
            )

            # Register the Job model with add_model (includes MQ setup)
            self.add_model(
                job_model,
                name=job_resource_name,
                job_handler=wrapped_handler,
            )

            # Store reference for _apply_create_actions
            self._async_job_registry[id(action.handler)] = (
                job_resource_name,
                job_model,
                target_rm,
                auto_payload_type,  # None for explicit Struct, else the auto type
                param_conversions if _is_auto_payload else None,
            )

            # Populate reverse mapping on target RM so that
            # target_rm.start_consume(custom_creation=...) can locate jobs.
            job_rm = self.resource_managers[job_resource_name]
            target_rm.register_async_create_job(job_resource_name, job_rm)

    def _register_async_update_job_models(self) -> None:
        """Auto-register Job models for ``async_mode='job'`` update actions.

        For each pending update action with ``async_mode='job'``:

        1. Inspects the handler to find the body parameter's Struct type
           (skipping the *existing_param*, *info_param*, *meta_param*
           parameters which are auto-injected at runtime).
        2. Creates a dynamic payload Struct that includes a ``resource_id``
           field alongside the handler's body fields.
        3. Creates a dynamic ``Job[PayloadType, dict]`` subclass via
           :func:`build_async_update_job_model`.
        4. Wraps the user handler so that the job consumer lazy-fetches
           the existing resource, calls the handler, and then calls
           ``rm.update()`` or ``rm.modify()`` based on the action's mode.
        5. Registers the Job model via ``add_model()`` with a message queue.

        The mapping ``_async_update_job_registry`` is populated so that
        ``_apply_update_actions()`` can create the correct endpoint handlers.
        """
        import msgspec as _msgspec

        from autocrud.crud.async_job_builder import (
            build_async_update_job_model,
            build_auto_payload_struct,
            derive_job_resource_name,
            resolve_payload_field_type,
        )

        # Registry: action handler id → (job_resource_name, job_model, target_rm,
        #   auto_payload_type, param_conversions, update_mode, existing_param,
        #   info_param, meta_param)
        self._async_update_job_registry: dict[int, tuple] = {}

        for action in self._pending_update_actions:
            if action.async_mode != "job":
                continue

            target_rm = self.resource_managers.get(action.resource_name)
            if target_rm is None:
                logger.warning(
                    f"async update_action '{action.path}' targets resource "
                    f"'{action.resource_name}' which is not registered. Skipping."
                )
                continue

            # Discover the handler's body parameter type (first Struct param),
            # skipping auto-injected params (existing, info, meta).
            skip_params = {action.existing_param, action.info_param, action.meta_param}
            sig = inspect.signature(action.handler)
            payload_type = None
            auto_payload_type = None
            for name, param in sig.parameters.items():
                if name in skip_params:
                    continue
                ann = param.annotation
                if ann is inspect.Parameter.empty:
                    continue
                raw_ann, _ = unwrap_annotated(ann)
                if isinstance(raw_ann, type) and issubclass(raw_ann, _msgspec.Struct):
                    payload_type = raw_ann
                    break

            # If no explicit Struct param, auto-generate a payload Struct
            # from the handler's parameters (excluding skip_params).
            # Always prepend a resource_id: str field.
            if payload_type is None:
                param_fields: list[tuple[str, type]] = []
                param_conversions: dict[str, tuple[str, type]] = {}

                for pname, param in sig.parameters.items():
                    if pname in skip_params:
                        continue
                    ann = param.annotation
                    if ann is inspect.Parameter.empty:
                        continue
                    raw_ann, _ = unwrap_annotated(ann)

                    ser_type, conv_kind = resolve_payload_field_type(raw_ann)
                    param_fields.append((pname, ser_type))
                    if conv_kind is not None:
                        param_conversions[pname] = (conv_kind, raw_ann)

                if not param_fields:
                    # No body params — still need resource_id in payload
                    pass

                # Build a dynamic payload Struct with resource_id prepended
                auto_payload_type = build_auto_payload_struct(
                    action_name=action.path,
                    resource_name=action.resource_name,
                    param_fields=param_fields,
                    extra_fields=[("resource_id", str)],
                )
                payload_type = auto_payload_type
            else:
                # Explicit Struct param — wrap it with resource_id
                # Create a new Struct: { resource_id: str, payload: OriginalStruct }
                wrapper_fields = [
                    ("resource_id", str),
                    ("payload_data", payload_type),
                ]
                clean_name = action.path.replace("-", " ").title().replace(" ", "")
                resource_pascal = (
                    action.resource_name.replace("-", " ").title().replace(" ", "")
                )
                wrapper_name = f"{clean_name}{resource_pascal}UpdatePayload"
                auto_payload_type = _msgspec.defstruct(wrapper_name, wrapper_fields)
                payload_type = auto_payload_type
                param_conversions = {}

            # Build dynamic Job model
            job_model = build_async_update_job_model(
                action_name=action.path,
                resource_name=action.resource_name,
                payload_type=payload_type,
                update_mode=action.mode,
            )
            job_resource_name = action.job_name or derive_job_resource_name(
                action.path, action.resource_name
            )

            # Build the job handler that wraps the user's update-action handler
            original_handler = action.handler
            target_resource_manager = target_rm
            _is_auto_payload = auto_payload_type is not None
            _update_mode = action.mode
            _existing_param = action.existing_param
            _info_param = action.info_param
            _meta_param = action.meta_param

            def _make_update_job_handler(
                handler,
                trm,
                *,
                auto_payload=False,
                has_explicit_struct=False,
                param_conversions=None,
                resource_managers=None,
                job_resource_name=None,
                update_mode="update",
                existing_param="existing",
                info_param="info",
                meta_param="meta",
            ):
                """Create a closure for the update-action job handler.

                At job-execution time:
                1. Extracts ``resource_id`` from the payload.
                2. Lazy-fetches existing resource via ``trm.get(resource_id)``.
                3. Injects existing/info/meta into handler kwargs.
                4. Calls the user handler.
                5. Calls ``trm.update()`` or ``trm.modify()`` based on mode.
                6. Stores the artifact.

                Args:
                    handler: The user's update-action function.
                    trm: The target resource's ResourceManager.
                    auto_payload: When ``True`` the payload fields mirror
                        the handler's parameters (plus ``resource_id``).
                    has_explicit_struct: When ``True`` the original handler
                        had an explicit Struct body param, and the payload
                        wraps it as ``payload_data``.
                    param_conversions: For type reconstruction.
                    resource_managers: For binary restoration.
                    job_resource_name: Name of the job resource.
                    update_mode: ``"update"`` or ``"modify"``.
                    existing_param: Handler param name for existing data.
                    info_param: Handler param name for RevisionInfo.
                    meta_param: Handler param name for ResourceMeta.
                """
                _is_async = inspect.iscoroutinefunction(handler)
                _conversions = param_conversions or {}
                _has_binary_conv = any(
                    k == "upload_file" for k, _ in _conversions.values()
                )
                _rms = resource_managers
                _jrn = job_resource_name
                _sig = inspect.signature(handler)
                _has_existing = existing_param in _sig.parameters
                _has_info = info_param in _sig.parameters
                _has_meta = meta_param in _sig.parameters

                def job_handler(resource, job_context=None):
                    payload = resource.data.payload
                    # Extract resource_id from payload
                    _resource_id = payload.resource_id

                    if has_explicit_struct:
                        # Payload wraps the original Struct as payload_data
                        kwargs = {
                            next(
                                (
                                    n
                                    for n in _sig.parameters
                                    if n not in {existing_param, info_param, meta_param}
                                    and isinstance(_sig.parameters[n].annotation, type)
                                ),
                                "body",
                            ): payload.payload_data
                        }
                    elif auto_payload:
                        # Restore Binary.data from blob store if needed
                        if _has_binary_conv and _rms and _jrn:
                            jrm = _rms.get(_jrn)
                            if jrm is not None:
                                restored = jrm.restore_binary(resource.data)
                                payload = restored.payload

                        kwargs = {
                            f: getattr(payload, f)
                            for f in payload.__struct_fields__
                            if f != "resource_id"
                        }
                        if _conversions:
                            kwargs = AutoCRUD._reconstruct_params(kwargs, _conversions)
                    else:
                        kwargs = {}

                    # Lazy-fetch existing resource
                    _job_user = resource.info.created_by or "system"
                    with trm.using(_job_user, dt.datetime.now()):
                        existing_resource = trm.get(_resource_id)
                    if _has_existing:
                        kwargs[existing_param] = existing_resource.data
                    if _has_info:
                        kwargs[info_param] = existing_resource.info
                    if _has_meta:
                        with trm.using(_job_user, dt.datetime.now()):
                            kwargs[meta_param] = trm.get_meta(_resource_id)

                    raw_result = handler(**kwargs)

                    # Handle async handlers called from the sync MQ consumer
                    if _is_async:
                        result = asyncio.run(raw_result)
                    else:
                        result = raw_result

                    if result is not None:
                        with trm.using(_job_user, dt.datetime.now()):
                            if update_mode == "modify":
                                info = trm.modify(_resource_id, data=result)
                            else:
                                info = trm.update(_resource_id, result)
                        artifact = {
                            "resource_id": info.resource_id,
                            "revision_id": info.revision_id,
                            "resource_name": trm.resource_name,
                        }
                        if job_context is not None:
                            job_context.set_artifact(artifact)
                        else:
                            resource.data.artifact = artifact

                return job_handler

            # Detect if the handler had an explicit Struct param
            _has_explicit_struct = False
            for _pname, _pparam in sig.parameters.items():
                if _pname in skip_params:
                    continue
                _ann = _pparam.annotation
                if _ann is inspect.Parameter.empty:
                    continue
                _raw_ann, _ = unwrap_annotated(_ann)
                if isinstance(_raw_ann, type) and issubclass(_raw_ann, _msgspec.Struct):
                    _has_explicit_struct = True
                    break

            wrapped_handler = _make_update_job_handler(
                original_handler,
                target_resource_manager,
                auto_payload=_is_auto_payload,
                has_explicit_struct=_has_explicit_struct,
                param_conversions=param_conversions
                if not _has_explicit_struct
                else None,
                resource_managers=self.resource_managers,
                job_resource_name=job_resource_name,
                update_mode=_update_mode,
                existing_param=_existing_param,
                info_param=_info_param,
                meta_param=_meta_param,
            )

            # Register the Job model with add_model (includes MQ setup)
            self.add_model(
                job_model,
                name=job_resource_name,
                job_handler=wrapped_handler,
            )

            # Store reference for _apply_update_actions
            self._async_update_job_registry[id(action.handler)] = (
                job_resource_name,
                job_model,
                target_rm,
                auto_payload_type,
                param_conversions if not _has_explicit_struct else None,
                action.mode,
                action.existing_param,
                action.info_param,
                action.meta_param,
            )

            # Populate reverse mapping on target RM
            job_rm = self.resource_managers[job_resource_name]
            target_rm.register_async_update_job(job_resource_name, job_rm)

    def _apply_create_actions(self, router: APIRouter) -> None:
        """Register routes for all pending custom create actions."""
        import msgspec as _msgspec

        from autocrud.crud.route_templates.basic import (
            BaseRouteTemplate,
            DependencyProvider,
            MsgspecResponse,
            jsonschema_to_json_schema_extra,
            struct_to_responses_type,
        )

        # Resolve DependencyProvider: try to reuse one from existing route
        # templates so that custom create actions share the same get_user /
        # get_now dependency as standard CRUD routes.
        deps: DependencyProvider | None = None
        for rt in self.route_templates:
            if isinstance(rt, BaseRouteTemplate) and hasattr(rt, "deps"):
                deps = rt.deps
                break
        if deps is None:
            # No route templates have a DP — create one that respects
            # default_user if configured.
            deps = DependencyProvider()
            if self.default_user is not UNSET:
                deps = deps.with_default_user(self.default_user)

        def _is_msgspec_struct_type(ann: type) -> bool:
            """Check if *ann* is a msgspec.Struct subclass."""
            return isinstance(ann, type) and issubclass(ann, _msgspec.Struct)

        def _is_upload_file_annotation(ann: Any) -> bool:
            """Check if *ann* is or contains ``UploadFile``."""
            raw, _ = unwrap_annotated(ann)
            return isinstance(raw, type) and issubclass(raw, UploadFile)

        async def _convert_params_for_payload(
            kwargs: dict,
            param_convs: dict[str, tuple[str, type]],
            auto_payload_type: type | None,
        ) -> None:
            """Convert non-serialisable kwargs to their payload surrogates.

            Mutates *kwargs* in place so they can be packed into the
            auto-generated payload Struct.
            """
            from autocrud.crud.async_job_builder import UploadFilePayload
            from autocrud.types import Binary

            # Get the struct field types for Pydantic conversion targets
            _field_types: dict[str, type] = {}
            if auto_payload_type is not None:
                for fi in _msgspec.structs.fields(auto_payload_type):
                    _field_types[fi.name] = fi.type

            for field_name, (conv_kind, _orig_type) in param_convs.items():
                if field_name not in kwargs:
                    continue
                val = kwargs[field_name]

                if conv_kind == "upload_file":
                    content = await val.read()
                    kwargs[field_name] = UploadFilePayload(
                        binary=Binary(
                            data=content,
                            content_type=val.content_type,
                            size=val.size,
                        ),
                        filename=val.filename,
                    )
                elif conv_kind == "pydantic":
                    target_type = _field_types.get(field_name)
                    if target_type is not None:
                        kwargs[field_name] = _msgspec.convert(
                            val.model_dump(mode="python"), target_type
                        )
                elif conv_kind == "to_str":
                    kwargs[field_name] = str(val)

        def _build_fastapi_compatible_handler(
            handler,
            resource_manager,
            *,
            async_job_config=None,
            background_mode=False,
            deps=None,
        ):
            """Build a FastAPI-compatible endpoint function.

            The user-provided handler may use ``msgspec.Struct`` type hints on
            ``Body()`` parameters.  FastAPI cannot introspect those directly
            (it requires Pydantic), so we build a new function whose signature
            replaces Struct-annotated Body parameters with un-typed
            ``Body(json_schema_extra=...)`` — the same pattern used by
            ``CreateRouteTemplate``.  Inside the wrapper we convert the raw
            dict back to the Struct via ``msgspec.convert`` before calling
            the user handler.

            Plain scalar parameters (``str``, ``int``, etc.) without any
            FastAPI decorator are left as-is — FastAPI will treat them as
            query parameters, which is the correct behaviour.

            Args:
                handler: The user's endpoint function.
                resource_manager: The target resource's ResourceManager.
                async_job_config: When set, a
                    ``(job_rm, job_resource_name, auto_payload_type,
                    param_conversions)`` tuple that switches the wrapper
                    into async-job mode.  Instead of calling *handler* and
                    creating the target resource, the wrapper creates a Job
                    resource and returns HTTP 202 with
                    :class:`JobRedirectInfo`.  When *auto_payload_type* is
                    not ``None``, individual kwargs are packed into the
                    auto-generated payload Struct before creating the Job.
                    *param_conversions* maps field names that need
                    serialisation conversion at endpoint time.
                background_mode: When ``True``, the wrapper uses
                    FastAPI ``BackgroundTasks`` to schedule the handler
                    execution in the background.  The endpoint returns
                    HTTP 202 with :class:`BackgroundTaskAccepted`
                    immediately.  No Job model is created.
                deps: A :class:`DependencyProvider` instance used to inject
                    ``current_user`` and ``current_time`` into the wrapper
                    function signature via ``Depends()``.  When ``None`` a
                    default ``DependencyProvider()`` is created.
            """
            if deps is None:
                deps = DependencyProvider()

            sig = inspect.signature(handler)
            # Identify parameters whose annotation is a msgspec.Struct subclass
            # so we can convert them from raw dicts.
            struct_params: dict[str, type] = {}
            # Pydantic BaseModel params that need manual conversion
            # (required when UploadFile forces multipart/form-data)
            pydantic_params: dict[str, type] = {}
            new_params: list[inspect.Parameter] = []
            new_annotations: dict[str, Any] = {}

            # Pre-scan: check if UploadFile is present — forces
            # multipart/form-data encoding where complex types arrive as
            # JSON strings.
            _has_upload_file = any(
                _is_upload_file_annotation(p.annotation)
                for p in sig.parameters.values()
            )

            def _is_pydantic_model_type(ann: type) -> bool:
                """Check if *ann* is a Pydantic BaseModel subclass."""
                if not isinstance(ann, type):
                    return False
                try:
                    from pydantic import BaseModel

                    return issubclass(ann, BaseModel)
                except ImportError:
                    return False

            for name, param in sig.parameters.items():
                ann = param.annotation
                if ann is inspect.Parameter.empty:
                    new_params.append(param)
                    continue

                # Unwrap Annotated[T, Body(...)] → check T
                raw_ann, _ = unwrap_annotated(ann)

                if _is_msgspec_struct_type(raw_ann):
                    # Replace with untyped Body(json_schema_extra=...)
                    struct_params[name] = raw_ann
                    new_default = Body(
                        json_schema_extra=jsonschema_to_json_schema_extra(raw_ann),
                    )
                    new_param = param.replace(
                        annotation=inspect.Parameter.empty,
                        default=new_default,
                    )
                    new_params.append(new_param)
                elif _has_upload_file and _is_pydantic_model_type(raw_ann):
                    # When UploadFile forces multipart/form-data, Pydantic
                    # model params arrive as JSON strings.  Replace them
                    # with untyped Body() and handle conversion in the
                    # wrapper — same approach as Struct params.
                    pydantic_params[name] = raw_ann
                    try:
                        _pydantic_schema = raw_ann.model_json_schema()
                    except Exception:
                        _pydantic_schema = {}
                    new_default = Body(
                        json_schema_extra=_pydantic_schema,
                    )
                    new_param = param.replace(
                        annotation=inspect.Parameter.empty,
                        default=new_default,
                    )
                    new_params.append(new_param)
                else:
                    new_params.append(param)
                    if ann is not inspect.Parameter.empty:
                        new_annotations[name] = ann

            # Inject current_user and current_time via Depends()
            new_params.append(
                inspect.Parameter(
                    "current_user",
                    inspect.Parameter.KEYWORD_ONLY,
                    default=Depends(deps.get_user),
                    annotation=str,
                )
            )
            new_params.append(
                inspect.Parameter(
                    "current_time",
                    inspect.Parameter.KEYWORD_ONLY,
                    default=Depends(deps.get_now),
                    annotation=dt.datetime,
                )
            )
            new_annotations["current_user"] = str
            new_annotations["current_time"] = dt.datetime

            # Inject BackgroundTasks when background_mode is enabled
            if background_mode:
                from starlette.background import BackgroundTasks

                new_params.append(
                    inspect.Parameter(
                        "background_tasks",
                        inspect.Parameter.KEYWORD_ONLY,
                        annotation=BackgroundTasks,
                    )
                )
                new_annotations["background_tasks"] = BackgroundTasks

            new_sig = sig.replace(
                parameters=new_params, return_annotation=inspect.Parameter.empty
            )

            def _ensure_dict(val: Any) -> Any:
                """Parse JSON string to dict when multipart/form-data
                delivers complex fields as strings."""
                if isinstance(val, str):
                    import json as _json

                    return _json.loads(val)
                return val

            # ---- async-job mode: create Job + return 202 ----------------
            if async_job_config is not None:
                from autocrud.types import JobRedirectInfo

                job_rm, job_resource_name, auto_payload_type, _param_convs = (
                    async_job_config
                )
                _param_convs = _param_convs or {}
                # First Struct param is the Job payload (explicit Struct case)
                payload_param_name = next(iter(struct_params), None)

                async def wrapper(*args, **kwargs):
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))

                    # Convert non-serialisable params before packing
                    if _param_convs:
                        await _convert_params_for_payload(
                            kwargs, _param_convs, auto_payload_type
                        )

                    if auto_payload_type is not None:
                        # Auto-generated payload: pack individual kwargs
                        payload_data = auto_payload_type(
                            **{
                                f: kwargs[f]
                                for f in auto_payload_type.__struct_fields__
                                if f in kwargs
                            }
                        )
                    elif payload_param_name is not None:
                        # Explicit Struct parameter: use it directly
                        payload_data = kwargs.get(payload_param_name)
                    else:
                        payload_data = None

                    if payload_data is None:
                        raise HTTPException(
                            status_code=400,
                            detail="Missing payload for async create action.",
                        )

                    job_data = job_rm.resource_type(payload=payload_data)
                    with job_rm.using(_current_user, _current_time):
                        info = job_rm.create(job_data)

                    redirect_url = f"/{job_resource_name}/{info.resource_id}"
                    return MsgspecResponse(
                        JobRedirectInfo(
                            job_resource_name=job_resource_name,
                            job_resource_id=info.resource_id,
                            redirect_url=redirect_url,
                        ),
                        status_code=202,
                    )

            # ---- background mode: schedule via BackgroundTasks + 202 ----
            elif background_mode:
                from autocrud.types import BackgroundTaskAccepted

                _bg_is_async = inspect.iscoroutinefunction(handler)

                async def wrapper(*args, **kwargs):
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    _bg_tasks = kwargs.pop("background_tasks")
                    # Convert raw dicts to Struct / Pydantic instances
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))

                    # Snapshot converted kwargs for the background closure
                    _snapshot_kwargs = dict(kwargs)

                    # Always define _run_bg as a sync function so that
                    # Starlette dispatches it via ``run_in_threadpool``.
                    # This ensures the HTTP 202 response is flushed to the
                    # client *before* the background work starts.  If the
                    # original handler is async we bridge into a new event
                    # loop inside the worker thread with ``asyncio.run()``.
                    def _run_bg() -> None:
                        try:
                            if _bg_is_async:
                                result = asyncio.run(handler(*args, **_snapshot_kwargs))
                            else:
                                result = handler(*args, **_snapshot_kwargs)
                            if result is not None:
                                with resource_manager.using(
                                    _current_user, _current_time
                                ):
                                    resource_manager.create(result)
                        except Exception:
                            logger.exception(
                                "Background create action '%s' failed",
                                handler.__name__,
                            )

                    _bg_tasks.add_task(_run_bg)
                    return MsgspecResponse(
                        BackgroundTaskAccepted(message="Task accepted"),
                        status_code=202,
                    )

            # ---- sync mode: call handler + create resource --------------
            elif inspect.iscoroutinefunction(handler):

                async def wrapper(*args, **kwargs):
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    # Convert raw dicts to Struct instances
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))
                    result = await handler(*args, **kwargs)
                    if result is None:
                        return None
                    with resource_manager.using(_current_user, _current_time):
                        info = resource_manager.create(result)
                    return MsgspecResponse(info)

            else:

                def wrapper(*args, **kwargs):
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))
                    result = handler(*args, **kwargs)
                    if result is None:
                        return None
                    with resource_manager.using(_current_user, _current_time):
                        info = resource_manager.create(result)
                    return MsgspecResponse(info)

            wrapper.__name__ = handler.__name__
            wrapper.__qualname__ = handler.__qualname__
            wrapper.__module__ = handler.__module__
            wrapper.__doc__ = handler.__doc__
            wrapper.__signature__ = new_sig
            wrapper.__annotations__ = new_annotations
            return wrapper

        for action in self._pending_create_actions:
            rm = self.resource_managers.get(action.resource_name)
            if rm is None:
                logger.warning(
                    f"create_action '{action.path}' targets resource "
                    f"'{action.resource_name}' which is not registered. Skipping."
                )
                continue

            # Strip leading slash from action.path to avoid double-slash
            action_path_segment = action.path.lstrip("/")
            route_path = f"/{action.resource_name}/{action_path_segment}"

            # --- async_mode='job': build an endpoint that creates a Job ---
            if action.async_mode == "job":
                registry_entry = self._async_job_registry.get(id(action.handler))
                if registry_entry is None:
                    logger.warning(
                        f"async create_action '{action.path}' has no registered "
                        f"Job model. Falling back to sync."
                    )
                    action.async_mode = None
                    # Fall through to sync handler below
                else:
                    (
                        job_resource_name,
                        job_model,
                        target_rm,
                        auto_payload_type,
                        param_conversions,
                    ) = registry_entry
                    job_rm = self.resource_managers[job_resource_name]

                    # Same handler, same FastAPI signature — only the
                    # wrapper behaviour changes (create Job + 202).
                    _wrapper = _build_fastapi_compatible_handler(
                        action.handler,
                        rm,
                        async_job_config=(
                            job_rm,
                            job_resource_name,
                            auto_payload_type,
                            param_conversions,
                        ),
                        deps=deps,
                    )

                    router.post(
                        route_path,
                        response_model=None,
                        status_code=202,
                        summary=f"{action.label} ({action.resource_name})",
                        tags=[f"{action.resource_name}"],
                        openapi_extra={
                            "x-autocrud-create-action": {
                                "resource": action.resource_name,
                                "label": action.label,
                            },
                        },
                    )(_wrapper)
                    continue

            # --- async_mode='background': fire-and-forget via BackgroundTasks ---
            if action.async_mode == "background":
                _wrapper = _build_fastapi_compatible_handler(
                    action.handler, rm, background_mode=True, deps=deps
                )

                router.post(
                    route_path,
                    response_model=None,
                    status_code=202,
                    summary=f"{action.label} ({action.resource_name})",
                    tags=[f"{action.resource_name}"],
                    openapi_extra={
                        "x-autocrud-create-action": {
                            "resource": action.resource_name,
                            "label": action.label,
                        },
                    },
                )(_wrapper)
                continue

            # --- sync (default) handler ---
            _wrapper = _build_fastapi_compatible_handler(action.handler, rm, deps=deps)

            router.post(
                route_path,
                response_model=None,
                responses=struct_to_responses_type(RevisionInfo),
                summary=f"{action.label} ({action.resource_name})",
                tags=[f"{action.resource_name}"],
                openapi_extra={
                    "x-autocrud-create-action": {
                        "resource": action.resource_name,
                        "label": action.label,
                    },
                },
            )(_wrapper)

    def _apply_update_actions(self, router: APIRouter) -> None:
        """Register routes for all pending custom update actions."""
        import msgspec as _msgspec

        from autocrud.crud.route_templates.basic import (
            BaseRouteTemplate,
            DependencyProvider,
            MsgspecResponse,
            jsonschema_to_json_schema_extra,
            struct_to_responses_type,
        )

        if not self._pending_update_actions:
            return

        # Resolve DependencyProvider (same logic as _apply_create_actions)
        deps: DependencyProvider | None = None
        for rt in self.route_templates:
            if isinstance(rt, BaseRouteTemplate) and hasattr(rt, "deps"):
                deps = rt.deps
                break
        if deps is None:
            deps = DependencyProvider()
            if self.default_user is not UNSET:
                deps = deps.with_default_user(self.default_user)

        def _is_msgspec_struct_type(ann: type) -> bool:
            return isinstance(ann, type) and issubclass(ann, _msgspec.Struct)

        def _build_fastapi_compatible_update_handler(
            handler,
            resource_manager,
            *,
            existing_param: str = "existing",
            info_param: str = "info",
            meta_param: str = "meta",
            update_mode: str = "update",
            async_job_config=None,
            background_mode=False,
            deps=None,
        ):
            """Build a FastAPI-compatible endpoint for a custom update action.

            Similar to ``_build_fastapi_compatible_handler`` but:
            - Adds ``resource_id`` as a path parameter.
            - Auto-fetches the existing resource via ``rm.get(resource_id)``
              and injects it into the handler's *existing_param*.
            - Auto-injects ``RevisionInfo`` into *info_param* and
              ``ResourceMeta`` into *meta_param* when declared.
            - Calls ``rm.update()`` or ``rm.modify()`` based on *update_mode*.

            Args:
                handler: The user's update-action endpoint function.
                resource_manager: The target resource's ResourceManager.
                existing_param: Handler param name for existing resource data.
                info_param: Handler param name for RevisionInfo.
                meta_param: Handler param name for ResourceMeta.
                update_mode: ``"update"`` or ``"modify"``.
                async_job_config: When set, a tuple
                    ``(job_rm, job_resource_name, auto_payload_type,
                    param_conversions)`` that switches the wrapper into
                    async-job mode (creates a Job + returns HTTP 202).
                background_mode: When ``True``, schedules the handler via
                    FastAPI ``BackgroundTasks`` and returns HTTP 202.
                deps: A :class:`DependencyProvider` for injecting
                    ``current_user`` and ``current_time``.
            """
            if deps is None:
                deps = DependencyProvider()

            sig = inspect.signature(handler)
            _has_existing_param = existing_param in sig.parameters
            _has_info_param = info_param in sig.parameters
            _has_meta_param = meta_param in sig.parameters
            struct_params: dict[str, type] = {}
            pydantic_params: dict[str, type] = {}
            new_params: list[inspect.Parameter] = []
            new_annotations: dict[str, Any] = {}

            # Add resource_id as first path parameter
            new_params.append(
                inspect.Parameter(
                    "resource_id",
                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
                    annotation=str,
                )
            )
            new_annotations["resource_id"] = str

            for name, param in sig.parameters.items():
                # Skip params that will be injected at runtime
                if name == existing_param:
                    continue
                if name == info_param or name == meta_param:
                    continue
                ann = param.annotation
                if ann is inspect.Parameter.empty:
                    new_params.append(param)
                    continue

                raw_ann, _ = unwrap_annotated(ann)

                if _is_msgspec_struct_type(raw_ann):
                    struct_params[name] = raw_ann
                    new_default = Body(
                        json_schema_extra=jsonschema_to_json_schema_extra(raw_ann),
                    )
                    new_param = param.replace(
                        annotation=inspect.Parameter.empty,
                        default=new_default,
                    )
                    new_params.append(new_param)
                else:
                    new_params.append(param)
                    if ann is not inspect.Parameter.empty:
                        new_annotations[name] = ann

            # Inject current_user and current_time via Depends()
            new_params.append(
                inspect.Parameter(
                    "current_user",
                    inspect.Parameter.KEYWORD_ONLY,
                    default=Depends(deps.get_user),
                    annotation=str,
                )
            )
            new_params.append(
                inspect.Parameter(
                    "current_time",
                    inspect.Parameter.KEYWORD_ONLY,
                    default=Depends(deps.get_now),
                    annotation=dt.datetime,
                )
            )
            new_annotations["current_user"] = str
            new_annotations["current_time"] = dt.datetime

            # Inject BackgroundTasks when background_mode is enabled
            if background_mode:
                from starlette.background import BackgroundTasks

                new_params.append(
                    inspect.Parameter(
                        "background_tasks",
                        inspect.Parameter.KEYWORD_ONLY,
                        annotation=BackgroundTasks,
                    )
                )
                new_annotations["background_tasks"] = BackgroundTasks

            new_sig = sig.replace(
                parameters=new_params, return_annotation=inspect.Parameter.empty
            )

            def _ensure_dict(val: Any) -> Any:
                if isinstance(val, str):
                    import json as _json

                    return _json.loads(val)
                return val

            from autocrud.types import ResourceIDNotFoundError

            # ---- async-job mode: create Job + return 202 ----------------
            if async_job_config is not None:
                from autocrud.types import JobRedirectInfo

                job_rm, job_resource_name, auto_payload_type, _param_convs = (
                    async_job_config
                )
                _param_convs = _param_convs or {}
                payload_param_name = next(iter(struct_params), None)

                async def wrapper(*args, **kwargs):
                    _resource_id = kwargs.pop("resource_id")
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )

                    if auto_payload_type is not None and payload_param_name is not None:
                        # Explicit Struct param — wrap as payload_data
                        inner_data = kwargs.get(payload_param_name)
                        payload_data = auto_payload_type(
                            resource_id=_resource_id, payload_data=inner_data
                        )
                    elif auto_payload_type is not None:
                        # Auto-generated payload: pack individual kwargs
                        payload_data = auto_payload_type(
                            resource_id=_resource_id,
                            **{
                                f: kwargs[f]
                                for f in auto_payload_type.__struct_fields__
                                if f in kwargs and f != "resource_id"
                            },
                        )
                    else:
                        raise HTTPException(
                            status_code=400,
                            detail="Missing payload for async update action.",
                        )

                    job_data = job_rm.resource_type(payload=payload_data)
                    with job_rm.using(_current_user, _current_time):
                        info = job_rm.create(job_data)

                    redirect_url = f"/{job_resource_name}/{info.resource_id}"
                    return MsgspecResponse(
                        JobRedirectInfo(
                            job_resource_name=job_resource_name,
                            job_resource_id=info.resource_id,
                            redirect_url=redirect_url,
                        ),
                        status_code=202,
                    )

            # ---- background mode: schedule via BackgroundTasks + 202 ----
            elif background_mode:
                from autocrud.types import BackgroundTaskAccepted

                _bg_is_async = inspect.iscoroutinefunction(handler)

                async def wrapper(*args, **kwargs):
                    _resource_id = kwargs.pop("resource_id")
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    _bg_tasks = kwargs.pop("background_tasks")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))

                    _snapshot_kwargs = dict(kwargs)

                    def _run_bg() -> None:
                        try:
                            # Lazy-fetch existing resource at BG execution time
                            with resource_manager.using(_current_user, _current_time):
                                existing_resource = resource_manager.get(_resource_id)
                            if _has_existing_param:
                                _snapshot_kwargs[existing_param] = (
                                    existing_resource.data
                                )
                            if _has_info_param:
                                _snapshot_kwargs[info_param] = existing_resource.info
                            if _has_meta_param:
                                with resource_manager.using(
                                    _current_user, _current_time
                                ):
                                    _snapshot_kwargs[meta_param] = (
                                        resource_manager.get_meta(_resource_id)
                                    )

                            if _bg_is_async:
                                result = asyncio.run(handler(*args, **_snapshot_kwargs))
                            else:
                                result = handler(*args, **_snapshot_kwargs)
                            if result is not None:
                                with resource_manager.using(
                                    _current_user, _current_time
                                ):
                                    if update_mode == "modify":
                                        resource_manager.modify(
                                            _resource_id, data=result
                                        )
                                    else:
                                        resource_manager.update(_resource_id, result)
                        except Exception:
                            logger.exception(
                                "Background update action '%s' failed",
                                handler.__name__,
                            )

                    _bg_tasks.add_task(_run_bg)
                    return MsgspecResponse(
                        BackgroundTaskAccepted(message="Task accepted"),
                        status_code=202,
                    )

            # ---- sync mode: call handler + update resource --------------
            elif inspect.iscoroutinefunction(handler):

                async def wrapper(*args, **kwargs):
                    _resource_id = kwargs.pop("resource_id")
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))
                    # Fetch existing resource and inject (only if handler declares it)
                    try:
                        with resource_manager.using(_current_user, _current_time):
                            existing_resource = resource_manager.get(_resource_id)
                    except ResourceIDNotFoundError:
                        raise HTTPException(
                            status_code=404,
                            detail=f"Resource '{_resource_id}' not found.",
                        )
                    if _has_existing_param:
                        kwargs[existing_param] = existing_resource.data
                    if _has_info_param:
                        kwargs[info_param] = existing_resource.info
                    if _has_meta_param:
                        with resource_manager.using(_current_user, _current_time):
                            kwargs[meta_param] = resource_manager.get_meta(_resource_id)
                    result = await handler(**kwargs)
                    if result is None:
                        return None
                    with resource_manager.using(_current_user, _current_time):
                        if update_mode == "modify":
                            info = resource_manager.modify(_resource_id, data=result)
                        else:
                            info = resource_manager.update(_resource_id, result)
                    return MsgspecResponse(info)

            else:

                def wrapper(*args, **kwargs):
                    _resource_id = kwargs.pop("resource_id")
                    _current_user = kwargs.pop("current_user")
                    _current_time = kwargs.pop("current_time")
                    for pname, struct_type in struct_params.items():
                        if pname in kwargs:
                            kwargs[pname] = _msgspec.convert(
                                _ensure_dict(kwargs[pname]), struct_type
                            )
                    for pname, pydantic_type in pydantic_params.items():
                        if pname in kwargs:
                            kwargs[pname] = pydantic_type(**_ensure_dict(kwargs[pname]))
                    # Fetch existing resource and inject (only if handler declares it)
                    try:
                        with resource_manager.using(_current_user, _current_time):
                            existing_resource = resource_manager.get(_resource_id)
                    except ResourceIDNotFoundError:
                        raise HTTPException(
                            status_code=404,
                            detail=f"Resource '{_resource_id}' not found.",
                        )
                    if _has_existing_param:
                        kwargs[existing_param] = existing_resource.data
                    if _has_info_param:
                        kwargs[info_param] = existing_resource.info
                    if _has_meta_param:
                        with resource_manager.using(_current_user, _current_time):
                            kwargs[meta_param] = resource_manager.get_meta(_resource_id)
                    result = handler(**kwargs)
                    if result is None:
                        return None
                    with resource_manager.using(_current_user, _current_time):
                        if update_mode == "modify":
                            info = resource_manager.modify(_resource_id, data=result)
                        else:
                            info = resource_manager.update(_resource_id, result)
                    return MsgspecResponse(info)

            wrapper.__name__ = handler.__name__
            wrapper.__qualname__ = handler.__qualname__
            wrapper.__module__ = handler.__module__
            wrapper.__doc__ = handler.__doc__
            wrapper.__signature__ = new_sig
            wrapper.__annotations__ = new_annotations
            return wrapper

        for action in self._pending_update_actions:
            rm = self.resource_managers.get(action.resource_name)
            if rm is None:
                logger.warning(
                    f"update_action '{action.path}' targets resource "
                    f"'{action.resource_name}' which is not registered. Skipping."
                )
                continue

            action_path_segment = action.path.lstrip("/")
            route_path = (
                f"/{action.resource_name}/{{resource_id}}/{action_path_segment}"
            )

            # --- async_mode='job': build an endpoint that creates a Job ---
            if action.async_mode == "job":
                registry_entry = self._async_update_job_registry.get(id(action.handler))
                if registry_entry is None:
                    logger.warning(
                        f"async update_action '{action.path}' has no registered "
                        f"Job model. Falling back to sync."
                    )
                    action.async_mode = None
                    # Fall through to sync handler below
                else:
                    (
                        job_resource_name,
                        job_model,
                        target_rm,
                        auto_payload_type,
                        param_conversions,
                        _update_mode,
                        _existing_param,
                        _info_param,
                        _meta_param,
                    ) = registry_entry
                    job_rm = self.resource_managers[job_resource_name]

                    _wrapper = _build_fastapi_compatible_update_handler(
                        action.handler,
                        rm,
                        existing_param=action.existing_param,
                        info_param=action.info_param,
                        meta_param=action.meta_param,
                        update_mode=action.mode,
                        async_job_config=(
                            job_rm,
                            job_resource_name,
                            auto_payload_type,
                            param_conversions,
                        ),
                        deps=deps,
                    )

                    router.post(
                        route_path,
                        response_model=None,
                        status_code=202,
                        summary=f"{action.label} ({action.resource_name})",
                        tags=[f"{action.resource_name}"],
                        openapi_extra={
                            "x-autocrud-update-action": {
                                "resource": action.resource_name,
                                "label": action.label,
                                "mode": action.mode,
                            },
                        },
                    )(_wrapper)
                    continue

            # --- async_mode='background': fire-and-forget via BackgroundTasks ---
            if action.async_mode == "background":
                _wrapper = _build_fastapi_compatible_update_handler(
                    action.handler,
                    rm,
                    existing_param=action.existing_param,
                    info_param=action.info_param,
                    meta_param=action.meta_param,
                    update_mode=action.mode,
                    background_mode=True,
                    deps=deps,
                )

                router.post(
                    route_path,
                    response_model=None,
                    status_code=202,
                    summary=f"{action.label} ({action.resource_name})",
                    tags=[f"{action.resource_name}"],
                    openapi_extra={
                        "x-autocrud-update-action": {
                            "resource": action.resource_name,
                            "label": action.label,
                            "mode": action.mode,
                        },
                    },
                )(_wrapper)
                continue

            # --- sync (default) handler ---
            _wrapper = _build_fastapi_compatible_update_handler(
                action.handler,
                rm,
                existing_param=action.existing_param,
                info_param=action.info_param,
                meta_param=action.meta_param,
                update_mode=action.mode,
                deps=deps,
            )

            router.post(
                route_path,
                response_model=None,
                responses=struct_to_responses_type(RevisionInfo),
                summary=f"{action.label} ({action.resource_name})",
                tags=[f"{action.resource_name}"],
                openapi_extra={
                    "x-autocrud-update-action": {
                        "resource": action.resource_name,
                        "label": action.label,
                        "mode": action.mode,
                    },
                },
            )(_wrapper)

    # ------------------------------------------------------------------
    # Ref query routes
    # ------------------------------------------------------------------

    def _apply_ref_routes(self, router: APIRouter) -> None:
        """Generate ref-related API routes on *router*.

        Creates:
        * ``GET /{target}/{resource_id}/referrers`` for each model that is a
          *target* of at least one ``Ref`` annotation.  Returns a list of
          referrer groups with ``source``, ``source_field``, ``ref_type``,
          ``on_delete``, and ``resource_ids``.
        * ``GET /_relationships`` — a global metadata endpoint returning the
          full relationship graph.
        """
        from collections import defaultdict

        # Build target -> list[_RefInfo]
        target_refs: dict[str, list[_RefInfo]] = defaultdict(list)
        for ref_info in self.relationships:
            target_refs[ref_info.target].append(ref_info)

        registered = set(self.resource_managers.keys())

        # Per-target referrers endpoint
        for target_name, refs in target_refs.items():
            if target_name not in registered:
                continue

            # Filter to refs whose source is also registered
            actionable_refs = [r for r in refs if r.source in registered]
            if not actionable_refs:
                continue

            self._add_referrers_route(router, target_name, actionable_refs)

        # Global relationships metadata endpoint
        all_rels = self.relationships

        @router.get(
            "/_relationships",
            summary="List all resource relationships",
            tags=["_meta"],
            description=(
                "Returns the complete relationship graph discovered from "
                "Ref / RefRevision annotations across all registered models."
            ),
        )
        async def _list_relationships() -> list[dict]:
            return [
                {
                    "source": r.source,
                    "source_field": r.source_field,
                    "target": r.target,
                    "ref_type": r.ref_type,
                    "on_delete": r.on_delete.value,
                    "nullable": r.nullable,
                }
                for r in all_rels
            ]

    def _add_referrers_route(
        self,
        router: APIRouter,
        target_name: str,
        refs: list[_RefInfo],
    ) -> None:
        """Register ``GET /{target_name}/{resource_id}/referrers`` on *router*."""
        resource_managers = self.resource_managers

        @router.get(
            f"/{target_name}/{{resource_id}}/referrers",
            summary=f"List referrers of a {target_name} resource",
            tags=[f"{target_name}"],
            description=(
                f"Find all resources that reference a specific `{target_name}` "
                f"resource via Ref-annotated fields.  Results are grouped by "
                f"source model and field."
            ),
        )
        async def _list_referrers(resource_id: str) -> list[dict]:
            # Verify the target resource exists
            target_rm = resource_managers.get(target_name)
            if target_rm is None:
                raise HTTPException(
                    status_code=404, detail=f"Unknown resource type: {target_name}"
                )
            try:
                target_rm.get_meta(resource_id)
            except (ResourceIDNotFoundError, ResourceIsDeletedError):
                raise HTTPException(
                    status_code=404,
                    detail=f"{target_name} '{resource_id}' not found",
                )
            results: list[dict] = []
            for ref_info in refs:
                source_rm = resource_managers.get(ref_info.source)
                if source_rm is None:
                    continue
                # Only resource_id refs are auto-indexed and searchable
                if ref_info.ref_type != "resource_id":
                    continue
                # For list ref fields (e.g. list[Annotated[str, Ref(...)]]),
                # use 'contains' to check if the list includes the target ID.
                # For scalar ref fields, use 'equals' for exact match.
                op = (
                    DataSearchOperator.contains
                    if ref_info.is_list
                    else DataSearchOperator.equals
                )
                metas = source_rm.search_resources(
                    ResourceMetaSearchQuery(
                        is_deleted=False,
                        conditions=[
                            DataSearchCondition(
                                field_path=ref_info.source_field,
                                operator=op,
                                value=resource_id,
                            )
                        ],
                        limit=10_000,
                    )
                )
                if metas:
                    results.append(
                        {
                            "source": ref_info.source,
                            "source_field": ref_info.source_field,
                            "ref_type": ref_info.ref_type,
                            "on_delete": ref_info.on_delete.value,
                            "resource_ids": [m.resource_id for m in metas],
                        }
                    )
            return results

    # ------------------------------------------------------------------
    # Global backup / restore routes
    # ------------------------------------------------------------------

    def _apply_backup_routes(self, router: APIRouter) -> None:
        """Register global ``/_backup/export`` and ``/_backup/import``
        endpoints on *router*.

        * ``GET /_backup/export``  — download a ``.acbak`` archive
          containing **all** registered models.
        * ``POST /_backup/import`` — upload a ``.acbak`` archive and
          load its contents into the matching resource managers.
        """
        import io as _io

        from fastapi import Query as _Query
        from fastapi.responses import StreamingResponse

        autocrud_ref = self  # closure over self

        @router.get(
            "/_backup/export",
            summary="Export all models",
            tags=["_backup"],
            description=(
                "Download a `.acbak` archive containing all registered "
                "models.  Optionally pass `models` query parameter to "
                "restrict which models are exported."
            ),
            response_class=StreamingResponse,
            responses={
                200: {
                    "content": {"application/octet-stream": {}},
                    "description": "Streaming .acbak archive.",
                }
            },
        )
        async def global_export(
            models: list[str] | None = _Query(
                None,
                description=(
                    "Model names to include.  When omitted all registered "
                    "models are exported."
                ),
            ),
        ):
            model_queries: dict[str, Query | ResourceMetaSearchQuery | None] | None = (
                None
            )
            if models:
                unknown = set(models) - set(autocrud_ref.resource_managers)
                if unknown:
                    raise HTTPException(
                        status_code=400,
                        detail=f"Unknown model(s): {', '.join(sorted(unknown))}",
                    )
                model_queries = {m: None for m in models}

            buf = _io.BytesIO()
            autocrud_ref.dump(buf, model_queries=model_queries)
            buf.seek(0)
            return StreamingResponse(
                buf,
                media_type="application/octet-stream",
                headers={
                    "Content-Disposition": 'attachment; filename="backup.acbak"',
                },
            )

        @router.post(
            "/_backup/import",
            summary="Import from archive",
            tags=["_backup"],
            description=(
                "Upload a `.acbak` archive.  All model sections found in "
                "the archive will be loaded into the corresponding resource "
                "managers.  Use `on_duplicate` to control the duplicate "
                "handling strategy."
            ),
        )
        async def global_import(
            file: UploadFile = File(..., description=".acbak archive file"),
            on_duplicate: str = _Query(
                "overwrite",
                description="Strategy: overwrite | skip | raise_error",
            ),
        ) -> dict:
            try:
                strategy = OnDuplicate(on_duplicate)
            except ValueError:
                raise HTTPException(
                    status_code=400,
                    detail=(
                        f"Invalid on_duplicate: {on_duplicate}. "
                        "Must be one of: overwrite, skip, raise_error"
                    ),
                )

            data = await file.read()
            try:
                stats = autocrud_ref.load(_io.BytesIO(data), on_duplicate=strategy)
            except ValueError as e:
                raise HTTPException(status_code=400, detail=str(e))

            return {
                model: {
                    "loaded": s.loaded,
                    "skipped": s.skipped,
                    "total": s.total,
                }
                for model, s in stats.items()
            }

    def dump(
        self,
        bio: IO[bytes],
        model_queries: dict[str, Query | ResourceMetaSearchQuery | None] | None = None,
    ) -> None:
        """Export resources to a streaming msgpack archive.

        Args:
            bio: Binary I/O stream to write to.
            model_queries: Optional ``{model_name: QB_query}`` mapping.
                When *None*, all registered models are exported in full.
                When provided, only the listed models are exported;
                each value is a ``Query`` / ``ResourceMetaSearchQuery``
                (or *None* for "all resources of that model").

        Example::

            # Dump everything
            with open("backup.acbak", "wb") as f:
                autocrud.dump(f)

            # Dump only User resources where name == "Alice"
            from autocrud.query import QB

            with open("backup.acbak", "wb") as f:
                autocrud.dump(f, model_queries={"user": QB.name == "Alice"})
        """
        from autocrud.resource_manager.dump_format import (
            DumpStreamWriter,
            EofRecord,
            HeaderRecord,
            ModelEndRecord,
            ModelStartRecord,
        )

        writer = DumpStreamWriter(bio)
        writer.write(HeaderRecord())

        # Determine which models to dump
        if model_queries is None:
            models_to_dump = {name: None for name in self.resource_managers}
        else:
            models_to_dump = model_queries

        for model_name, query in models_to_dump.items():
            if model_name not in self.resource_managers:
                raise ValueError(
                    f"Model '{model_name}' not found in resource managers."
                )
            mgr = self.resource_managers[model_name]
            writer.write(ModelStartRecord(model_name=model_name))
            for record in mgr.dump(query=query):
                writer.write(record)
            writer.write(ModelEndRecord(model_name=model_name))

        writer.write(EofRecord())

    def load(
        self,
        bio: IO[bytes],
        on_duplicate: "OnDuplicate | None" = None,
    ) -> dict[str, "LoadStats"]:
        """Import resources from a streaming msgpack archive.

        Args:
            bio: Binary I/O stream to read from.
            on_duplicate: Strategy for duplicate resource IDs.
                Defaults to ``OnDuplicate.overwrite``.

        Returns:
            Per-model load statistics: ``{model_name: LoadStats}``.

        Raises:
            ValueError: If the archive format is invalid or contains
                unknown models.
        """
        from autocrud.resource_manager.dump_format import (
            BlobRecord,
            DumpStreamReader,
            EofRecord,
            HeaderRecord,
            MetaRecord,
            ModelEndRecord,
            ModelStartRecord,
            RevisionRecord,
        )
        from autocrud.types import OnDuplicate as _OnDuplicate

        if on_duplicate is None:
            on_duplicate = _OnDuplicate.overwrite

        reader = DumpStreamReader(bio)
        stats: dict[str, LoadStats] = {}

        # Read header
        first = next(reader)
        if not isinstance(first, HeaderRecord):
            raise ValueError(f"Expected HeaderRecord, got {type(first).__name__}.")
        if first.version != 2:
            raise ValueError(f"Unsupported dump format version {first.version}.")

        current_model: str | None = None
        current_mgr = None
        # Per-model record buffers for bulk load
        meta_buf: list[MetaRecord] = []
        rev_buf: list[RevisionRecord] = []
        blob_buf: list[BlobRecord] = []

        for record in reader:
            if isinstance(record, ModelStartRecord):
                current_model = record.model_name
                if current_model not in self.resource_managers:
                    raise ValueError(
                        f"Model '{current_model}' not found in resource managers."
                    )
                current_mgr = self.resource_managers[current_model]
                meta_buf.clear()
                rev_buf.clear()
                blob_buf.clear()
                if current_model not in stats:
                    stats[current_model] = LoadStats()

            elif isinstance(record, ModelEndRecord):
                # Flush buffered records via bulk load
                if current_mgr is not None and current_model is not None:
                    st = current_mgr.load_records_bulk(
                        meta_buf,
                        rev_buf,
                        blob_buf,
                        on_duplicate=on_duplicate,
                    )
                    s = stats[current_model]
                    s.loaded += st.loaded
                    s.skipped += st.skipped
                    s.total += st.total
                current_model = None
                current_mgr = None
                meta_buf.clear()
                rev_buf.clear()
                blob_buf.clear()

            elif isinstance(record, MetaRecord):
                if current_mgr is None:
                    raise ValueError("MetaRecord outside of model section.")
                meta_buf.append(record)

            elif isinstance(record, RevisionRecord):
                if current_mgr is None:
                    raise ValueError("RevisionRecord outside of model section.")
                rev_buf.append(record)

            elif isinstance(record, BlobRecord):
                if current_mgr is None:
                    raise ValueError("BlobRecord outside of model section.")
                blob_buf.append(record)

            elif isinstance(record, EofRecord):
                break

        return stats

Attributes

resource_managers instance-attribute

resource_managers: OrderedDict[str, IResourceManager] = (
    OrderedDict()
)

message_queues instance-attribute

message_queues: OrderedDict[str, IMessageQueue] = (
    OrderedDict()
)

model_names instance-attribute

model_names: dict[type[T], str | None] = {}

relationships instance-attribute

relationships: list[_RefInfo] = []

storage_factory instance-attribute

storage_factory = MemoryStorageFactory()

blob_store instance-attribute

blob_store = MemoryBlobStore()

model_naming instance-attribute

model_naming = 'kebab'

message_queue_factory instance-attribute

message_queue_factory = None

route_templates instance-attribute

route_templates: list[IRouteTemplate] = []

permission_checker instance-attribute

permission_checker = AllowAll()

event_handlers instance-attribute

event_handlers = None

default_encoding instance-attribute

default_encoding = json

default_user instance-attribute

default_user = UNSET

default_now instance-attribute

default_now = UNSET

strict_operation_context instance-attribute

strict_operation_context = False

Functions

configure

configure(
    *,
    model_naming: Literal[
        "same", "pascal", "camel", "snake", "kebab"
    ]
    | Callable[[type], str]
    | UnsetType = UNSET,
    route_templates: list[IRouteTemplate]
    | dict[type, dict[str, Any]]
    | UnsetType = UNSET,
    storage_factory: IStorageFactory | UnsetType = UNSET,
    message_queue_factory: IMessageQueueFactory
    | UnsetType = UNSET,
    admin: str | None | UnsetType = UNSET,
    permission_checker: IPermissionChecker
    | UnsetType = UNSET,
    dependency_provider: DependencyProvider
    | UnsetType = UNSET,
    event_handlers: Sequence[IEventHandler]
    | UnsetType = UNSET,
    encoding: Encoding | UnsetType = UNSET,
    default_user: str
    | Callable[[], str]
    | UnsetType = UNSET,
    default_now: Callable[[], datetime] | UnsetType = UNSET,
    strict_operation_context: bool | UnsetType = UNSET,
) -> None

Configure the AutoCRUD instance dynamically.

This method allows you to reconfigure an existing AutoCRUD instance, useful for the global instance pattern where you import a pre-created instance and configure it later in your application startup.

Warning

This method should only be called during application initialization, before any models are registered or routes are applied. Calling this after models have been registered may lead to inconsistent behavior.

PARAMETER	DESCRIPTION
model_naming	Controls how model names are converted to URL paths. TYPE: Literal['same', 'pascal', 'camel', 'snake', 'kebab'] | Callable[[type], str] | UnsetType DEFAULT: UNSET
route_templates	Custom list of route templates or configuration dict. TYPE: list[IRouteTemplate] | dict[type, dict[str, Any]] | UnsetType DEFAULT: UNSET
storage_factory	Storage backend to use for all models. TYPE: IStorageFactory | UnsetType DEFAULT: UNSET
message_queue_factory	Message queue factory for async job processing. TYPE: IMessageQueueFactory | UnsetType DEFAULT: UNSET
admin	Admin user for RBAC permission system. TYPE: str | None | UnsetType DEFAULT: UNSET
permission_checker	Custom permission checker implementation. TYPE: IPermissionChecker | UnsetType DEFAULT: UNSET
dependency_provider	Dependency injection provider for routes. TYPE: DependencyProvider | UnsetType DEFAULT: UNSET
event_handlers	List of event handlers for lifecycle hooks. TYPE: Sequence[IEventHandler] | UnsetType DEFAULT: UNSET
encoding	Default encoding format (json/msgpack). TYPE: Encoding | UnsetType DEFAULT: UNSET
default_user	Default user for operations when not specified. When set, the DependencyProvider's default get_user will return this value instead of "anonymous". A custom get_user on the provider always takes priority. TYPE: str | Callable[[], str] | UnsetType DEFAULT: UNSET
default_now	Default timestamp function for operations. TYPE: Callable[[], datetime] | UnsetType DEFAULT: UNSET
strict_operation_context	When True, write operations on all registered models will raise :class:MissingOperationContextError if user and now are not resolved from any source (explicit kwargs, using() scope, or manager defaults). TYPE: bool | UnsetType DEFAULT: UNSET

Example

from autocrud import crud
from autocrud.resource_manager.storage_factory import DiskStorageFactory

# Configure the global instance
crud.configure(
    storage_factory=DiskStorageFactory("./data"),
    model_naming="snake",
    admin="root@example.com",
)

# Now register models
crud.add_model(User)

Source code in autocrud/crud/core.py

def configure(
    self,
    *,
    model_naming: Literal["same", "pascal", "camel", "snake", "kebab"]
    | Callable[[type], str]
    | UnsetType = UNSET,
    route_templates: list[IRouteTemplate]
    | dict[type, dict[str, Any]]
    | UnsetType = UNSET,
    storage_factory: IStorageFactory | UnsetType = UNSET,
    message_queue_factory: IMessageQueueFactory | UnsetType = UNSET,
    admin: str | None | UnsetType = UNSET,
    permission_checker: IPermissionChecker | UnsetType = UNSET,
    dependency_provider: DependencyProvider | UnsetType = UNSET,
    event_handlers: Sequence[IEventHandler] | UnsetType = UNSET,
    encoding: Encoding | UnsetType = UNSET,
    default_user: str | Callable[[], str] | UnsetType = UNSET,
    default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
    strict_operation_context: bool | UnsetType = UNSET,
) -> None:
    """Configure the AutoCRUD instance dynamically.

    This method allows you to reconfigure an existing AutoCRUD instance,
    useful for the global instance pattern where you import a pre-created
    instance and configure it later in your application startup.

    Warning:
        This method should only be called during application initialization,
        before any models are registered or routes are applied. Calling this
        after models have been registered may lead to inconsistent behavior.

    Args:
        model_naming: Controls how model names are converted to URL paths.
        route_templates: Custom list of route templates or configuration dict.
        storage_factory: Storage backend to use for all models.
        message_queue_factory: Message queue factory for async job processing.
        admin: Admin user for RBAC permission system.
        permission_checker: Custom permission checker implementation.
        dependency_provider: Dependency injection provider for routes.
        event_handlers: List of event handlers for lifecycle hooks.
        encoding: Default encoding format (json/msgpack).
        default_user: Default user for operations when not specified.  When set,
            the ``DependencyProvider``'s default ``get_user`` will return this
            value instead of ``"anonymous"``.  A custom ``get_user`` on the
            provider always takes priority.
        default_now: Default timestamp function for operations.
        strict_operation_context: When ``True``, write operations on all
            registered models will raise
            :class:`MissingOperationContextError` if ``user`` and ``now``
            are not resolved from any source (explicit kwargs,
            ``using()`` scope, or manager defaults).

    Example:
        ```python
        from autocrud import crud
        from autocrud.resource_manager.storage_factory import DiskStorageFactory

        # Configure the global instance
        crud.configure(
            storage_factory=DiskStorageFactory("./data"),
            model_naming="snake",
            admin="root@example.com",
        )

        # Now register models
        crud.add_model(User)
        ```
    """
    if self.resource_managers:
        logger.warning(
            "configure() called after models have been registered. "
            "This may lead to inconsistent behavior."
        )

    # Apply configuration using shared logic
    self._apply_configuration(
        model_naming=model_naming,
        route_templates=route_templates,
        storage_factory=storage_factory,
        message_queue_factory=message_queue_factory,
        admin=admin,
        permission_checker=permission_checker,
        dependency_provider=dependency_provider,
        event_handlers=event_handlers,
        encoding=encoding,
        default_user=default_user,
        default_now=default_now,
        strict_operation_context=strict_operation_context,
    )

get_resource_manager

get_resource_manager(
    model: type[T] | str,
) -> IResourceManager[T]

Get the resource manager for a registered model.

This method allows you to access the underlying ResourceManager for a specific model. The ResourceManager provides low-level access to storage, events, and other internal components for that model.

PARAMETER	DESCRIPTION
model	The model class or its registered resource name. TYPE: type[T] | str

RETURNS	DESCRIPTION
IResourceManager[T]	The IResourceManager instance associated with the model.

RAISES	DESCRIPTION
KeyError	If the model is not registered.
ValueError	If the model class is registered with multiple names (ambiguous).

Example

# Get by model class
manager = autocrud.get_resource_manager(User)

# Get by resource name
manager = autocrud.get_resource_manager("users")

# Access underlying storage
storage = manager.storage

Source code in autocrud/crud/core.py

def get_resource_manager(self, model: type[T] | str) -> IResourceManager[T]:
    """Get the resource manager for a registered model.

    This method allows you to access the underlying ResourceManager for a specific model.
    The ResourceManager provides low-level access to storage, events, and other
    internal components for that model.

    Args:
        model: The model class or its registered resource name.

    Returns:
        The IResourceManager instance associated with the model.

    Raises:
        KeyError: If the model is not registered.
        ValueError: If the model class is registered with multiple names (ambiguous).

    Example:
        ```python
        # Get by model class
        manager = autocrud.get_resource_manager(User)

        # Get by resource name
        manager = autocrud.get_resource_manager("users")

        # Access underlying storage
        storage = manager.storage
        ```
    """
    if isinstance(model, str):
        return self.resource_managers[model]
    model_name = self.model_names[model]
    if model_name is None:
        raise ValueError(
            f"Model {get_type_name(model) or repr(model)} is registered with multiple names."
        )
    return self.resource_managers[model_name]

add_route_template

add_route_template(template: IRouteTemplate) -> None

Add a custom route template to extend the API with additional endpoints.

Route templates define how to generate specific API endpoints for models. By adding custom templates, you can extend the default CRUD functionality with specialized endpoints for your use cases.

If a template of the same type already exists (e.g. added by the default configure()), it is replaced rather than duplicated. This prevents Duplicate Operation ID warnings for templates that mount global routes such as BlobRouteTemplate and GraphQLRouteTemplate.

PARAMETER	DESCRIPTION
template	A custom route template implementing IRouteTemplate interface. TYPE: IRouteTemplate

Example

class CustomSearchTemplate(BaseRouteTemplate):
    def apply(self, model_name, resource_manager, router):
        @router.get(f"/{model_name}/search")
        async def search_resources(query: str):
            # Custom search logic
            pass


autocrud = AutoCRUD()
autocrud.add_route_template(CustomSearchTemplate())
autocrud.add_model(User)

Note

Templates are sorted by their order property before being applied. Add templates before calling add_model() or apply() for best results.

Source code in autocrud/crud/core.py

def add_route_template(self, template: IRouteTemplate) -> None:
    """Add a custom route template to extend the API with additional endpoints.

    Route templates define how to generate specific API endpoints for models.
    By adding custom templates, you can extend the default CRUD functionality
    with specialized endpoints for your use cases.

    If a template of the **same type** already exists (e.g. added by the
    default ``configure()``), it is **replaced** rather than duplicated.
    This prevents ``Duplicate Operation ID`` warnings for templates that
    mount global routes such as ``BlobRouteTemplate`` and
    ``GraphQLRouteTemplate``.

    Args:
        template: A custom route template implementing IRouteTemplate interface.

    Example:
        ```python
        class CustomSearchTemplate(BaseRouteTemplate):
            def apply(self, model_name, resource_manager, router):
                @router.get(f"/{model_name}/search")
                async def search_resources(query: str):
                    # Custom search logic
                    pass


        autocrud = AutoCRUD()
        autocrud.add_route_template(CustomSearchTemplate())
        autocrud.add_model(User)
        ```

    Note:
        Templates are sorted by their order property before being applied.
        Add templates before calling add_model() or apply() for best results.
    """
    # Replace any existing template of the same type to avoid duplicates.
    # This is important for templates that mount global routes (e.g.
    # BlobRouteTemplate, GraphQLRouteTemplate) — having two instances
    # would register the same path twice, producing a FastAPI
    # "Duplicate Operation ID" warning.
    template_type = type(template)
    self.route_templates = [
        t for t in self.route_templates if type(t) is not template_type
    ]
    self.route_templates.append(template)

create_action

create_action(
    resource_name: str,
    *,
    path: str | None = None,
    label: str | None = None,
    async_mode: Literal["job", "background"] | None = None,
    job_name: str | None = None,
) -> Callable

Decorator to register a custom create action for a resource.

The decorated function is a standard FastAPI endpoint handler — all input parsing (Body, Query, Path, Depends, etc.) is handled by FastAPI. If the handler returns a resource-type object, AutoCRUD will automatically call resource_manager.create() and respond with RevisionInfo. If it returns None, no automatic creation occurs.

When async_mode='job' is set, the framework automatically:

1. Generates a Job model with the handler's body type as payload.
2. Registers the Job model with a message queue.
3. On POST, creates a Job instance (PENDING) and enqueues it.
4. Returns HTTP 202 with :class:~autocrud.types.JobRedirectInfo.
5. In the background, executes the handler with the payload.
6. If the handler returns a resource object, auto-creates it and stores the RevisionInfo as the Job's artifact.

When async_mode='background' is set, the framework:

1. On POST, schedules the handler via FastAPI BackgroundTasks.
2. Returns HTTP 202 with :class:~autocrud.types.BackgroundTaskAccepted immediately.
3. The handler runs in the background; if it returns a resource object, resource_manager.create() is called automatically.
4. No Job model is created — the task is fire-and-forget.
5. Errors are logged but not surfaced to the client.

This mode is suitable for tasks that take a few seconds to complete and do not require progress tracking.

PARAMETER	DESCRIPTION
resource_name	The name of the resource this action belongs to. TYPE: str
path	URL path suffix (e.g. "import-from-url"). If None, inferred from the function name (underscores → hyphens). TYPE: str | None DEFAULT: None
label	Human-friendly label shown in the UI. If None, inferred from path (hyphens → spaces, title-cased). TYPE: str | None DEFAULT: None
async_mode	Execution mode for the action. None (default) executes synchronously. 'job' executes asynchronously via the message queue system. 'background' executes asynchronously via FastAPI BackgroundTasks (fire-and-forget, no Job tracking). TYPE: Literal['job', 'background'] | None DEFAULT: None
job_name	Custom resource name for the auto-generated Job model (e.g. "my-custom-job"). If None, derived automatically from path and resource_name. Only meaningful when async_mode='job'. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
Callable	A decorator that registers the handler and returns it unchanged.

Example

class ImportFromUrl(Struct):
    url: str


@crud.create_action("article", label="Import from URL")
async def import_from_url(body: ImportFromUrl = Body(...)):
    content = await fetch_and_parse(body.url)
    return Article(content=content)  # auto-created


class GenerateRequest(Struct):
    prompt: str


@crud.create_action("article", async_mode="job", label="Generate")
def generate_article(payload: GenerateRequest = Body(...)) -> Article:
    content = call_llm(payload.prompt)  # long-running
    return Article(content=content)  # auto-created in background

Note

This decorator is lazy — it stores metadata without registering any route. Routes are created when apply() is called, so the decorator can be used before or after add_model().

Source code in autocrud/crud/core.py

def create_action(
    self,
    resource_name: str,
    *,
    path: str | None = None,
    label: str | None = None,
    async_mode: Literal["job", "background"] | None = None,
    job_name: str | None = None,
) -> Callable:
    """Decorator to register a custom create action for a resource.

    The decorated function is a standard FastAPI endpoint handler — all input
    parsing (``Body``, ``Query``, ``Path``, ``Depends``, etc.) is handled by
    FastAPI.  If the handler returns a resource-type object, AutoCRUD will
    automatically call ``resource_manager.create()`` and respond with
    ``RevisionInfo``.  If it returns ``None``, no automatic creation occurs.

    When ``async_mode='job'`` is set, the framework automatically:

    1. Generates a ``Job`` model with the handler's body type as payload.
    2. Registers the Job model with a message queue.
    3. On POST, creates a Job instance (PENDING) and enqueues it.
    4. Returns HTTP 202 with :class:`~autocrud.types.JobRedirectInfo`.
    5. In the background, executes the handler with the payload.
    6. If the handler returns a resource object, auto-creates it and
       stores the ``RevisionInfo`` as the Job's artifact.

    When ``async_mode='background'`` is set, the framework:

    1. On POST, schedules the handler via FastAPI ``BackgroundTasks``.
    2. Returns HTTP 202 with :class:`~autocrud.types.BackgroundTaskAccepted`
       immediately.
    3. The handler runs in the background; if it returns a resource object,
       ``resource_manager.create()`` is called automatically.
    4. No Job model is created — the task is fire-and-forget.
    5. Errors are logged but not surfaced to the client.

    This mode is suitable for tasks that take a few seconds to complete
    and do not require progress tracking.

    Args:
        resource_name: The name of the resource this action belongs to.
        path: URL path suffix (e.g. ``"import-from-url"``).  If ``None``,
            inferred from the function name (underscores → hyphens).
        label: Human-friendly label shown in the UI.  If ``None``,
            inferred from *path* (hyphens → spaces, title-cased).
        async_mode: Execution mode for the action.  ``None`` (default)
            executes synchronously.  ``'job'`` executes asynchronously
            via the message queue system.  ``'background'`` executes
            asynchronously via FastAPI ``BackgroundTasks``
            (fire-and-forget, no Job tracking).
        job_name: Custom resource name for the auto-generated Job model
            (e.g. ``"my-custom-job"``).  If ``None``, derived automatically
            from *path* and *resource_name*.  Only meaningful when
            ``async_mode='job'``.

    Returns:
        A decorator that registers the handler and returns it unchanged.

    Example:
        ```python
        class ImportFromUrl(Struct):
            url: str


        @crud.create_action("article", label="Import from URL")
        async def import_from_url(body: ImportFromUrl = Body(...)):
            content = await fetch_and_parse(body.url)
            return Article(content=content)  # auto-created


        class GenerateRequest(Struct):
            prompt: str


        @crud.create_action("article", async_mode="job", label="Generate")
        def generate_article(payload: GenerateRequest = Body(...)) -> Article:
            content = call_llm(payload.prompt)  # long-running
            return Article(content=content)  # auto-created in background
        ```

    Note:
        This decorator is lazy — it stores metadata without registering any
        route.  Routes are created when ``apply()`` is called, so the
        decorator can be used before or after ``add_model()``.
    """

    def decorator(func: Callable) -> Callable:
        action_path = path or func.__name__.replace("_", "-")
        action_label = label or action_path.replace("-", " ").title()
        self._pending_create_actions.append(
            _PendingCreateAction(
                resource_name=resource_name,
                path=action_path,
                label=action_label,
                handler=func,
                async_mode=async_mode,
                job_name=job_name,
            )
        )
        return func

    return decorator

update_action

update_action(
    resource_name: str,
    *,
    path: str | None = None,
    label: str | None = None,
    mode: Literal["update", "modify"] = "update",
    existing_param: str = "existing",
    info_param: str = "info",
    meta_param: str = "meta",
    async_mode: Literal["job", "background"] | None = None,
    job_name: str | None = None,
) -> Callable

Decorator to register a custom update action for a resource.

The decorated function receives the existing resource data (auto-injected) and any custom input parameters. If the handler returns a resource-type object, AutoCRUD will automatically call resource_manager.update() (or resource_manager.modify() when mode='modify') and respond with RevisionInfo. If it returns None, no update occurs.

The existing resource data is automatically fetched via resource_manager.get(resource_id) and injected into the handler parameter named by existing_param (default "existing").

Similarly, the handler may declare parameters named info_param (default "info") and meta_param (default "meta") to receive the existing resource's RevisionInfo and ResourceMeta respectively. Like existing_param, these are detected by parameter name and only injected when the handler declares them.

When async_mode='job' is set, the framework automatically:

1. Generates a Job model with the handler's body type as payload (plus an auto-injected resource_id field).
2. Registers the Job model with a message queue.
3. On POST, creates a Job instance (PENDING) and enqueues it.
4. Returns HTTP 202 with :class:~autocrud.types.JobRedirectInfo.
5. In the background, fetches existing resource (lazy), executes the handler with the payload and existing data.
6. If the handler returns a resource object, auto-updates it and stores the RevisionInfo as the Job's artifact.

When async_mode='background' is set, the framework:

1. On POST, schedules the handler via FastAPI BackgroundTasks.
2. Returns HTTP 202 with :class:~autocrud.types.BackgroundTaskAccepted immediately.
3. The handler runs in the background; if it returns a resource object, resource_manager.update() (or modify()) is called automatically.
4. No Job model is created — the task is fire-and-forget.
5. Errors are logged but not surfaced to the client.

PARAMETER	DESCRIPTION
resource_name	The name of the resource this action belongs to. TYPE: str
path	URL path suffix (e.g. "level-up"). If None, inferred from the function name (underscores → hyphens). TYPE: str | None DEFAULT: None
label	Human-friendly label shown in the UI. If None, inferred from path (hyphens → spaces, title-cased). TYPE: str | None DEFAULT: None
mode	Update mode. "update" (default) creates a new revision. "modify" performs an in-place edit (only valid for draft-status resources). TYPE: Literal['update', 'modify'] DEFAULT: 'update'
existing_param	The handler parameter name into which the existing resource data will be injected. Defaults to "existing". TYPE: str DEFAULT: 'existing'
info_param	The handler parameter name into which the existing resource's RevisionInfo will be injected. Defaults to "info". TYPE: str DEFAULT: 'info'
meta_param	The handler parameter name into which the existing resource's ResourceMeta will be injected. Defaults to "meta". TYPE: str DEFAULT: 'meta'
async_mode	Execution mode for the action. None (default) executes synchronously. 'job' executes asynchronously via the message queue system. 'background' executes asynchronously via FastAPI BackgroundTasks (fire-and-forget, no Job tracking). TYPE: Literal['job', 'background'] | None DEFAULT: None
job_name	Custom resource name for the auto-generated Job model (e.g. "my-custom-job"). If None, derived automatically from path and resource_name. Only meaningful when async_mode='job'. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
Callable	A decorator that registers the handler and returns it unchanged.

Example

class LevelUpInput(Struct):
    levels: int = 1


@crud.update_action("character", label="Level Up")
def level_up(
    existing: Character,
    body: LevelUpInput = Body(...),
) -> Character:
    return Character(
        name=existing.name,
        level=existing.level + body.levels,
    )


@crud.update_action(
    "character",
    label="Train",
    async_mode="job",
)
def train(
    existing: Character,
    body: LevelUpInput = Body(...),
) -> Character:
    import time

    time.sleep(10)  # long-running training
    return Character(
        name=existing.name,
        level=existing.level + body.levels,
    )


@crud.update_action(
    "character",
    label="Background Heal",
    async_mode="background",
)
def bg_heal(existing: Character) -> Character:
    import time

    time.sleep(5)
    return Character(name=existing.name, level=existing.level + 1)

Note

This decorator is lazy — it stores metadata without registering any route. Routes are created when apply() is called. The route is POST /{resource_name}/{resource_id}/{action_path}.

Source code in autocrud/crud/core.py

def update_action(
    self,
    resource_name: str,
    *,
    path: str | None = None,
    label: str | None = None,
    mode: Literal["update", "modify"] = "update",
    existing_param: str = "existing",
    info_param: str = "info",
    meta_param: str = "meta",
    async_mode: Literal["job", "background"] | None = None,
    job_name: str | None = None,
) -> Callable:
    """Decorator to register a custom update action for a resource.

    The decorated function receives the existing resource data (auto-injected)
    and any custom input parameters.  If the handler returns a resource-type
    object, AutoCRUD will automatically call ``resource_manager.update()`` (or
    ``resource_manager.modify()`` when ``mode='modify'``) and respond with
    ``RevisionInfo``.  If it returns ``None``, no update occurs.

    The existing resource data is automatically fetched via
    ``resource_manager.get(resource_id)`` and injected into the handler
    parameter named by *existing_param* (default ``"existing"``).

    Similarly, the handler may declare parameters named *info_param*
    (default ``"info"``) and *meta_param* (default ``"meta"``) to
    receive the existing resource's ``RevisionInfo`` and ``ResourceMeta``
    respectively.  Like *existing_param*, these are detected by
    **parameter name** and only injected when the handler declares them.

    When ``async_mode='job'`` is set, the framework automatically:

    1. Generates a ``Job`` model with the handler's body type as payload
       (plus an auto-injected ``resource_id`` field).
    2. Registers the Job model with a message queue.
    3. On POST, creates a Job instance (PENDING) and enqueues it.
    4. Returns HTTP 202 with :class:`~autocrud.types.JobRedirectInfo`.
    5. In the background, fetches existing resource (lazy), executes
       the handler with the payload and existing data.
    6. If the handler returns a resource object, auto-updates it and
       stores the ``RevisionInfo`` as the Job's artifact.

    When ``async_mode='background'`` is set, the framework:

    1. On POST, schedules the handler via FastAPI ``BackgroundTasks``.
    2. Returns HTTP 202 with :class:`~autocrud.types.BackgroundTaskAccepted`
       immediately.
    3. The handler runs in the background; if it returns a resource object,
       ``resource_manager.update()`` (or ``modify()``) is called
       automatically.
    4. No Job model is created — the task is fire-and-forget.
    5. Errors are logged but not surfaced to the client.

    Args:
        resource_name: The name of the resource this action belongs to.
        path: URL path suffix (e.g. ``"level-up"``).  If ``None``,
            inferred from the function name (underscores → hyphens).
        label: Human-friendly label shown in the UI.  If ``None``,
            inferred from *path* (hyphens → spaces, title-cased).
        mode: Update mode.  ``"update"`` (default) creates a new
            revision.  ``"modify"`` performs an in-place edit (only
            valid for draft-status resources).
        existing_param: The handler parameter name into which the
            existing resource data will be injected.  Defaults to
            ``"existing"``.
        info_param: The handler parameter name into which the
            existing resource's ``RevisionInfo`` will be injected.
            Defaults to ``"info"``.
        meta_param: The handler parameter name into which the
            existing resource's ``ResourceMeta`` will be injected.
            Defaults to ``"meta"``.
        async_mode: Execution mode for the action.  ``None`` (default)
            executes synchronously.  ``'job'`` executes asynchronously
            via the message queue system.  ``'background'`` executes
            asynchronously via FastAPI ``BackgroundTasks``
            (fire-and-forget, no Job tracking).
        job_name: Custom resource name for the auto-generated Job model
            (e.g. ``"my-custom-job"``).  If ``None``, derived automatically
            from *path* and *resource_name*.  Only meaningful when
            ``async_mode='job'``.

    Returns:
        A decorator that registers the handler and returns it unchanged.

    Example:
        ```python
        class LevelUpInput(Struct):
            levels: int = 1


        @crud.update_action("character", label="Level Up")
        def level_up(
            existing: Character,
            body: LevelUpInput = Body(...),
        ) -> Character:
            return Character(
                name=existing.name,
                level=existing.level + body.levels,
            )


        @crud.update_action(
            "character",
            label="Train",
            async_mode="job",
        )
        def train(
            existing: Character,
            body: LevelUpInput = Body(...),
        ) -> Character:
            import time

            time.sleep(10)  # long-running training
            return Character(
                name=existing.name,
                level=existing.level + body.levels,
            )


        @crud.update_action(
            "character",
            label="Background Heal",
            async_mode="background",
        )
        def bg_heal(existing: Character) -> Character:
            import time

            time.sleep(5)
            return Character(name=existing.name, level=existing.level + 1)
        ```

    Note:
        This decorator is lazy — it stores metadata without registering any
        route.  Routes are created when ``apply()`` is called.
        The route is ``POST /{resource_name}/{resource_id}/{action_path}``.
    """

    def decorator(func: Callable) -> Callable:
        action_path = path or func.__name__.replace("_", "-")
        action_label = label or action_path.replace("-", " ").title()
        self._pending_update_actions.append(
            _PendingUpdateAction(
                resource_name=resource_name,
                path=action_path,
                label=action_label,
                handler=func,
                mode=mode,
                existing_param=existing_param,
                info_param=info_param,
                meta_param=meta_param,
                async_mode=async_mode,
                job_name=job_name,
            )
        )
        return func

    return decorator

add_model

add_model(
    model: "type[T] | Schema[T]",
    *,
    name: str | None = None,
    id_generator: Callable[[], str] | None = None,
    storage: IStorage | None = None,
    migration: "IMigration | Schema | None" = None,
    indexed_fields: list[
        str | tuple[str, type] | IndexableField
    ]
    | None = None,
    event_handlers: Sequence[IEventHandler] | None = None,
    permission_checker: IPermissionChecker | None = None,
    encoding: Encoding | None = None,
    default_status: RevisionStatus | None = None,
    default_user: str
    | Callable[[], str]
    | UnsetType = UNSET,
    default_now: Callable[[], datetime] | UnsetType = UNSET,
    message_queue_factory: IMessageQueueFactory
    | None
    | UnsetType = UNSET,
    job_handler: Callable[[Resource[Job[T]]], None]
    | None = None,
    job_handler_factory: Callable[
        [], Callable[[Resource[Job[T]]], None]
    ]
    | None = None,
    validator: "Callable[[T], None] | IValidator | type | None" = None,
    constraint_checkers: "Sequence[IConstraintChecker | Callable[[ResourceManager], IConstraintChecker]] | None" = None,
) -> None

Register a resource model (or Schema) and create its ResourceManager.

After a model is registered, calling apply(router) will generate FastAPI routes for it using the configured route templates.

You can register either: - a plain model type: add_model(User) - a Schema: add_model(Schema(User, version=...))

PARAMETER	DESCRIPTION
model	Resource type or Schema. Supported types depend on your project setup, commonly msgspec Struct. Pydantic BaseModel is supported and will be converted to a struct. TYPE: 'type[T] | Schema[T]'
name	Resource name (used as route base path). If None, derived from the model type and model_naming. TYPE: str | None DEFAULT: None
id_generator	Custom ID generator for created resources. If None, the default generator is used by ResourceManager. TYPE: Callable[[], str] | None DEFAULT: None
storage	Storage instance for this resource. If None, a storage is created via self.storage_factory.build(model_name). TYPE: IStorage | None DEFAULT: None
migration	Schema/migration configuration. - If model is a Schema, migration must be None. - If migration is a Schema, it is used as the resolved schema for this model. - Passing IMigration is supported but deprecated (converted via Schema.from_legacy). TYPE: 'IMigration | Schema | None' DEFAULT: None
indexed_fields	Fields to index for search/query. Each element can be: - IndexableField - str (field path) - (field_path: str, field_type: type) tuple TYPE: list[str | tuple[str, type] | IndexableField] | None DEFAULT: None
event_handlers	Per-model event handlers. If self.event_handlers is configured globally, it takes precedence; otherwise these handlers are used. TYPE: Sequence[IEventHandler] | None DEFAULT: None
permission_checker	Per-model permission checker. If self.permission_checker is configured globally, it takes precedence; otherwise this checker is used. TYPE: IPermissionChecker | None DEFAULT: None
encoding	Encoding for stored payloads. If None, uses self.default_encoding. TYPE: Encoding | None DEFAULT: None
default_status	Default revision status for this resource (if supported by the revision model). TYPE: RevisionStatus | None DEFAULT: None
default_user	Per-model default user (or factory). If UNSET, falls back to self.default_user when configured. TYPE: str | Callable[[], str] | UnsetType DEFAULT: UNSET
default_now	Per-model default timestamp function. If UNSET, falls back to self.default_now when configured. TYPE: Callable[[], datetime] | UnsetType DEFAULT: UNSET
message_queue_factory	Overrides message queue behavior for Job models: - UNSET: use self.message_queue_factory - None: explicitly disable queue - factory instance: use the provided factory TYPE: IMessageQueueFactory | None | UnsetType DEFAULT: UNSET
job_handler	Handler for Job resources (when the model is detected as a Job subclass). TYPE: Callable[[Resource[Job[T]]], None] | None DEFAULT: None
job_handler_factory	Lazy factory producing a job handler. If provided, it is wrapped as a lazy handler. TYPE: Callable[[], Callable[[Resource[Job[T]]], None]] | None DEFAULT: None
validator	Validation hook(s). When the model is a Pydantic BaseModel and no validator is set on the resolved schema, the Pydantic model is used as validator by default. TYPE: 'Callable[[T], None] | IValidator | type | None' DEFAULT: None
constraint_checkers	Extra constraint checkers for this resource. Each element can be an instance or a factory callable that receives the ResourceManager and returns a checker. TYPE: 'Sequence[IConstraintChecker | Callable[[ResourceManager], IConstraintChecker]] | None' DEFAULT: None

Behavior

• If model is a Schema, it must declare resource_type; schema-level migration/validator should be provided on the Schema itself.
• If the model is a Pydantic type, it is converted to a struct for storage and the Pydantic model can be used for validation.
• Ref relationships are collected from Ref / RefRevision annotations for later route and referential integrity setup.
• Ref fields (resource_id refs only) are auto-indexed for searchability.
• For Job models with a message queue enabled, status and retries are auto-indexed (if not already present in indexed_fields).

RAISES	DESCRIPTION
ValueError	if the resource name already exists if Schema is passed as first argument but migration/validator is also provided if Ref(..., on_delete=set_null) is used on a non-optional field
TypeError	if indexed_fields contains an invalid item

Examples:

Basic registration:

from autocrud import AutoCRUD

autocrud = AutoCRUD()
autocrud.add_model(User)

Custom resource name:

autocrud.add_model(User, name="people")

Provide explicit storage:

# storage is per-model; if you want a default for all models, pass `storage_factory=...`
# when constructing AutoCRUD / calling configure().
model_name = "people"
st = autocrud.storage_factory.build(model_name)
autocrud.add_model(User, name=model_name, storage=st)

Using Schema as the first argument:

schema = Schema(User, version="v1")
autocrud.add_model(schema)

Source code in autocrud/crud/core.py

def add_model(
    self,
    model: "type[T] | Schema[T]",
    *,
    name: str | None = None,
    id_generator: Callable[[], str] | None = None,
    storage: IStorage | None = None,
    migration: "IMigration | Schema | None" = None,
    indexed_fields: list[str | tuple[str, type] | IndexableField] | None = None,
    event_handlers: Sequence[IEventHandler] | None = None,
    permission_checker: IPermissionChecker | None = None,
    encoding: Encoding | None = None,
    default_status: RevisionStatus | None = None,
    default_user: str | Callable[[], str] | UnsetType = UNSET,
    default_now: Callable[[], dt.datetime] | UnsetType = UNSET,
    message_queue_factory: IMessageQueueFactory | None | UnsetType = UNSET,
    job_handler: Callable[[Resource[Job[T]]], None] | None = None,
    job_handler_factory: Callable[[], Callable[[Resource[Job[T]]], None]]
    | None = None,
    validator: "Callable[[T], None] | IValidator | type | None" = None,
    constraint_checkers: "Sequence[IConstraintChecker | Callable[[ResourceManager], IConstraintChecker]] | None" = None,
) -> None:
    """Register a resource model (or `Schema`) and create its `ResourceManager`.

    After a model is registered, calling `apply(router)` will generate FastAPI routes for it
    using the configured route templates.

    You can register either:
    - a plain model type: `add_model(User)`
    - a `Schema`: `add_model(Schema(User, version=...))`

    Args:
        model:
            Resource type or `Schema`. Supported types depend on your project setup, commonly
            msgspec `Struct`. Pydantic `BaseModel` is supported and will be converted to a struct.
        name:
            Resource name (used as route base path). If `None`, derived from the model type and
            `model_naming`.
        id_generator:
            Custom ID generator for created resources. If `None`, the default generator is used
            by `ResourceManager`.
        storage:
            Storage instance for this resource. If `None`, a storage is created via
            `self.storage_factory.build(model_name)`.
        migration:
            Schema/migration configuration.
            - If `model` is a `Schema`, `migration` must be `None`.
            - If `migration` is a `Schema`, it is used as the resolved schema for this model.
            - Passing `IMigration` is supported but **deprecated** (converted via `Schema.from_legacy`).
        indexed_fields:
            Fields to index for search/query. Each element can be:
            - `IndexableField`
            - `str` (field path)
            - `(field_path: str, field_type: type)` tuple
        event_handlers:
            Per-model event handlers. If `self.event_handlers` is configured globally, it takes
            precedence; otherwise these handlers are used.
        permission_checker:
            Per-model permission checker. If `self.permission_checker` is configured globally, it
            takes precedence; otherwise this checker is used.
        encoding:
            Encoding for stored payloads. If `None`, uses `self.default_encoding`.
        default_status:
            Default revision status for this resource (if supported by the revision model).
        default_user:
            Per-model default user (or factory). If `UNSET`, falls back to `self.default_user`
            when configured.
        default_now:
            Per-model default timestamp function. If `UNSET`, falls back to `self.default_now`
            when configured.
        message_queue_factory:
            Overrides message queue behavior for Job models:
            - `UNSET`: use `self.message_queue_factory`
            - `None`: explicitly disable queue
            - factory instance: use the provided factory
        job_handler:
            Handler for Job resources (when the model is detected as a Job subclass).
        job_handler_factory:
            Lazy factory producing a job handler. If provided, it is wrapped as a lazy handler.
        validator:
            Validation hook(s). When the model is a Pydantic `BaseModel` and no validator is set
            on the resolved schema, the Pydantic model is used as validator by default.
        constraint_checkers:
            Extra constraint checkers for this resource. Each element can be an instance or a
            factory callable that receives the `ResourceManager` and returns a checker.

    Behavior:
        - If `model` is a `Schema`, it must declare `resource_type`; schema-level migration/validator
        should be provided on the `Schema` itself.
        - If the model is a Pydantic type, it is converted to a struct for storage and the Pydantic
        model can be used for validation.
        - Ref relationships are collected from `Ref` / `RefRevision` annotations for later route and
        referential integrity setup.
        - Ref fields (resource_id refs only) are auto-indexed for searchability.
        - For Job models with a message queue enabled, `status` and `retries` are auto-indexed
        (if not already present in `indexed_fields`).

    Raises:
        ValueError:
            - if the resource name already exists
            - if `Schema` is passed as first argument but `migration`/`validator` is also provided
            - if `Ref(..., on_delete=set_null)` is used on a non-optional field
        TypeError:
            - if `indexed_fields` contains an invalid item

    Examples:
        Basic registration:

        ```python
        from autocrud import AutoCRUD

        autocrud = AutoCRUD()
        autocrud.add_model(User)
        ```

        Custom resource name:

        ```python
        autocrud.add_model(User, name="people")
        ```

        Provide explicit storage:

        ```python
        # storage is per-model; if you want a default for all models, pass `storage_factory=...`
        # when constructing AutoCRUD / calling configure().
        model_name = "people"
        st = autocrud.storage_factory.build(model_name)
        autocrud.add_model(User, name=model_name, storage=st)
        ```

        Using Schema as the first argument:

        ```python
        schema = Schema(User, version="v1")
        autocrud.add_model(schema)
        ```
    """
    _indexed_fields: list[IndexableField] = []
    for field in indexed_fields or []:
        if isinstance(field, IndexableField):
            _indexed_fields.append(field)
        elif (
            isinstance(field, tuple)
            and len(field) == 2
            and isinstance(field[0], str)
        ):
            field = IndexableField(field_path=field[0], field_type=field[1])
            _indexed_fields.append(field)
        elif isinstance(field, str):
            field = IndexableField(field_path=field, field_type=UNSET)
            _indexed_fields.append(field)
        else:
            raise TypeError(
                "Invalid indexed field, should be IndexableField or tuple[field_name, field_type]",
            )

    # ── Resolve Schema vs type argument ────────────────────────
    import warnings

    resolved_schema: Schema | None = None
    if isinstance(model, Schema):
        # Schema passed as first argument
        if migration is not None:
            raise ValueError(
                "Cannot specify 'migration' when passing Schema as the first argument. "
                "Define migration steps on the Schema instead."
            )
        if validator is not None:
            raise ValueError(
                "Cannot specify 'validator' when passing Schema as the first argument. "
                "Pass validator to Schema(..., validator=...) instead."
            )
        resolved_schema = model
        model = resolved_schema.resource_type  # type: ignore[assignment]
        if model is None:
            raise ValueError(
                "Schema passed as first argument must have a resource_type."
            )
    else:
        # model is a plain type
        if isinstance(migration, Schema):
            resolved_schema = migration
        elif isinstance(migration, IMigration):
            warnings.warn(
                "Passing IMigration to migration= is deprecated. "
                "Use Schema(resource_type, version).step(...) instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            resolved_schema = Schema.from_legacy(migration)
        # else migration is None → no schema

    model_name = name or self._resource_name(model)

    # Handle Pydantic BaseModel as model type:
    # auto-generate struct and use Pydantic for validation
    pydantic_model = None
    if is_pydantic_model(model):
        pydantic_model = model
        model = pydantic_to_struct(pydantic_model)
        if validator is None and (
            resolved_schema is None or not resolved_schema.has_validator
        ):
            validator = pydantic_model

    if model_name in self.resource_managers:
        raise ValueError(f"Model name {model_name} already exists.")
    if model in self.model_names:
        self.model_names[model] = None
        logger.warning(
            f"Model {get_type_name(model) or repr(model)} is already registered with a different name. "
            f"This resource manager will not be accessible by its type.",
        )
    else:
        self.model_names[model] = model_name
    if storage is None:
        storage = self.storage_factory.build(model_name)
    if encoding is None:
        encoding = self.default_encoding
    other_options = {}
    if default_status is not None:
        other_options["default_status"] = default_status
    if default_user is not UNSET:
        other_options["default_user"] = default_user
    elif self.default_user is not UNSET:
        other_options["default_user"] = self.default_user
    if default_now is not UNSET:
        other_options["default_now"] = default_now
    elif self.default_now is not UNSET:
        other_options["default_now"] = self.default_now
    # Auto-detect Job subclass and create message queue
    if self._is_job_subclass(model) and (
        job_handler is not None or job_handler_factory is not None
    ):
        # Determine which factory to use
        if message_queue_factory is UNSET:
            mq_factory = self.message_queue_factory
        elif message_queue_factory is None:
            mq_factory = None  # Explicitly disabled
        else:
            mq_factory = message_queue_factory

        if mq_factory is not None:
            real_handler = job_handler
            if job_handler_factory is not None:
                real_handler = LazyJobHandler(job_handler_factory)

            # Create message queue with job handler
            other_options["message_queue"] = mq_factory.build(real_handler)

            # Check if status is already in indexed fields
            if not any(field.field_path == "status" for field in _indexed_fields):
                _indexed_fields.append(
                    IndexableField(field_path="status", field_type=TaskStatus)
                )

            # Check if retries is already in indexed fields
            if not any(field.field_path == "retries" for field in _indexed_fields):
                _indexed_fields.append(
                    IndexableField(field_path="retries", field_type=int)
                )

    resource_manager = ResourceManager(
        model,
        storage=storage,
        blob_store=self.blob_store,
        id_generator=id_generator,
        migration=resolved_schema or migration,
        indexed_fields=_indexed_fields,
        event_handlers=self.event_handlers or event_handlers,
        permission_checker=self.permission_checker or permission_checker,
        encoding=encoding,
        name=model_name,
        validator=validator,
        pydantic_type=pydantic_model,
        constraint_checkers=constraint_checkers,
        strict_operation_context=self.strict_operation_context,
        **other_options,
    )
    self.resource_managers[model_name] = resource_manager

    # Scan Ref / RefRevision annotations and collect relationships
    refs = extract_refs(model, model_name)
    self.relationships.extend(refs)
    # Validate set_null requires nullable field
    for ref_info in refs:
        if ref_info.on_delete == OnDelete.set_null and not ref_info.nullable:
            raise ValueError(
                f"Ref on '{get_type_name(model) or repr(model)}.{ref_info.source_field}' uses "
                f"on_delete=set_null but the field is not Optional. "
                f"Use Annotated[str | None, Ref(...)] instead."
            )

    # Auto-index Ref fields (resource_id refs only) for searchability
    for ref_info in refs:
        if ref_info.ref_type == "resource_id":
            # Use list[str] for list refs, str for scalar refs
            field_type = list[str] if ref_info.is_list else str
            resource_manager.add_indexed_field(
                IndexableField(
                    field_path=ref_info.source_field,
                    field_type=field_type,
                )
            )

openapi

openapi(app: FastAPI, structs: list[type] = None) -> None

Generate and register the OpenAPI schema for the FastAPI application.

This method customizes the OpenAPI schema generation to include all the AutoCRUD-specific types, models, and response schemas. It ensures that the generated API documentation (Swagger UI / ReDoc) correctly reflects the structure of your resources and their endpoints.

PARAMETER	DESCRIPTION
app	The FastAPI application instance. TYPE: FastAPI
structs	Optional list of additional msgspec Structs to include in the schema. TYPE: list[type] DEFAULT: None

Note

When :meth:apply is called with a FastAPI instance as the first argument, this method is called automatically at the end of apply(). You only need to call it manually if you passed a bare APIRouter to apply() or need to customise the structs parameter separately.

Source code in autocrud/crud/core.py

def openapi(self, app: FastAPI, structs: list[type] = None) -> None:
    """Generate and register the OpenAPI schema for the FastAPI application.

    This method customizes the OpenAPI schema generation to include all the
    AutoCRUD-specific types, models, and response schemas. It ensures that
    the generated API documentation (Swagger UI / ReDoc) correctly reflects
    the structure of your resources and their endpoints.

    Args:
        app: The FastAPI application instance.
        structs: Optional list of additional msgspec Structs to include in the schema.

    Note:
        When :meth:`apply` is called with a ``FastAPI`` instance as the
        first argument, this method is called automatically at the end of
        ``apply()``.  You only need to call it manually if you passed a
        bare ``APIRouter`` to ``apply()`` or need to customise the
        ``structs`` parameter separately.
    """

    # Handle root_path by setting servers if not already set
    structs = structs or []
    servers = app.servers
    if app.root_path and not servers:
        servers = [{"url": app.root_path}]

    app.openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        openapi_version=app.openapi_version,
        summary=app.summary,
        description=app.description,
        terms_of_service=app.terms_of_service,
        contact=app.contact,
        license_info=app.license_info,
        routes=app.routes,
        webhooks=app.webhooks.routes,
        tags=app.openapi_tags,
        servers=servers,
        separate_input_output_schemas=app.separate_input_output_schemas,
    )
    app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
        [
            ResourceMeta,
            RevisionInfo,
            RevisionListResponse,
            *[rm.resource_type for rm in self.resource_managers.values()],
            *[
                FullResourceResponse[rm.resource_type]
                for rm in self.resource_managers.values()
            ],
            RFC6902_Add,
            RFC6902_Remove,
            RFC6902_Replace,
            RFC6902_Move,
            RFC6902_Test,
            RFC6902_Copy,
            RFC6902,
            *structs,
        ],
    )[1]

    # Include MigrateProgress and MigrateResult when MigrateRouteTemplate is active
    if any(isinstance(rt, MigrateRouteTemplate) for rt in self.route_templates):
        app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
            [MigrateProgress, MigrateResult],
        )[1]

    # Include custom create action body schemas in components
    action_body_structs = []
    for action in self._pending_create_actions:
        if action.resource_name not in self.resource_managers:
            warnings.warn(
                f"Resource '{action.resource_name}' not found in resource managers. "
                f"Skipping action '{action.handler.__name__}'.",
                stacklevel=2,
            )
            continue
        action_body_structs.extend(self._collect_action_body_structs(action))
    # Also include custom update action body schemas
    for action in self._pending_update_actions:
        if action.resource_name not in self.resource_managers:
            continue
        action_body_structs.extend(
            self._collect_action_body_structs(
                action,
                skip_params={
                    action.existing_param,
                    action.info_param,
                    action.meta_param,
                },
            )
        )
    if action_body_structs:
        app.openapi_schema["components"]["schemas"] |= jsonschema_to_openapi(
            action_body_structs,
        )[1]

    # Inject x-ref-* / x-ref-revision-* metadata into schema properties
    self._inject_ref_metadata(app.openapi_schema)

    # Inject x-autocrud-custom-create-actions top-level extension
    self._inject_custom_create_actions(app.openapi_schema)

    # Inject x-autocrud-custom-update-actions top-level extension
    self._inject_custom_update_actions(app.openapi_schema)

    # Inject x-autocrud-async-create-jobs mapping (job resource → parent)
    self._inject_async_create_jobs(app.openapi_schema)

    # Inject x-autocrud-async-update-jobs mapping (job resource → parent)
    self._inject_async_update_jobs(app.openapi_schema)

    # Inject x-autocrud-indexed-fields mapping (resource → indexed field paths)
    self._inject_indexed_fields(app.openapi_schema)

    # Promote inline $defs (from Pydantic-generated schemas) into
    # components/schemas and rewrite $ref paths so swagger-parser can
    # resolve them.  Must run before _resolve_missing_schema_refs.
    self._promote_defs_to_components(app.openapi_schema)

    # Resolve dangling $ref pointers caused by module-qualified name
    # divergence (e.g. route $ref → "Skill" but components only has
    # "__main___Skill" due to pydantic_to_struct round-trip duplication).
    self._resolve_missing_schema_refs(app.openapi_schema)

apply

apply(
    app: FastAPI | APIRouter,
    *,
    router: APIRouter | None = None,
    structs: list[type] | None = None,
    auto_include: bool = True,
) -> APIRouter

Apply all route templates to generate API endpoints.

This method generates all the CRUD endpoints for all registered models. When app is a :class:~fastapi.FastAPI instance, the OpenAPI schema is automatically customised via :meth:openapi after route generation.

PARAMETER	DESCRIPTION
app	The FastAPI application or an APIRouter to attach routes to. When a FastAPI instance is provided, :meth:openapi is called automatically after route generation. TYPE: FastAPI | APIRouter
router	Optional sub-router. When provided, routes are generated on this router instead of directly on app. If auto_include is True and app is a FastAPI instance, the router is automatically included on app via app.include_router(router) before OpenAPI generation. TYPE: APIRouter | None DEFAULT: None
structs	Additional msgspec.Struct types to include in the OpenAPI components/schemas. Forwarded to :meth:openapi. TYPE: list[type] | None DEFAULT: None
auto_include	When True (the default) and both app is a FastAPI instance and router is provided, automatically call app.include_router(router) so that the sub-router's routes are reachable and visible in the OpenAPI schema. Set to False if you have already called app.include_router(router) yourself. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
APIRouter	The router that routes were generated on — either router
APIRouter	(if provided) or app.

Example

from fastapi import FastAPI, APIRouter
from autocrud import AutoCRUD

app = FastAPI()
autocrud = AutoCRUD()
autocrud.add_model(User)
autocrud.add_model(Post)

# 1. Simplest — routes on app, auto OpenAPI
autocrud.apply(app)

# 2. With a sub-router — auto include + auto OpenAPI
api_router = APIRouter(prefix="/api/v1")
autocrud.apply(app, router=api_router)

# 3. Manual include (e.g. already included elsewhere)
api_router = APIRouter(prefix="/api/v1")
autocrud.apply(app, router=api_router, auto_include=False)
app.include_router(api_router)
autocrud.openapi(app)

# 4. Pure APIRouter (no FastAPI, no OpenAPI)
api_router = APIRouter(prefix="/api/v1")
autocrud.apply(api_router)

Note

• Call this method after adding all models and custom route templates.
• When app is a bare APIRouter, OpenAPI customisation is skipped (APIRouter has no OpenAPI schema).
• structs is ignored when app is not a FastAPI instance.

Source code in autocrud/crud/core.py

def apply(
    self,
    app: FastAPI | APIRouter,
    *,
    router: APIRouter | None = None,
    structs: list[type] | None = None,
    auto_include: bool = True,
) -> APIRouter:
    """Apply all route templates to generate API endpoints.

    This method generates all the CRUD endpoints for all registered models.
    When ``app`` is a :class:`~fastapi.FastAPI` instance, the OpenAPI schema
    is automatically customised via :meth:`openapi` after route generation.

    Args:
        app: The FastAPI application or an APIRouter to attach routes to.
            When a ``FastAPI`` instance is provided, :meth:`openapi` is
            called automatically after route generation.
        router: Optional sub-router.  When provided, routes are generated
            on this router instead of directly on ``app``.  If
            ``auto_include`` is ``True`` and ``app`` is a ``FastAPI``
            instance, the router is automatically included on ``app``
            via ``app.include_router(router)`` before OpenAPI generation.
        structs: Additional ``msgspec.Struct`` types to include in the
            OpenAPI ``components/schemas``.  Forwarded to :meth:`openapi`.
        auto_include: When ``True`` (the default) and both ``app`` is a
            ``FastAPI`` instance and ``router`` is provided, automatically
            call ``app.include_router(router)`` so that the sub-router's
            routes are reachable and visible in the OpenAPI schema.
            Set to ``False`` if you have already called
            ``app.include_router(router)`` yourself.

    Returns:
        The router that routes were generated on — either ``router``
        (if provided) or ``app``.

    Example:
        ```python
        from fastapi import FastAPI, APIRouter
        from autocrud import AutoCRUD

        app = FastAPI()
        autocrud = AutoCRUD()
        autocrud.add_model(User)
        autocrud.add_model(Post)

        # 1. Simplest — routes on app, auto OpenAPI
        autocrud.apply(app)

        # 2. With a sub-router — auto include + auto OpenAPI
        api_router = APIRouter(prefix="/api/v1")
        autocrud.apply(app, router=api_router)

        # 3. Manual include (e.g. already included elsewhere)
        api_router = APIRouter(prefix="/api/v1")
        autocrud.apply(app, router=api_router, auto_include=False)
        app.include_router(api_router)
        autocrud.openapi(app)

        # 4. Pure APIRouter (no FastAPI, no OpenAPI)
        api_router = APIRouter(prefix="/api/v1")
        autocrud.apply(api_router)
        ```

    Note:
        - Call this method after adding all models and custom route templates.
        - When ``app`` is a bare ``APIRouter``, OpenAPI customisation is
          skipped (``APIRouter`` has no OpenAPI schema).
        - ``structs`` is ignored when ``app`` is not a ``FastAPI`` instance.
    """
    # Determine the target router for route generation
    target = router if router is not None else app

    # Validate all Ref targets point to registered resources
    registered = set(self.resource_managers.keys())
    for ref_info in self.relationships:
        if ref_info.target not in registered:
            logger.warning(
                f"Ref on '{ref_info.source}.{ref_info.source_field}' targets "
                f"resource '{ref_info.target}' which is not registered. "
                f"The reference will be dangling at runtime."
            )

    # Install referential integrity event handlers
    self._install_ref_integrity_handlers()

    # Auto-register Job models for async create actions BEFORE applying
    # route templates so the Jobs get their own CRUD endpoints.
    self._register_async_job_models()

    # Auto-register Job models for async update actions.
    self._register_async_update_job_models()

    self.route_templates.sort()
    for model_name, resource_manager in self.resource_managers.items():
        for route_template in self.route_templates:
            try:
                route_template.apply(model_name, resource_manager, target)
            except Exception:
                pass

    # Register custom create action routes
    self._apply_create_actions(target)

    # Register custom update action routes
    self._apply_update_actions(target)

    # Add ref-specific routes (referrers + relationships)
    self._apply_ref_routes(target)

    # Global backup / restore endpoints
    self._apply_backup_routes(target)

    # Auto include_router + auto openapi when app is a FastAPI instance
    is_fastapi = isinstance(app, FastAPI)
    if is_fastapi:
        if router is not None and auto_include:
            app.include_router(router)
        # Only generate OpenAPI when routes are actually on the app.
        # When router is provided but auto_include is False, the routes
        # live on the sub-router and are not yet reachable from app.routes,
        # so skip openapi and let the user call it manually.
        if router is None or auto_include:
            self.openapi(app, structs or [])

    return target

dump

dump(
    bio: IO[bytes],
    model_queries: dict[
        str, Query | ResourceMetaSearchQuery | None
    ]
    | None = None,
) -> None

Export resources to a streaming msgpack archive.

PARAMETER	DESCRIPTION
bio	Binary I/O stream to write to. TYPE: IO[bytes]
model_queries	Optional {model_name: QB_query} mapping. When None, all registered models are exported in full. When provided, only the listed models are exported; each value is a Query / ResourceMetaSearchQuery (or None for "all resources of that model"). TYPE: dict[str, Query | ResourceMetaSearchQuery | None] | None DEFAULT: None

Example::

# Dump everything
with open("backup.acbak", "wb") as f:
    autocrud.dump(f)

# Dump only User resources where name == "Alice"
from autocrud.query import QB

with open("backup.acbak", "wb") as f:
    autocrud.dump(f, model_queries={"user": QB.name == "Alice"})

Source code in autocrud/crud/core.py

def dump(
    self,
    bio: IO[bytes],
    model_queries: dict[str, Query | ResourceMetaSearchQuery | None] | None = None,
) -> None:
    """Export resources to a streaming msgpack archive.

    Args:
        bio: Binary I/O stream to write to.
        model_queries: Optional ``{model_name: QB_query}`` mapping.
            When *None*, all registered models are exported in full.
            When provided, only the listed models are exported;
            each value is a ``Query`` / ``ResourceMetaSearchQuery``
            (or *None* for "all resources of that model").

    Example::

        # Dump everything
        with open("backup.acbak", "wb") as f:
            autocrud.dump(f)

        # Dump only User resources where name == "Alice"
        from autocrud.query import QB

        with open("backup.acbak", "wb") as f:
            autocrud.dump(f, model_queries={"user": QB.name == "Alice"})
    """
    from autocrud.resource_manager.dump_format import (
        DumpStreamWriter,
        EofRecord,
        HeaderRecord,
        ModelEndRecord,
        ModelStartRecord,
    )

    writer = DumpStreamWriter(bio)
    writer.write(HeaderRecord())

    # Determine which models to dump
    if model_queries is None:
        models_to_dump = {name: None for name in self.resource_managers}
    else:
        models_to_dump = model_queries

    for model_name, query in models_to_dump.items():
        if model_name not in self.resource_managers:
            raise ValueError(
                f"Model '{model_name}' not found in resource managers."
            )
        mgr = self.resource_managers[model_name]
        writer.write(ModelStartRecord(model_name=model_name))
        for record in mgr.dump(query=query):
            writer.write(record)
        writer.write(ModelEndRecord(model_name=model_name))

    writer.write(EofRecord())

load

load(
    bio: IO[bytes],
    on_duplicate: "OnDuplicate | None" = None,
) -> dict[str, "LoadStats"]

Import resources from a streaming msgpack archive.

PARAMETER	DESCRIPTION
bio	Binary I/O stream to read from. TYPE: IO[bytes]
on_duplicate	Strategy for duplicate resource IDs. Defaults to OnDuplicate.overwrite. TYPE: 'OnDuplicate | None' DEFAULT: None

RETURNS	DESCRIPTION
dict[str, 'LoadStats']	Per-model load statistics: {model_name: LoadStats}.

RAISES	DESCRIPTION
ValueError	If the archive format is invalid or contains unknown models.

Source code in autocrud/crud/core.py

def load(
    self,
    bio: IO[bytes],
    on_duplicate: "OnDuplicate | None" = None,
) -> dict[str, "LoadStats"]:
    """Import resources from a streaming msgpack archive.

    Args:
        bio: Binary I/O stream to read from.
        on_duplicate: Strategy for duplicate resource IDs.
            Defaults to ``OnDuplicate.overwrite``.

    Returns:
        Per-model load statistics: ``{model_name: LoadStats}``.

    Raises:
        ValueError: If the archive format is invalid or contains
            unknown models.
    """
    from autocrud.resource_manager.dump_format import (
        BlobRecord,
        DumpStreamReader,
        EofRecord,
        HeaderRecord,
        MetaRecord,
        ModelEndRecord,
        ModelStartRecord,
        RevisionRecord,
    )
    from autocrud.types import OnDuplicate as _OnDuplicate

    if on_duplicate is None:
        on_duplicate = _OnDuplicate.overwrite

    reader = DumpStreamReader(bio)
    stats: dict[str, LoadStats] = {}

    # Read header
    first = next(reader)
    if not isinstance(first, HeaderRecord):
        raise ValueError(f"Expected HeaderRecord, got {type(first).__name__}.")
    if first.version != 2:
        raise ValueError(f"Unsupported dump format version {first.version}.")

    current_model: str | None = None
    current_mgr = None
    # Per-model record buffers for bulk load
    meta_buf: list[MetaRecord] = []
    rev_buf: list[RevisionRecord] = []
    blob_buf: list[BlobRecord] = []

    for record in reader:
        if isinstance(record, ModelStartRecord):
            current_model = record.model_name
            if current_model not in self.resource_managers:
                raise ValueError(
                    f"Model '{current_model}' not found in resource managers."
                )
            current_mgr = self.resource_managers[current_model]
            meta_buf.clear()
            rev_buf.clear()
            blob_buf.clear()
            if current_model not in stats:
                stats[current_model] = LoadStats()

        elif isinstance(record, ModelEndRecord):
            # Flush buffered records via bulk load
            if current_mgr is not None and current_model is not None:
                st = current_mgr.load_records_bulk(
                    meta_buf,
                    rev_buf,
                    blob_buf,
                    on_duplicate=on_duplicate,
                )
                s = stats[current_model]
                s.loaded += st.loaded
                s.skipped += st.skipped
                s.total += st.total
            current_model = None
            current_mgr = None
            meta_buf.clear()
            rev_buf.clear()
            blob_buf.clear()

        elif isinstance(record, MetaRecord):
            if current_mgr is None:
                raise ValueError("MetaRecord outside of model section.")
            meta_buf.append(record)

        elif isinstance(record, RevisionRecord):
            if current_mgr is None:
                raise ValueError("RevisionRecord outside of model section.")
            rev_buf.append(record)

        elif isinstance(record, BlobRecord):
            if current_mgr is None:
                raise ValueError("BlobRecord outside of model section.")
            blob_buf.append(record)

        elif isinstance(record, EofRecord):
            break

    return stats

LoadStats

Per-model statistics returned by :meth:AutoCRUD.load.

Source code in autocrud/crud/core.py

class LoadStats:
    """Per-model statistics returned by :meth:`AutoCRUD.load`."""

    __slots__ = ("loaded", "skipped", "total")

    def __init__(self) -> None:
        self.loaded = 0
        self.skipped = 0
        self.total = 0

    def __repr__(self) -> str:
        return (
            f"LoadStats(loaded={self.loaded}, skipped={self.skipped}, "
            f"total={self.total})"
        )

Attributes

loaded instance-attribute

loaded = 0

skipped instance-attribute

skipped = 0

total instance-attribute

total = 0

ResourceOps

Bases: Generic[T]

Context-capturing proxy returned by :meth:ResourceManager.using.

ResourceOps captures user, now, and resource_id at creation time. Each method call re-applies these values via the manager's context system, ensuring that multiple ResourceOps instances created from the same manager do not interfere with each other.

This enables the multiple using() pattern::

with mgr.using(user="u1") as op1, mgr.using(user="u2") as op2:
    op1.create(data1)  # created_by = "u1"
    op2.create(data2)  # created_by = "u2"

After the with block exits, or if an exception propagates, the proxy is deactivated and any subsequent method call raises :class:RuntimeError.

Note

Calling op.using(...) or op.meta_provide(...) is forbidden and raises :class:RuntimeError.

Source code in autocrud/resource_manager/core.py

class ResourceOps(Generic[T]):
    """Context-capturing proxy returned by :meth:`ResourceManager.using`.

    ``ResourceOps`` captures ``user``, ``now``, and ``resource_id`` at
    creation time.  Each method call **re-applies** these values via the
    manager's context system, ensuring that multiple ``ResourceOps``
    instances created from the same manager do not interfere with each
    other.

    This enables the *multiple using()* pattern::

        with mgr.using(user="u1") as op1, mgr.using(user="u2") as op2:
            op1.create(data1)  # created_by = "u1"
            op2.create(data2)  # created_by = "u2"

    After the ``with`` block exits, or if an exception propagates, the
    proxy is **deactivated** and any subsequent method call raises
    :class:`RuntimeError`.

    Note:
        Calling ``op.using(...)`` or ``op.meta_provide(...)`` is
        forbidden and raises :class:`RuntimeError`.
    """

    __slots__ = ("_mgr", "_user", "_now", "_resource_id", "_active")

    def __init__(
        self,
        mgr: "IResourceManager[T]",
        user: "str | UnsetType",
        now: "dt.datetime | UnsetType",
        resource_id: "str | UnsetType",
    ) -> None:
        object.__setattr__(self, "_mgr", mgr)
        object.__setattr__(self, "_user", user)
        object.__setattr__(self, "_now", now)
        object.__setattr__(self, "_resource_id", resource_id)
        object.__setattr__(self, "_active", True)

    def _deactivate(self) -> None:
        """Mark this proxy as inactive (called automatically on context exit)."""
        object.__setattr__(self, "_active", False)

    def __getattr__(self, name: str) -> Any:
        if name in ("using", "meta_provide"):
            raise RuntimeError("Cannot rebind context through ResourceOps")
        if not self._active:
            raise RuntimeError("ResourceOps is no longer active")
        attr = getattr(self._mgr, name)
        if inspect.ismethod(attr):

            @wraps(attr)
            def _wrapper(*args: Any, **kwargs: Any) -> Any:
                if not self._active:
                    raise RuntimeError("ResourceOps is no longer active")
                with self._mgr._apply_context(
                    self._user, self._now, resource_id=self._resource_id
                ):
                    return attr(*args, **kwargs)

            return _wrapper
        return attr

    if TYPE_CHECKING:
        # -- Stubs for IDE auto-complete / type checking --
        def create(
            self,
            data: T,
            *,
            status: RevisionStatus | UnsetType = ...,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
            resource_id: str | UnsetType = ...,
        ) -> RevisionInfo: ...
        def update(
            self,
            resource_id: str,
            data: T,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> RevisionInfo: ...
        def create_or_update(
            self,
            resource_id: str,
            data: T,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> RevisionInfo: ...
        def get(
            self,
            resource_id: str,
            *,
            revision_id: str | UnsetType = ...,
            schema_version: str | None | UnsetType = ...,
        ) -> Resource[T]: ...
        def get_partial(
            self,
            resource_id: str,
            revision_id: str,
            partial: Iterable[str | JsonPointer],
        ) -> Struct: ...
        def get_revision_info(
            self,
            resource_id: str,
            revision_id: str | UnsetType = ...,
        ) -> RevisionInfo: ...
        def get_resource_revision(
            self,
            resource_id: str,
            revision_id: str,
            schema_version: str | None | UnsetType = ...,
        ) -> Resource[T]: ...
        def list_revisions(self, resource_id: str) -> list[str]: ...
        def get_meta(
            self, resource_id: str, include_deleted: bool = ...
        ) -> ResourceMeta: ...
        def exists(self, resource_id: str) -> bool: ...
        def revision_exists(self, resource_id: str, revision_id: str) -> bool: ...
        def count_resources(self, query: ResourceMetaSearchQuery) -> int: ...
        def search_resources(
            self, query: ResourceMetaSearchQuery
        ) -> list[ResourceMeta]: ...
        def list_resources(
            self,
            query: ResourceMetaSearchQuery,
            *,
            returns: list[str] | None = ...,
            partial: list[str] | None = ...,
        ) -> list[SearchedResource[T]]: ...
        def modify(
            self,
            resource_id: str,
            data: T | JsonPatch | UnsetType = ...,
            status: RevisionStatus | UnsetType = ...,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> RevisionInfo: ...
        def patch(
            self,
            resource_id: str,
            patch_data: JsonPatch,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> RevisionInfo: ...
        def switch(
            self,
            resource_id: str,
            revision_id: str,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> ResourceMeta: ...
        def delete(
            self,
            resource_id: str,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> ResourceMeta: ...
        def restore(
            self,
            resource_id: str,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> ResourceMeta: ...
        def permanently_delete(
            self,
            resource_id: str,
            *,
            user: str | UnsetType = ...,
            now: dt.datetime | UnsetType = ...,
        ) -> ResourceMeta: ...
        def migrate(
            self,
            resource_id: str,
            *,
            revision_id: str | UnsetType = ...,
        ) -> ResourceMeta: ...
        def get_blob(self, file_id: str) -> Binary: ...
        def get_blob_url(self, file_id: str) -> str | None: ...

Functions

create

create(
    data: T,
    *,
    status: RevisionStatus | UnsetType = ...,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
    resource_id: str | UnsetType = ...,
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def create(
    self,
    data: T,
    *,
    status: RevisionStatus | UnsetType = ...,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
    resource_id: str | UnsetType = ...,
) -> RevisionInfo: ...

update

update(
    resource_id: str,
    data: T,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def update(
    self,
    resource_id: str,
    data: T,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> RevisionInfo: ...

create_or_update

create_or_update(
    resource_id: str,
    data: T,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def create_or_update(
    self,
    resource_id: str,
    data: T,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> RevisionInfo: ...

get

get(
    resource_id: str,
    *,
    revision_id: str | UnsetType = ...,
    schema_version: str | None | UnsetType = ...,
) -> Resource[T]

Source code in autocrud/resource_manager/core.py

def get(
    self,
    resource_id: str,
    *,
    revision_id: str | UnsetType = ...,
    schema_version: str | None | UnsetType = ...,
) -> Resource[T]: ...

get_partial

get_partial(
    resource_id: str,
    revision_id: str,
    partial: Iterable[str | JsonPointer],
) -> Struct

Source code in autocrud/resource_manager/core.py

def get_partial(
    self,
    resource_id: str,
    revision_id: str,
    partial: Iterable[str | JsonPointer],
) -> Struct: ...

get_revision_info

get_revision_info(
    resource_id: str, revision_id: str | UnsetType = ...
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def get_revision_info(
    self,
    resource_id: str,
    revision_id: str | UnsetType = ...,
) -> RevisionInfo: ...

get_resource_revision

get_resource_revision(
    resource_id: str,
    revision_id: str,
    schema_version: str | None | UnsetType = ...,
) -> Resource[T]

Source code in autocrud/resource_manager/core.py

def get_resource_revision(
    self,
    resource_id: str,
    revision_id: str,
    schema_version: str | None | UnsetType = ...,
) -> Resource[T]: ...

list_revisions

list_revisions(resource_id: str) -> list[str]

Source code in autocrud/resource_manager/core.py

def list_revisions(self, resource_id: str) -> list[str]: ...

get_meta

get_meta(
    resource_id: str, include_deleted: bool = ...
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def get_meta(
    self, resource_id: str, include_deleted: bool = ...
) -> ResourceMeta: ...

exists

exists(resource_id: str) -> bool

Source code in autocrud/resource_manager/core.py

def exists(self, resource_id: str) -> bool: ...

revision_exists

revision_exists(resource_id: str, revision_id: str) -> bool

Source code in autocrud/resource_manager/core.py

def revision_exists(self, resource_id: str, revision_id: str) -> bool: ...

count_resources

count_resources(query: ResourceMetaSearchQuery) -> int

Source code in autocrud/resource_manager/core.py

def count_resources(self, query: ResourceMetaSearchQuery) -> int: ...

search_resources

search_resources(
    query: ResourceMetaSearchQuery,
) -> list[ResourceMeta]

Source code in autocrud/resource_manager/core.py

def search_resources(
    self, query: ResourceMetaSearchQuery
) -> list[ResourceMeta]: ...

list_resources

list_resources(
    query: ResourceMetaSearchQuery,
    *,
    returns: list[str] | None = ...,
    partial: list[str] | None = ...,
) -> list[SearchedResource[T]]

Source code in autocrud/resource_manager/core.py

def list_resources(
    self,
    query: ResourceMetaSearchQuery,
    *,
    returns: list[str] | None = ...,
    partial: list[str] | None = ...,
) -> list[SearchedResource[T]]: ...

modify

modify(
    resource_id: str,
    data: T | JsonPatch | UnsetType = ...,
    status: RevisionStatus | UnsetType = ...,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def modify(
    self,
    resource_id: str,
    data: T | JsonPatch | UnsetType = ...,
    status: RevisionStatus | UnsetType = ...,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> RevisionInfo: ...

patch

patch(
    resource_id: str,
    patch_data: JsonPatch,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> RevisionInfo

Source code in autocrud/resource_manager/core.py

def patch(
    self,
    resource_id: str,
    patch_data: JsonPatch,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> RevisionInfo: ...

switch

switch(
    resource_id: str,
    revision_id: str,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def switch(
    self,
    resource_id: str,
    revision_id: str,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> ResourceMeta: ...

delete

delete(
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def delete(
    self,
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> ResourceMeta: ...

restore

restore(
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def restore(
    self,
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> ResourceMeta: ...

permanently_delete

permanently_delete(
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: datetime | UnsetType = ...,
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def permanently_delete(
    self,
    resource_id: str,
    *,
    user: str | UnsetType = ...,
    now: dt.datetime | UnsetType = ...,
) -> ResourceMeta: ...

migrate

migrate(
    resource_id: str, *, revision_id: str | UnsetType = ...
) -> ResourceMeta

Source code in autocrud/resource_manager/core.py

def migrate(
    self,
    resource_id: str,
    *,
    revision_id: str | UnsetType = ...,
) -> ResourceMeta: ...

get_blob

get_blob(file_id: str) -> Binary

Source code in autocrud/resource_manager/core.py

def get_blob(self, file_id: str) -> Binary: ...

get_blob_url

get_blob_url(file_id: str) -> str | None

Source code in autocrud/resource_manager/core.py

def get_blob_url(self, file_id: str) -> str | None: ...

Schema

Bases: Generic[T]

Unified migration + validation descriptor.

Parameters

resource_type : type[T] The data model class (msgspec Struct or Pydantic BaseModel). version : str The target schema version. validator : Callable | IValidator | type | None Optional validator (same types accepted by the old validator= parameter on add_model).

Examples

Simple reindex (version bump, no data change)::

Schema(User, "v2")

Single-step migration::

Schema(User, "v2").step("v1", migrate_v1_to_v2)

Chain migration with auto-inferred to::

Schema(User, "v3").step("v1", fn1).step("v2", fn2)
# fn1: v1 → v2  (inferred from next step's from_ver)
# fn2: v2 → v3  (inferred from Schema target version)

Parallel paths::

Schema(User, "v3").step("v1", fn1).step("v2", fn2).plus("v1", fn_shortcut)
# fn_shortcut: v1 → v3  (last in chain, inferred from target)

With validation::

Schema(User, "v2", validator=my_validator).step("v1", fn)

Source code in autocrud/schema.py

class Schema(Generic[T]):
    """Unified migration + validation descriptor.

    Parameters
    ----------
    resource_type : type[T]
        The data model class (msgspec Struct or Pydantic BaseModel).
    version : str
        The **target** schema version.
    validator : Callable | IValidator | type | None
        Optional validator (same types accepted by the old ``validator=``
        parameter on ``add_model``).

    Examples
    --------
    Simple reindex (version bump, no data change)::

        Schema(User, "v2")

    Single-step migration::

        Schema(User, "v2").step("v1", migrate_v1_to_v2)

    Chain migration with auto-inferred ``to``::

        Schema(User, "v3").step("v1", fn1).step("v2", fn2)
        # fn1: v1 → v2  (inferred from next step's from_ver)
        # fn2: v2 → v3  (inferred from Schema target version)

    Parallel paths::

        Schema(User, "v3").step("v1", fn1).step("v2", fn2).plus("v1", fn_shortcut)
        # fn_shortcut: v1 → v3  (last in chain, inferred from target)

    With validation::

        Schema(User, "v2", validator=my_validator).step("v1", fn)
    """

    def __init__(
        self,
        resource_type: type[T],
        version: str,
        *,
        validator: "Callable | Any | None" = None,
    ):
        self._resource_type: type[T] = resource_type
        self._version: str = version
        self._raw_validator = validator
        self._validator: Callable | None = build_validator(validator)
        # List of flushed chains.  Each chain is a list of
        # (from_ver, fn, source_type) tuples whose ``to`` will be
        # resolved lazily.
        # ``from_ver`` can be a plain string or a compiled regex pattern.
        # ``source_type`` is ``None`` for legacy IO[bytes]-based steps or
        # a concrete type for typed steps.
        self._chains: list[
            list[tuple[str | re.Pattern[str], Callable, type | None]]
        ] = []
        # The chain currently being built (not yet flushed).
        self._current_chain: list[
            tuple[str | re.Pattern[str], Callable, type | None]
        ] = []
        # Explicit ``to`` overrides: {chain_idx: {step_idx: to_ver}}
        self._explicit_to: dict[int, dict[int, str]] = {}
        # Resolved directed graph cache (invalidated on mutation).
        self._graph: dict[str, list[tuple[str, Callable, type | None]]] | None = None
        # Regex edges resolved at build time but expanded at runtime.
        self._regex_edges: list[tuple[re.Pattern[str], str, Callable, type | None]] = []
        # Path cache: (from_ver, to_ver) → path.  Cleared on mutation.
        self._path_cache: dict[
            tuple[str | None, str],
            list[tuple[str, str, Callable, type | None]],
        ] = {}
        # Legacy migration wrapper (set by ``from_legacy``).
        self._legacy_migration: IMigration | None = None
        # Encoder for re-serializing intermediate migration results.
        # Default is JSON; call ``set_encoding()`` to switch to msgpack.
        self._encoder = msgspec.json.Encoder()
        self._encoding: str = "json"

    # ------------------------------------------------------------------
    # Encoding configuration
    # ------------------------------------------------------------------

    def set_encoding(self, encoding: str) -> None:
        """Set the serialization format for intermediate migration results.

        Parameters
        ----------
        encoding : str
            ``"json"`` (default) or ``"msgpack"``.

        This is called automatically by ``ResourceManager`` so that
        multi-step migrations re-encode intermediate results in the
        same format as the stored data.
        """
        self._encoding = encoding
        if encoding == "msgpack":
            self._encoder = msgspec.msgpack.Encoder()
        else:
            self._encoder = msgspec.json.Encoder()

    # ------------------------------------------------------------------
    # Fluent builders
    # ------------------------------------------------------------------

    def step(
        self,
        from_ver: str | re.Pattern[str],
        fn: Callable,
        *,
        to: str | None = None,
        source_type: type | None = None,
    ) -> Schema[T]:
        """Append a migration step to the *current* chain.

        Parameters
        ----------
        from_ver : str | re.Pattern[str]
            Source version that this step handles.  Can be a compiled regex
            pattern (``re.compile(...)``), in which case the step creates
            edges from **every** known version that matches the pattern.
        fn : Callable[[IO[bytes]], Any] | Callable[[source_type], Any]
            Transform function.  When *source_type* is ``None`` (default),
            the function receives ``IO[bytes]`` (legacy behaviour).  When
            *source_type* is provided, the function receives an already-decoded
            instance of that type.
        to : str | None
            Explicit target version.  If ``None`` it is auto-inferred:
            * Middle step → next step's ``from_ver`` (must be a literal string).
            * Last step in a chain → ``Schema.version``.
        source_type : type | None
            When provided, the framework automatically decodes the raw bytes
            into *source_type* before calling *fn*.  This removes the
            boilerplate of ``msgspec.json.decode(data.read(), type=...)``
            inside every migration function.  In multi-step chains, if the
            previous step already returned the expected type, the object is
            passed directly without re-encoding/decoding.

        Examples
        --------
        Legacy (IO[bytes]) style::

            def migrate_v1_to_v2(data: IO[bytes]) -> V2:
                obj = msgspec.json.decode(data.read(), type=V1)
                return V2(name=obj.name, extra="new")


            Schema(V2, "v2").step("v1", migrate_v1_to_v2)

        Typed style (recommended)::

            def migrate_v1_to_v2(data: V1) -> V2:
                return V2(name=data.name, extra="new")


            Schema(V2, "v2").step("v1", migrate_v1_to_v2, source_type=V1)
        """
        self._current_chain.append((from_ver, fn, source_type))
        if to is not None:
            chain_idx = len(self._chains)  # current chain's future index
            step_idx = len(self._current_chain) - 1
            self._explicit_to.setdefault(chain_idx, {})[step_idx] = to
        self._graph = None  # invalidate cache
        self._regex_edges = []
        self._path_cache = {}
        return self

    def plus(
        self,
        from_ver: str | re.Pattern[str],
        fn: Callable,
        *,
        to: str | None = None,
        source_type: type | None = None,
    ) -> Schema[T]:
        """Start a **new** parallel chain with the given first step.

        Same semantics as ``.step()`` but the previous chain is flushed first
        so that its ``to`` versions get resolved independently.

        Parameters
        ----------
        from_ver : str | re.Pattern[str]
            Source version (same as ``.step()``).
        fn : Callable
            Transform function (same as ``.step()``).
        to : str | None
            Explicit target version (same as ``.step()``).
        source_type : type | None
            Typed source (same as ``.step()``).  See ``.step()`` for details.
        """
        # Flush current chain
        if self._current_chain:
            self._chains.append(self._current_chain)
            self._current_chain = []
        # ``.step()`` will use ``len(self._chains)`` as chain_idx for the new
        # chain that is being started.
        return self.step(from_ver, fn, to=to, source_type=source_type)

    # ------------------------------------------------------------------
    # Resolution
    # ------------------------------------------------------------------

    def _resolve(self) -> dict[str, list[tuple[str, Callable, type | None]]]:
        """Lazily resolve all chains into a directed graph.

        Literal ``from_ver`` strings are placed directly into the graph.
        Regex ``from_ver`` patterns are stored separately in
        ``_regex_edges`` and expanded at **runtime** (inside
        ``_edges_for``) so that they can match versions from the
        persistence layer that are not known at definition time.
        """
        if self._graph is not None:
            return self._graph

        # Collect all chains (include current if non-empty).
        all_chains: list[list[tuple[str | re.Pattern[str], Callable, type | None]]] = (
            list(self._chains)
        )
        if self._current_chain:
            all_chains.append(self._current_chain)

        # ── Pass 1: resolve to_ver for every step ─────────────────────
        resolved_steps: list[
            tuple[str | re.Pattern[str], str, Callable, type | None]
        ] = []

        for chain_idx, chain in enumerate(all_chains):
            chain_explicit = self._explicit_to.get(chain_idx, {})
            for i, (from_ver, fn, source_type) in enumerate(chain):
                # Determine ``to_ver``
                if i in chain_explicit:
                    to_ver: str | None = chain_explicit[i]
                elif i + 1 < len(chain):
                    next_from = chain[i + 1][0]
                    if isinstance(next_from, re.Pattern):
                        raise ValueError(
                            f"Cannot infer target version for step from {from_ver!r} "
                            f"because the next step uses a regex pattern. "
                            f"Use explicit to= parameter."
                        )
                    to_ver = next_from
                else:
                    # Last step in chain → target is ``Schema.version``
                    to_ver = self._version
                if to_ver is None:
                    raise ValueError(
                        f"Cannot infer target version for step from {from_ver!r}. "
                        f"Set Schema(version=...) or provide to= explicitly."
                    )
                resolved_steps.append((from_ver, to_ver, fn, source_type))

        # ── Pass 2: separate literal edges vs regex edges ─────────────
        graph: dict[str, list[tuple[str, Callable, type | None]]] = defaultdict(list)
        regex_edges: list[tuple[re.Pattern[str], str, Callable, type | None]] = []

        for from_ver, to_ver, fn, source_type in resolved_steps:
            if isinstance(from_ver, re.Pattern):
                regex_edges.append((from_ver, to_ver, fn, source_type))
            else:
                graph[from_ver].append((to_ver, fn, source_type))

        self._graph = dict(graph)
        self._regex_edges = regex_edges
        return self._graph

    # ------------------------------------------------------------------
    # Runtime edge lookup
    # ------------------------------------------------------------------

    def _edges_for(self, version: str) -> list[tuple[str, Callable, type | None]]:
        """Return outgoing edges for *version* (literal + regex matches).

        Each edge is ``(to_ver, fn, source_type)``.

        Called at runtime so that regex patterns can match versions from
        the persistence layer that were never declared in the Schema.
        """
        self._resolve()
        edges = list(self._graph.get(version, []))
        for pattern, to_ver, fn, source_type in self._regex_edges:
            if version != to_ver and pattern.fullmatch(version):
                edges.append((to_ver, fn, source_type))
        return edges

    # ------------------------------------------------------------------
    # Path finding (BFS shortest path)
    # ------------------------------------------------------------------

    def _find_path(
        self, from_ver: str | None, to_ver: str
    ) -> list[tuple[str, str, Callable, type | None]]:
        """Find shortest path from *from_ver* to *to_ver* via BFS.

        Returns list of ``(src, dst, fn, source_type)`` tuples forming
        the path.  Results are cached per ``(from_ver, to_ver)`` pair.

        Raises ``ValueError`` if no path exists.
        """
        self._resolve()

        if from_ver == to_ver:
            return []

        cache_key = (from_ver, to_ver)
        if cache_key in self._path_cache:
            return self._path_cache[cache_key]

        initial_edges = self._edges_for(from_ver)  # type: ignore[arg-type]
        if not initial_edges:
            raise ValueError(
                f"No migration path from version {from_ver!r} to {to_ver!r}. "
                f"No outgoing edges for {from_ver!r}."
            )

        # BFS
        queue: deque[list[tuple[str, str, Callable, type | None]]] = deque()
        visited: set[str | None] = {from_ver}

        for dst, fn, source_type in initial_edges:
            queue.append([(from_ver, dst, fn, source_type)])
            visited.add(dst)

        while queue:
            path = queue.popleft()
            current = path[-1][1]  # last destination
            if current == to_ver:
                self._path_cache[cache_key] = path
                return path
            for dst, fn, source_type in self._edges_for(current):
                if dst not in visited:
                    visited.add(dst)
                    queue.append([*path, (current, dst, fn, source_type)])

        raise ValueError(
            f"No migration path from {from_ver!r} to {to_ver!r}. "
            f"Reachable versions exhausted."
        )

    # ------------------------------------------------------------------
    # IMigration-compatible interface
    # ------------------------------------------------------------------

    @property
    def schema_version(self) -> str | None:
        """Target schema version (``IMigration`` compat)."""
        return self._version

    def migrate(self, data: IO[bytes], schema_version: str | None) -> T:
        """Migrate *data* from *schema_version* to the target version.

        If a legacy ``IMigration`` is wrapped, delegates to it directly.
        Otherwise uses graph-based BFS path finding and executes the
        transform chain.

        Compatible with the ``IMigration.migrate()`` signature so that
        ``ResourceManager`` can use ``Schema`` as a drop-in replacement.
        """
        # Legacy delegation
        if self._legacy_migration is not None:
            return self._legacy_migration.migrate(data, schema_version)

        target = self._version
        if target is None:  # pragma: no cover — defensive; from_legacy always delegates
            raise ValueError("Schema has no target version; cannot migrate.")

        if schema_version == target:
            # Already at target — return raw bytes for caller to decode.
            return data.read()  # type: ignore[return-value]

        path = self._find_path(schema_version, target)
        if not path:  # pragma: no cover — from_ver==to_ver caught above
            return data.read()  # type: ignore[return-value]

        result: Any = data
        for _src, _dst, fn, source_type in path:
            if source_type is not None:
                # ── Typed step: auto-decode to source_type ────────────
                if isinstance(result, source_type):
                    # Direct pass-through (e.g. previous typed step
                    # returned exactly the type we need).
                    pass
                elif isinstance(result, (io.IOBase, io.BufferedIOBase)):
                    result = self._decode_to_type(result.read(), source_type)
                elif isinstance(result, bytes):
                    result = self._decode_to_type(result, source_type)
                else:
                    # Different decoded object — re-encode then decode.
                    encoded = self._encode_intermediate(result)
                    result = self._decode_to_type(encoded, source_type)
                result = fn(result)
            else:
                # ── Legacy step: fn expects IO[bytes] ─────────────────
                if not isinstance(result, (io.IOBase, io.BufferedIOBase)):
                    result = io.BytesIO(self._encode_intermediate(result))
                result = fn(result)

        return result  # type: ignore[return-value]

    # ------------------------------------------------------------------
    # Intermediate encoding
    # ------------------------------------------------------------------

    def _decode_to_type(self, data: bytes, source_type: type) -> Any:
        """Decode *data* bytes into an instance of *source_type*.

        Uses the Schema's current encoding (json or msgpack).
        For Pydantic ``BaseModel`` subclasses, decodes to a dict first
        and then constructs the model via ``model_validate`` / ``parse_obj``.
        """
        # Check for Pydantic BaseModel
        try:
            from pydantic import BaseModel

            if isinstance(source_type, type) and issubclass(source_type, BaseModel):
                if self._encoding == "msgpack":
                    raw = msgspec.msgpack.decode(data)
                else:
                    raw = msgspec.json.decode(data)
                # Pydantic v2 / v1 compat
                if hasattr(source_type, "model_validate"):
                    return source_type.model_validate(raw)
                return source_type.parse_obj(raw)  # type: ignore[union-attr]
        except ImportError:
            pass
        if self._encoding == "msgpack":
            return msgspec.msgpack.decode(data, type=source_type)
        return msgspec.json.decode(data, type=source_type)

    def _encode_intermediate(self, obj: Any) -> bytes:
        """Encode a decoded intermediate object back to bytes.

        Handles ``bytes``, ``msgspec.Struct`` (via the Schema's encoder),
        and Pydantic ``BaseModel`` (v1/v2 via ``pydantic_to_dict``).
        """
        if isinstance(obj, bytes):
            return obj
        try:
            return self._encoder.encode(obj)
        except TypeError:
            # Pydantic BaseModel — convert to dict first, then encode.
            return self._encoder.encode(pydantic_to_dict(obj))

    # ------------------------------------------------------------------
    # Validation
    # ------------------------------------------------------------------

    def validate(self, data: Any) -> None:
        """Run the attached validator, if any."""
        if self._validator is not None:
            from autocrud.types import ValidationError

            try:
                self._validator(data)
            except ValidationError:
                raise
            except Exception as e:
                raise ValidationError(str(e)) from e

    @property
    def has_validator(self) -> bool:
        """Whether a validator is attached."""
        return self._validator is not None

    @property
    def raw_validator(self) -> Callable | Any | None:
        """The original validator argument (before normalization)."""
        return self._raw_validator

    # ------------------------------------------------------------------
    # Legacy adapter
    # ------------------------------------------------------------------

    @classmethod
    def from_legacy(cls, migration: Any) -> Schema[T]:
        """Wrap an existing ``IMigration`` instance as a ``Schema``.

        The resulting ``Schema`` delegates ``.migrate()`` calls directly
        to the wrapped ``IMigration``.

        Note: the returned Schema has ``resource_type = None`` because
        ``IMigration`` does not carry type information.
        """
        from autocrud.types import IMigration

        if not isinstance(migration, IMigration):
            raise TypeError(
                f"Expected IMigration instance, got {type(migration).__name__}"
            )

        schema: Schema[T] = cls.__new__(cls)
        schema._resource_type = None  # type: ignore[assignment]
        schema._version = migration.schema_version  # type: ignore[assignment]
        schema._raw_validator = None
        schema._validator = None
        schema._chains = []
        schema._current_chain = []
        schema._explicit_to = {}
        schema._graph = None
        schema._regex_edges = []
        schema._path_cache = {}
        schema._legacy_migration = migration
        schema._encoder = msgspec.json.Encoder()
        schema._encoding = "json"
        return schema

    # ------------------------------------------------------------------
    # Introspection
    # ------------------------------------------------------------------

    @property
    def resource_type(self) -> type[T] | None:
        """The resource type this Schema is bound to.

        Returns ``None`` for schemas created via ``from_legacy()``.
        """
        return self._resource_type

    @property
    def has_migration(self) -> bool:
        """Whether this schema defines any migration steps (or wraps legacy)."""
        if self._legacy_migration is not None:
            return True
        return bool(self._current_chain or self._chains)

    @property
    def version(self) -> str | None:
        """The target schema version."""
        return self._version

    def __repr__(self) -> str:
        rt = get_type_name(self._resource_type) or repr(self._resource_type)
        parts = [f"Schema({rt}, {self._version!r}"]
        if self._validator is not None:
            parts[0] += ", validator=..."
        parts[0] += ")"
        all_chains: list[list[tuple[str | re.Pattern[str], Callable, type | None]]] = (
            list(self._chains)
        )
        if self._current_chain:
            all_chains.append(self._current_chain)
        for chain_idx, chain in enumerate(all_chains):
            for step_idx, (from_ver, _fn, src_type) in enumerate(chain):
                if chain_idx > 0 and step_idx == 0:
                    method = ".plus"
                else:
                    method = ".step"
                if isinstance(from_ver, re.Pattern):
                    ver_part = f"re.compile({from_ver.pattern!r})"
                else:
                    ver_part = repr(from_ver)
                if src_type is not None:
                    src_name = get_type_name(src_type) or src_type.__name__
                    parts.append(f"{method}({ver_part}, ..., source_type={src_name})")
                else:
                    parts.append(f"{method}({ver_part}, ...)")
        return "".join(parts)

Attributes

schema_version property

schema_version: str | None

Target schema version (IMigration compat).

has_validator property

has_validator: bool

Whether a validator is attached.

raw_validator property

raw_validator: Callable | Any | None

The original validator argument (before normalization).

resource_type property

resource_type: type[T] | None

The resource type this Schema is bound to.

Returns None for schemas created via from_legacy().

has_migration property

has_migration: bool

Whether this schema defines any migration steps (or wraps legacy).

version property

version: str | None

The target schema version.

Functions

set_encoding

set_encoding(encoding: str) -> None

Set the serialization format for intermediate migration results.

Parameters

encoding : str "json" (default) or "msgpack".

This is called automatically by ResourceManager so that multi-step migrations re-encode intermediate results in the same format as the stored data.

Source code in autocrud/schema.py

def set_encoding(self, encoding: str) -> None:
    """Set the serialization format for intermediate migration results.

    Parameters
    ----------
    encoding : str
        ``"json"`` (default) or ``"msgpack"``.

    This is called automatically by ``ResourceManager`` so that
    multi-step migrations re-encode intermediate results in the
    same format as the stored data.
    """
    self._encoding = encoding
    if encoding == "msgpack":
        self._encoder = msgspec.msgpack.Encoder()
    else:
        self._encoder = msgspec.json.Encoder()

step

step(
    from_ver: str | Pattern[str],
    fn: Callable,
    *,
    to: str | None = None,
    source_type: type | None = None,
) -> Schema[T]

Append a migration step to the current chain.

Parameters

from_ver : str | re.Pattern[str] Source version that this step handles. Can be a compiled regex pattern (re.compile(...)), in which case the step creates edges from every known version that matches the pattern. fn : Callable[[IO[bytes]], Any] | Callable[[source_type], Any] Transform function. When source_type is None (default), the function receives IO[bytes] (legacy behaviour). When source_type is provided, the function receives an already-decoded instance of that type. to : str | None Explicit target version. If None it is auto-inferred: * Middle step → next step's from_ver (must be a literal string). * Last step in a chain → Schema.version. source_type : type | None When provided, the framework automatically decodes the raw bytes into source_type before calling fn. This removes the boilerplate of msgspec.json.decode(data.read(), type=...) inside every migration function. In multi-step chains, if the previous step already returned the expected type, the object is passed directly without re-encoding/decoding.

Examples

Legacy (IO[bytes]) style::

def migrate_v1_to_v2(data: IO[bytes]) -> V2:
    obj = msgspec.json.decode(data.read(), type=V1)
    return V2(name=obj.name, extra="new")


Schema(V2, "v2").step("v1", migrate_v1_to_v2)

Typed style (recommended)::

def migrate_v1_to_v2(data: V1) -> V2:
    return V2(name=data.name, extra="new")


Schema(V2, "v2").step("v1", migrate_v1_to_v2, source_type=V1)

Source code in autocrud/schema.py

def step(
    self,
    from_ver: str | re.Pattern[str],
    fn: Callable,
    *,
    to: str | None = None,
    source_type: type | None = None,
) -> Schema[T]:
    """Append a migration step to the *current* chain.

    Parameters
    ----------
    from_ver : str | re.Pattern[str]
        Source version that this step handles.  Can be a compiled regex
        pattern (``re.compile(...)``), in which case the step creates
        edges from **every** known version that matches the pattern.
    fn : Callable[[IO[bytes]], Any] | Callable[[source_type], Any]
        Transform function.  When *source_type* is ``None`` (default),
        the function receives ``IO[bytes]`` (legacy behaviour).  When
        *source_type* is provided, the function receives an already-decoded
        instance of that type.
    to : str | None
        Explicit target version.  If ``None`` it is auto-inferred:
        * Middle step → next step's ``from_ver`` (must be a literal string).
        * Last step in a chain → ``Schema.version``.
    source_type : type | None
        When provided, the framework automatically decodes the raw bytes
        into *source_type* before calling *fn*.  This removes the
        boilerplate of ``msgspec.json.decode(data.read(), type=...)``
        inside every migration function.  In multi-step chains, if the
        previous step already returned the expected type, the object is
        passed directly without re-encoding/decoding.

    Examples
    --------
    Legacy (IO[bytes]) style::

        def migrate_v1_to_v2(data: IO[bytes]) -> V2:
            obj = msgspec.json.decode(data.read(), type=V1)
            return V2(name=obj.name, extra="new")


        Schema(V2, "v2").step("v1", migrate_v1_to_v2)

    Typed style (recommended)::

        def migrate_v1_to_v2(data: V1) -> V2:
            return V2(name=data.name, extra="new")


        Schema(V2, "v2").step("v1", migrate_v1_to_v2, source_type=V1)
    """
    self._current_chain.append((from_ver, fn, source_type))
    if to is not None:
        chain_idx = len(self._chains)  # current chain's future index
        step_idx = len(self._current_chain) - 1
        self._explicit_to.setdefault(chain_idx, {})[step_idx] = to
    self._graph = None  # invalidate cache
    self._regex_edges = []
    self._path_cache = {}
    return self

plus

plus(
    from_ver: str | Pattern[str],
    fn: Callable,
    *,
    to: str | None = None,
    source_type: type | None = None,
) -> Schema[T]

Start a new parallel chain with the given first step.

Same semantics as .step() but the previous chain is flushed first so that its to versions get resolved independently.

Parameters

from_ver : str | re.Pattern[str] Source version (same as .step()). fn : Callable Transform function (same as .step()). to : str | None Explicit target version (same as .step()). source_type : type | None Typed source (same as .step()). See .step() for details.

Source code in autocrud/schema.py

def plus(
    self,
    from_ver: str | re.Pattern[str],
    fn: Callable,
    *,
    to: str | None = None,
    source_type: type | None = None,
) -> Schema[T]:
    """Start a **new** parallel chain with the given first step.

    Same semantics as ``.step()`` but the previous chain is flushed first
    so that its ``to`` versions get resolved independently.

    Parameters
    ----------
    from_ver : str | re.Pattern[str]
        Source version (same as ``.step()``).
    fn : Callable
        Transform function (same as ``.step()``).
    to : str | None
        Explicit target version (same as ``.step()``).
    source_type : type | None
        Typed source (same as ``.step()``).  See ``.step()`` for details.
    """
    # Flush current chain
    if self._current_chain:
        self._chains.append(self._current_chain)
        self._current_chain = []
    # ``.step()`` will use ``len(self._chains)`` as chain_idx for the new
    # chain that is being started.
    return self.step(from_ver, fn, to=to, source_type=source_type)

migrate

migrate(data: IO[bytes], schema_version: str | None) -> T

Migrate data from schema_version to the target version.

If a legacy IMigration is wrapped, delegates to it directly. Otherwise uses graph-based BFS path finding and executes the transform chain.

Compatible with the IMigration.migrate() signature so that ResourceManager can use Schema as a drop-in replacement.

Source code in autocrud/schema.py

def migrate(self, data: IO[bytes], schema_version: str | None) -> T:
    """Migrate *data* from *schema_version* to the target version.

    If a legacy ``IMigration`` is wrapped, delegates to it directly.
    Otherwise uses graph-based BFS path finding and executes the
    transform chain.

    Compatible with the ``IMigration.migrate()`` signature so that
    ``ResourceManager`` can use ``Schema`` as a drop-in replacement.
    """
    # Legacy delegation
    if self._legacy_migration is not None:
        return self._legacy_migration.migrate(data, schema_version)

    target = self._version
    if target is None:  # pragma: no cover — defensive; from_legacy always delegates
        raise ValueError("Schema has no target version; cannot migrate.")

    if schema_version == target:
        # Already at target — return raw bytes for caller to decode.
        return data.read()  # type: ignore[return-value]

    path = self._find_path(schema_version, target)
    if not path:  # pragma: no cover — from_ver==to_ver caught above
        return data.read()  # type: ignore[return-value]

    result: Any = data
    for _src, _dst, fn, source_type in path:
        if source_type is not None:
            # ── Typed step: auto-decode to source_type ────────────
            if isinstance(result, source_type):
                # Direct pass-through (e.g. previous typed step
                # returned exactly the type we need).
                pass
            elif isinstance(result, (io.IOBase, io.BufferedIOBase)):
                result = self._decode_to_type(result.read(), source_type)
            elif isinstance(result, bytes):
                result = self._decode_to_type(result, source_type)
            else:
                # Different decoded object — re-encode then decode.
                encoded = self._encode_intermediate(result)
                result = self._decode_to_type(encoded, source_type)
            result = fn(result)
        else:
            # ── Legacy step: fn expects IO[bytes] ─────────────────
            if not isinstance(result, (io.IOBase, io.BufferedIOBase)):
                result = io.BytesIO(self._encode_intermediate(result))
            result = fn(result)

    return result  # type: ignore[return-value]

validate

validate(data: Any) -> None

Run the attached validator, if any.

Source code in autocrud/schema.py

def validate(self, data: Any) -> None:
    """Run the attached validator, if any."""
    if self._validator is not None:
        from autocrud.types import ValidationError

        try:
            self._validator(data)
        except ValidationError:
            raise
        except Exception as e:
            raise ValidationError(str(e)) from e

from_legacy classmethod

from_legacy(migration: Any) -> Schema[T]

Wrap an existing IMigration instance as a Schema.

The resulting Schema delegates .migrate() calls directly to the wrapped IMigration.

Note: the returned Schema has resource_type = None because IMigration does not carry type information.

Source code in autocrud/schema.py

@classmethod
def from_legacy(cls, migration: Any) -> Schema[T]:
    """Wrap an existing ``IMigration`` instance as a ``Schema``.

    The resulting ``Schema`` delegates ``.migrate()`` calls directly
    to the wrapped ``IMigration``.

    Note: the returned Schema has ``resource_type = None`` because
    ``IMigration`` does not carry type information.
    """
    from autocrud.types import IMigration

    if not isinstance(migration, IMigration):
        raise TypeError(
            f"Expected IMigration instance, got {type(migration).__name__}"
        )

    schema: Schema[T] = cls.__new__(cls)
    schema._resource_type = None  # type: ignore[assignment]
    schema._version = migration.schema_version  # type: ignore[assignment]
    schema._raw_validator = None
    schema._validator = None
    schema._chains = []
    schema._current_chain = []
    schema._explicit_to = {}
    schema._graph = None
    schema._regex_edges = []
    schema._path_cache = {}
    schema._legacy_migration = migration
    schema._encoder = msgspec.json.Encoder()
    schema._encoding = "json"
    return schema

BackgroundTaskAccepted

Bases: Struct

Response body returned by background create actions (HTTP 202).

When a custom create action uses async_mode='background', the endpoint returns this struct immediately while the handler continues executing in a FastAPI BackgroundTasks worker. Unlike async_mode='job', no Job model is created and the task cannot be tracked from the frontend.

ATTRIBUTE	DESCRIPTION
message	A human-readable acceptance message. TYPE: str

Source code in autocrud/types.py

class BackgroundTaskAccepted(Struct, kw_only=True):
    """Response body returned by background create actions (HTTP 202).

    When a custom create action uses ``async_mode='background'``, the
    endpoint returns this struct immediately while the handler continues
    executing in a FastAPI ``BackgroundTasks`` worker.  Unlike
    ``async_mode='job'``, no Job model is created and the task cannot be
    tracked from the frontend.

    Attributes:
        message: A human-readable acceptance message.
    """

    message: str
    """A human-readable acceptance message."""

Attributes

message instance-attribute

message: str

A human-readable acceptance message.

BlobUploadSession

Bases: Struct

Represents an active or completed blob upload session.

Lifecycle (single upload)::

pending → uploaded → finalized  (or → aborted)

Lifecycle (chunked upload)::

pending → uploading → … → uploading → finalized  (or → aborted)

upload_method indicates how the client should deliver file bytes:

• "proxy": PUT bytes to /blobs/upload-sessions/{upload_id}/content. May be called multiple times for chunked uploads.
• "single_put": upload directly to upload_url (e.g. S3 presigned PUT)

Source code in autocrud/types.py

class BlobUploadSession(Struct, kw_only=True):
    """Represents an active or completed blob upload session.

    Lifecycle (single upload)::

        pending → uploaded → finalized  (or → aborted)

    Lifecycle (chunked upload)::

        pending → uploading → … → uploading → finalized  (or → aborted)

    ``upload_method`` indicates how the client should deliver file bytes:

    - ``"proxy"``: PUT bytes to ``/blobs/upload-sessions/{upload_id}/content``.
      May be called multiple times for chunked uploads.
    - ``"single_put"``: upload directly to ``upload_url`` (e.g. S3 presigned PUT)
    """

    upload_id: str
    """Unique identifier for this upload session."""

    file_id: str
    """Pre-allocated file ID (may be a placeholder until finalize)."""

    status: Literal["pending", "uploading", "uploaded", "finalized", "aborted"] = (
        "pending"
    )
    """Current lifecycle state of the session."""

    upload_method: Literal["proxy", "single_put"] = "proxy"
    """How the client should deliver file bytes."""

    upload_url: str = ""
    """URL for the client to upload bytes (only relevant for ``single_put``)."""

    content_type: str | UnsetType = UNSET
    """MIME type of the content being uploaded."""

    size: int | None = None
    """Expected size of the content in bytes (``None`` if unknown)."""

    uploaded_size: int = 0
    """Number of bytes already uploaded (useful for progress tracking)."""

    total_parts: int | None = None
    """Expected number of parts for parallel chunked upload (``None`` if unknown)."""

    parts_received: list[int] = []
    """Sorted list of 1-based part numbers that have been received so far."""

    expires_at: dt.datetime | None = None
    """When this upload session expires (``None`` for no expiry)."""

Attributes

upload_id instance-attribute

upload_id: str

Unique identifier for this upload session.

file_id instance-attribute

file_id: str

Pre-allocated file ID (may be a placeholder until finalize).

status class-attribute instance-attribute

status: Literal[
    "pending",
    "uploading",
    "uploaded",
    "finalized",
    "aborted",
] = "pending"

Current lifecycle state of the session.

upload_method class-attribute instance-attribute

upload_method: Literal['proxy', 'single_put'] = 'proxy'

How the client should deliver file bytes.

upload_url class-attribute instance-attribute

upload_url: str = ''

URL for the client to upload bytes (only relevant for single_put).

content_type class-attribute instance-attribute

content_type: str | UnsetType = UNSET

MIME type of the content being uploaded.

size class-attribute instance-attribute

size: int | None = None

Expected size of the content in bytes (None if unknown).

uploaded_size class-attribute instance-attribute

uploaded_size: int = 0

Number of bytes already uploaded (useful for progress tracking).

total_parts class-attribute instance-attribute

total_parts: int | None = None

Expected number of parts for parallel chunked upload (None if unknown).

parts_received class-attribute instance-attribute

parts_received: list[int] = []

Sorted list of 1-based part numbers that have been received so far.

expires_at class-attribute instance-attribute

expires_at: datetime | None = None

When this upload session expires (None for no expiry).

DisplayName

Annotation marker designating a str field as the display name.

Usage::

class Character(Struct):
    name: Annotated[str, DisplayName()]  # ← this field is the display name
    level: int = 1

The AutoCRUD framework will inject x-display-name-field into the OpenAPI schema so the web frontend can show a friendly name instead of just the resource ID.

Source code in autocrud/types.py

class DisplayName:
    """Annotation marker designating a ``str`` field as the display name.

    Usage::

        class Character(Struct):
            name: Annotated[str, DisplayName()]  # ← this field is the display name
            level: int = 1

    The AutoCRUD framework will inject ``x-display-name-field`` into the
    OpenAPI schema so the web frontend can show a friendly name instead of
    just the resource ID.
    """

    __slots__ = ("label",)

    def __init__(self, label: str | None = None) -> None:
        self.label = label

    def __repr__(self) -> str:
        if self.label is None:
            return "DisplayName()"
        return f"DisplayName({self.label!r})"

Attributes

label instance-attribute

label = label

DuplicateResourceError

Bases: ResourceConflictError

Raised when loading a resource with an ID that already exists and on_duplicate is set to :attr:OnDuplicate.raise_error.

Source code in autocrud/types.py

class DuplicateResourceError(ResourceConflictError):
    """Raised when loading a resource with an ID that already exists
    and *on_duplicate* is set to :attr:`OnDuplicate.raise_error`."""

    def __init__(self, resource_id: str) -> None:
        super().__init__(
            f"Duplicate resource '{resource_id}' already exists during load."
        )
        self.resource_id = resource_id

Attributes

resource_id instance-attribute

resource_id = resource_id

IConstraintChecker

Bases: ABC

Interface for custom constraint checkers.

Implement this to define reusable data constraints that are automatically enforced during create, update, modify, switch and restore operations. The framework handles all event lifecycle (before / on_success) and compensation (rollback) logic — you only need to implement the check.

Example::

class NoDuplicateEmailChecker(IConstraintChecker):
    def __init__(self, rm: ResourceManager) -> None:
        self.rm = rm

    def check(
        self, data: Any, *, exclude_resource_id: str | None = None
    ) -> None:
        email = getattr(data, "email", None)
        if email and self._email_exists(email, exclude_resource_id):
            raise ValueError(f"Email {email!r} already in use")


# Pass a factory callable (receives ResourceManager):
crud.add_model(User, constraint_checkers=[NoDuplicateEmailChecker])
# Or a lambda factory:
crud.add_model(
    User, constraint_checkers=[lambda rm: NoDuplicateEmailChecker(rm)]
)

Source code in autocrud/types.py

class IConstraintChecker(ABC):
    """Interface for custom constraint checkers.

    Implement this to define reusable data constraints that are automatically
    enforced during create, update, modify, switch and restore operations.
    The framework handles all event lifecycle (before / on_success) and
    compensation (rollback) logic — you only need to implement the check.

    Example::

        class NoDuplicateEmailChecker(IConstraintChecker):
            def __init__(self, rm: ResourceManager) -> None:
                self.rm = rm

            def check(
                self, data: Any, *, exclude_resource_id: str | None = None
            ) -> None:
                email = getattr(data, "email", None)
                if email and self._email_exists(email, exclude_resource_id):
                    raise ValueError(f"Email {email!r} already in use")


        # Pass a factory callable (receives ResourceManager):
        crud.add_model(User, constraint_checkers=[NoDuplicateEmailChecker])
        # Or a lambda factory:
        crud.add_model(
            User, constraint_checkers=[lambda rm: NoDuplicateEmailChecker(rm)]
        )
    """

    @abstractmethod
    def check(self, data: Any, *, exclude_resource_id: str | None = None) -> None:
        """Validate that *data* satisfies this constraint.

        Args:
            data: The resource data (msgspec Struct instance).
            exclude_resource_id: When updating an existing resource, pass its
                ID so the checker can allow the resource to keep its own values.

        Raises:
            Any exception to signal a constraint violation.  The framework
            will catch it, execute compensation, and re-raise.
        """
        ...

    def data_relevant_changed(self, current_data: Any, new_data: Any) -> bool:
        """Return whether the fields relevant to this constraint changed.

        Called during *modify* to skip unnecessary checks when the
        constrained fields are unchanged.  The default implementation
        returns ``True`` (always re-check).  Override for optimisation.
        """
        return True

Functions

check abstractmethod

check(
    data: Any, *, exclude_resource_id: str | None = None
) -> None

Validate that data satisfies this constraint.

PARAMETER	DESCRIPTION
data	The resource data (msgspec Struct instance). TYPE: Any
exclude_resource_id	When updating an existing resource, pass its ID so the checker can allow the resource to keep its own values. TYPE: str | None DEFAULT: None

Source code in autocrud/types.py

@abstractmethod
def check(self, data: Any, *, exclude_resource_id: str | None = None) -> None:
    """Validate that *data* satisfies this constraint.

    Args:
        data: The resource data (msgspec Struct instance).
        exclude_resource_id: When updating an existing resource, pass its
            ID so the checker can allow the resource to keep its own values.

    Raises:
        Any exception to signal a constraint violation.  The framework
        will catch it, execute compensation, and re-raise.
    """
    ...

data_relevant_changed

data_relevant_changed(
    current_data: Any, new_data: Any
) -> bool

Return whether the fields relevant to this constraint changed.

Called during modify to skip unnecessary checks when the constrained fields are unchanged. The default implementation returns True (always re-check). Override for optimisation.

Source code in autocrud/types.py

def data_relevant_changed(self, current_data: Any, new_data: Any) -> bool:
    """Return whether the fields relevant to this constraint changed.

    Called during *modify* to skip unnecessary checks when the
    constrained fields are unchanged.  The default implementation
    returns ``True`` (always re-check).  Override for optimisation.
    """
    return True

IValidator

Bases: ABC

Interface for custom data validators.

Implement this to create reusable validators that can be attached via add_model(validator=...) or Schema(..., validator=...).

Example::

class PriceValidator(IValidator):
    def validate(self, data) -> None:
        if data.price < 0:
            raise ValueError("Price must be non-negative")


crud.add_model(Item, validator=PriceValidator())

Source code in autocrud/types.py

class IValidator(ABC):
    """Interface for custom data validators.

    Implement this to create reusable validators that can be attached via
    `add_model(validator=...)` or `Schema(..., validator=...)`.

    Example::

        class PriceValidator(IValidator):
            def validate(self, data) -> None:
                if data.price < 0:
                    raise ValueError("Price must be non-negative")


        crud.add_model(Item, validator=PriceValidator())
    """

    @abstractmethod
    def validate(self, data: Any) -> None:
        """Validate the data.

        Raises:
            ValidationError:
                If validation fails. Raising `ValueError` is allowed and will be
                wrapped as `ValidationError` by AutoCRUD.
        """

Functions

validate abstractmethod

validate(data: Any) -> None

Validate the data.

RAISES	DESCRIPTION
ValidationError	If validation fails. Raising ValueError is allowed and will be wrapped as ValidationError by AutoCRUD.

Source code in autocrud/types.py

@abstractmethod
def validate(self, data: Any) -> None:
    """Validate the data.

    Raises:
        ValidationError:
            If validation fails. Raising `ValueError` is allowed and will be
            wrapped as `ValidationError` by AutoCRUD.
    """

JobRedirectInfo

Bases: Struct

Response body returned by async create actions (HTTP 202).

When a custom create action uses async_mode='job', the endpoint returns this struct instead of RevisionInfo so the client can navigate to the auto-generated Job resource to track progress.

ATTRIBUTE	DESCRIPTION
job_resource_name	The registered name of the auto-generated Job resource. TYPE: str
job_resource_id	The resource ID of the newly created Job instance. TYPE: str
redirect_url	A URL path to the Job detail endpoint. TYPE: str

Source code in autocrud/types.py

class JobRedirectInfo(Struct, kw_only=True):
    """Response body returned by async create actions (HTTP 202).

    When a custom create action uses ``async_mode='job'``, the endpoint
    returns this struct instead of ``RevisionInfo`` so the client can
    navigate to the auto-generated Job resource to track progress.

    Attributes:
        job_resource_name: The registered name of the auto-generated Job resource.
        job_resource_id: The resource ID of the newly created Job instance.
        redirect_url: A URL path to the Job detail endpoint.
    """

    job_resource_name: str
    """The registered name of the auto-generated Job resource."""

    job_resource_id: str
    """The resource ID of the newly created Job instance."""

    redirect_url: str
    """A URL path to the Job detail endpoint."""

Attributes

job_resource_name instance-attribute

job_resource_name: str

The registered name of the auto-generated Job resource.

job_resource_id instance-attribute

job_resource_id: str

The resource ID of the newly created Job instance.

redirect_url instance-attribute

redirect_url: str

A URL path to the Job detail endpoint.

MissingOperationContextError

Bases: Exception

Raised when a write operation is missing required context fields.

This error occurs when a mutating method (create, update, delete, etc.) is called without the required user and/or now context. Context can be supplied via:

1. Explicit keyword arguments: mgr.create(data, user=..., now=...)
2. A using() scope: with mgr.using(user=..., now=...): ...
3. Manager defaults: ResourceManager(..., default_user=..., default_now=...)

ATTRIBUTE	DESCRIPTION
missing_fields	Names of the context fields that were not resolved.
method_name	The name of the method that required the context.

Source code in autocrud/types.py

class MissingOperationContextError(Exception):
    """Raised when a write operation is missing required context fields.

    This error occurs when a mutating method (create, update, delete, etc.)
    is called without the required ``user`` and/or ``now`` context.  Context
    can be supplied via:

    1. Explicit keyword arguments: ``mgr.create(data, user=..., now=...)``
    2. A ``using()`` scope: ``with mgr.using(user=..., now=...): ...``
    3. Manager defaults: ``ResourceManager(..., default_user=..., default_now=...)``

    Attributes:
        missing_fields: Names of the context fields that were not resolved.
        method_name: The name of the method that required the context.
    """

    def __init__(
        self,
        missing_fields: list[str],
        method_name: str | None = None,
    ) -> None:
        self.missing_fields = missing_fields
        self.method_name = method_name
        fields_str = ", ".join(missing_fields)
        if method_name:
            msg = (
                f"Missing required operation context for '{method_name}': "
                f"{fields_str}. "
                f"Provide via explicit kwargs, using() scope, or manager defaults."
            )
        else:
            msg = (
                f"Missing required operation context: {fields_str}. "
                f"Provide via explicit kwargs, using() scope, or manager defaults."
            )
        super().__init__(msg)

Attributes

missing_fields instance-attribute

missing_fields = missing_fields

method_name instance-attribute

method_name = method_name

OnDelete

Bases: StrEnum

Defines the referential action when the referenced resource is deleted.

Source code in autocrud/types.py

class OnDelete(StrEnum):
    """Defines the referential action when the referenced resource is deleted."""

    dangling = "dangling"
    """No action taken. The reference becomes dangling. (default)"""

    set_null = "set_null"
    """Set the referencing field to null. Requires the field to be Optional."""

    cascade = "cascade"
    """Delete the referencing resource as well."""

Attributes

dangling class-attribute instance-attribute

dangling = 'dangling'

No action taken. The reference becomes dangling. (default)

set_null class-attribute instance-attribute

set_null = 'set_null'

Set the referencing field to null. Requires the field to be Optional.

cascade class-attribute instance-attribute

cascade = 'cascade'

Delete the referencing resource as well.

OnDuplicate

Bases: StrEnum

Strategy for handling duplicate resource IDs during incremental load.

Source code in autocrud/types.py

class OnDuplicate(StrEnum):
    """Strategy for handling duplicate resource IDs during incremental load."""

    overwrite = "overwrite"
    """Overwrite existing resources with loaded data."""

    skip = "skip"
    """Skip resources that already exist."""

    raise_error = "raise_error"
    """Raise DuplicateResourceError when a duplicate is found."""

Attributes

overwrite class-attribute instance-attribute

overwrite = 'overwrite'

Overwrite existing resources with loaded data.

skip class-attribute instance-attribute

skip = 'skip'

Skip resources that already exist.

raise_error class-attribute instance-attribute

raise_error = 'raise_error'

Raise DuplicateResourceError when a duplicate is found.

Ref

Metadata marker for a field that references another AutoCRUD resource.

Use with Annotated to annotate a str field that holds a reference to another AutoCRUD resource.

By default ref_type is RefType.resource_id, meaning the field stores a resource_id and participates in referential integrity.

Set ref_type=RefType.revision_id for version-aware references where the field may store either a revision_id (pinned) or a resource_id (meaning latest). Revision refs are always on_delete=dangling.

Example::

class Monster(Struct):
    zone_id: Annotated[str, Ref("zone")]
    guild_id: Annotated[
        str | None, Ref("guild", on_delete=OnDelete.set_null)
    ] = None
    owner_id: Annotated[str, Ref("character", on_delete=OnDelete.cascade)]
    zone_snapshot_id: Annotated[str, Ref("zone", ref_type=RefType.revision_id)]

Source code in autocrud/types.py

class Ref:
    """Metadata marker for a field that references another AutoCRUD resource.

    Use with ``Annotated`` to annotate a ``str`` field that holds a reference
    to another AutoCRUD resource.

    By default ``ref_type`` is ``RefType.resource_id``, meaning the field
    stores a ``resource_id`` and participates in referential integrity.

    Set ``ref_type=RefType.revision_id`` for version-aware references where
    the field may store either a ``revision_id`` (pinned) or a ``resource_id``
    (meaning *latest*).  Revision refs are always ``on_delete=dangling``.

    Example::

        class Monster(Struct):
            zone_id: Annotated[str, Ref("zone")]
            guild_id: Annotated[
                str | None, Ref("guild", on_delete=OnDelete.set_null)
            ] = None
            owner_id: Annotated[str, Ref("character", on_delete=OnDelete.cascade)]
            zone_snapshot_id: Annotated[str, Ref("zone", ref_type=RefType.revision_id)]
    """

    __slots__ = ("resource", "on_delete", "ref_type")

    def __init__(
        self,
        resource: str,
        *,
        on_delete: OnDelete = OnDelete.dangling,
        ref_type: RefType = RefType.resource_id,
    ) -> None:
        self.resource = resource
        self.on_delete = OnDelete(on_delete)
        self.ref_type = RefType(ref_type)
        if self.ref_type != RefType.resource_id and self.on_delete != OnDelete.dangling:
            raise ValueError(
                f"Ref({resource!r}) with ref_type={self.ref_type!r} "
                f"requires on_delete=OnDelete.dangling, "
                f"got on_delete={self.on_delete!r}."
            )

    def __repr__(self) -> str:
        parts = [repr(self.resource), f"on_delete={self.on_delete!r}"]
        if self.ref_type != RefType.resource_id:
            parts.append(f"ref_type={self.ref_type!r}")
        return f"Ref({', '.join(parts)})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Ref):
            return NotImplemented
        return (
            self.resource == other.resource
            and self.on_delete == other.on_delete
            and self.ref_type == other.ref_type
        )

    def __hash__(self) -> int:
        return hash((self.resource, self.on_delete, self.ref_type))

Attributes

resource instance-attribute

resource = resource

on_delete instance-attribute

on_delete = OnDelete(on_delete)

ref_type instance-attribute

ref_type = RefType(ref_type)

RefRevision

Metadata marker for a field that references another resource's revision_id.

.. deprecated:: 0.9.0 Use Ref(resource, ref_type=RefType.revision_id) instead.

Example::

class Monster(Struct):
    zone_revision_id: Annotated[str, RefRevision("zone")]

Source code in autocrud/types.py

class RefRevision:
    """Metadata marker for a field that references another resource's revision_id.

    .. deprecated:: 0.9.0
        Use ``Ref(resource, ref_type=RefType.revision_id)`` instead.

    Example::

        class Monster(Struct):
            zone_revision_id: Annotated[str, RefRevision("zone")]
    """

    __slots__ = ("resource",)

    def __init__(self, resource: str) -> None:
        import warnings

        warnings.warn(
            "RefRevision is deprecated. "
            "Use Ref(resource, ref_type=RefType.revision_id) instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        self.resource = resource

    def __repr__(self) -> str:
        return f"RefRevision({self.resource!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, RefRevision):
            return NotImplemented
        return self.resource == other.resource

    def __hash__(self) -> int:
        return hash(self.resource)

Attributes

resource instance-attribute

resource = resource

RefType

Bases: StrEnum

Defines the type of reference a field holds.

Source code in autocrud/types.py

class RefType(StrEnum):
    """Defines the type of reference a field holds."""

    resource_id = "resource_id"
    """The field stores a resource_id. The reference targets the resource as
    a whole and participates in referential integrity (on_delete), auto-indexing,
    and referrers queries."""

    revision_id = "revision_id"
    """The field stores a version-aware reference: either a revision_id
    (pinned to a specific revision) or a resource_id (meaning *latest*).
    Revision refs are always ``on_delete=dangling``, are not auto-indexed,
    and are excluded from referrers queries."""

Attributes

resource_id class-attribute instance-attribute

resource_id = 'resource_id'

The field stores a resource_id. The reference targets the resource as a whole and participates in referential integrity (on_delete), auto-indexing, and referrers queries.

revision_id class-attribute instance-attribute

revision_id = 'revision_id'

The field stores a version-aware reference: either a revision_id (pinned to a specific revision) or a resource_id (meaning latest). Revision refs are always on_delete=dangling, are not auto-indexed, and are excluded from referrers queries.

RevisionNotMigratedError

Bases: SchemaConflictError

Raised when switching to a revision whose schema version differs from the resource's current schema version.

The revision must be migrated first via resource_manager.migrate(resource_id, revision_id=...).

ATTRIBUTE	DESCRIPTION
resource_id	The resource that was being switched.
revision_id	The target revision that is not yet migrated.
revision_schema_version	The schema version stored on the target revision.
current_schema_version	The resource-level (latest) schema version.

Source code in autocrud/types.py

class RevisionNotMigratedError(SchemaConflictError):
    """Raised when switching to a revision whose schema version differs from
    the resource's current schema version.

    The revision must be migrated first via
    ``resource_manager.migrate(resource_id, revision_id=...)``.

    Attributes:
        resource_id: The resource that was being switched.
        revision_id: The target revision that is not yet migrated.
        revision_schema_version: The schema version stored on the target revision.
        current_schema_version: The resource-level (latest) schema version.
    """

    def __init__(
        self,
        resource_id: str,
        revision_id: str,
        revision_schema_version: str | None,
        current_schema_version: str | None,
    ) -> None:
        super().__init__(
            f"Revision '{revision_id}' of resource '{resource_id}' is at "
            f"schema version '{revision_schema_version}' but the resource is at "
            f"'{current_schema_version}'. Migrate the revision first with "
            f"migrate('{resource_id}', revision_id='{revision_id}')."
        )
        self.resource_id = resource_id
        self.revision_id = revision_id
        self.revision_schema_version = revision_schema_version
        self.current_schema_version = current_schema_version

Attributes

resource_id instance-attribute

resource_id = resource_id

revision_id instance-attribute

revision_id = revision_id

revision_schema_version instance-attribute

revision_schema_version = revision_schema_version

current_schema_version instance-attribute

current_schema_version = current_schema_version

SearchedResource

Bases: Struct, Generic[T]

A resource item returned by list_resources.

Each field may be the full type, a partial Struct (when partial fields are requested), or UNSET (when excluded via the returns parameter).

Source code in autocrud/types.py

class SearchedResource(Struct, Generic[T]):
    """A resource item returned by list_resources.

    Each field may be the full type, a partial Struct (when partial fields
    are requested), or UNSET (when excluded via the *returns* parameter).
    """

    data: T | Struct | UnsetType = UNSET
    info: RevisionInfo | Struct | UnsetType = UNSET
    meta: ResourceMeta | Struct | UnsetType = UNSET

Attributes

data class-attribute instance-attribute

data: T | Struct | UnsetType = UNSET

info class-attribute instance-attribute

info: RevisionInfo | Struct | UnsetType = UNSET

meta class-attribute instance-attribute

meta: ResourceMeta | Struct | UnsetType = UNSET

Unique

Annotation marker that enforces uniqueness of a field.

Use with Annotated to mark a field as unique among non-deleted resources of the same type.

Semantics: - Soft-deleted resources are ignored. - None values are ignored (None may repeat).

AutoCRUD ensures the field is indexed and checks uniqueness on write operations (create/update/modify/patch) when the unique-relevant value changes.

Usage::

class User(Struct):
    username: Annotated[str, Unique()]
    email: Annotated[str, Unique()]
    nickname: Annotated[str | None, Unique()] = None  # None can repeat

RAISES	DESCRIPTION
	exc:UniqueConstraintError: When a duplicate non-None value is detected.

Source code in autocrud/types.py

class Unique:
    """Annotation marker that enforces uniqueness of a field.

    Use with ``Annotated`` to mark a field as unique among **non-deleted**
    resources of the same type.

    Semantics:
    - Soft-deleted resources are ignored.
    - ``None`` values are ignored (``None`` may repeat).

    AutoCRUD ensures the field is indexed and checks uniqueness on write
    operations (create/update/modify/patch) when the unique-relevant value
    changes.

    Usage::

        class User(Struct):
            username: Annotated[str, Unique()]
            email: Annotated[str, Unique()]
            nickname: Annotated[str | None, Unique()] = None  # None can repeat

    Raises:
        :exc:`UniqueConstraintError`: When a duplicate non-None value is detected.
    """

    __slots__ = ()

    def __repr__(self) -> str:
        return "Unique()"

UniqueConstraintError

Bases: ResourceConflictError

Raised when a field annotated with :class:Unique already has the given value on another (non-deleted) resource.

ATTRIBUTE	DESCRIPTION
field	The name of the unique-constrained field.
value	The duplicate value that caused the conflict.
conflicting_resource_id	The resource_id that already holds the value.

Source code in autocrud/types.py

class UniqueConstraintError(ResourceConflictError):
    """Raised when a field annotated with :class:`Unique` already has the given value
    on another (non-deleted) resource.

    Attributes:
        field: The name of the unique-constrained field.
        value: The duplicate value that caused the conflict.
        conflicting_resource_id: The ``resource_id`` that already holds the value.
    """

    def __init__(self, field: str, value: Any, conflicting_resource_id: str) -> None:
        super().__init__(
            f"Unique constraint violated: field '{field}' value {value!r} "
            f"already exists on resource '{conflicting_resource_id}'."
        )
        self.field = field
        self.value = value
        self.conflicting_resource_id = conflicting_resource_id

Attributes

field instance-attribute

field = field

value instance-attribute

value = value

conflicting_resource_id instance-attribute

conflicting_resource_id = conflicting_resource_id

ValidationError

Bases: ValueError

Raised when data fails custom validation.

Inherits from ValueError so it can be caught broadly. This is distinct from msgspec.ValidationError which handles type-level validation.

Source code in autocrud/types.py

class ValidationError(ValueError):
    """Raised when data fails custom validation.

    Inherits from ValueError so it can be caught broadly.
    This is distinct from msgspec.ValidationError which handles
    type-level validation.
    """

    pass

Functions

struct_to_pydantic

struct_to_pydantic(struct_cls: type) -> type

Convert a msgspec Struct class to a Pydantic BaseModel class.

This is the reverse of pydantic_to_struct. It allows using a Struct-based type as a FastAPI request body parameter by generating an equivalent Pydantic model that FastAPI can introspect for OpenAPI schema generation and validation.

Usage::

@app.post("/action")
async def my_action(body: struct_to_pydantic(MyStruct) = Body(...)): ...

Handles: - Simple scalar types (str, int, float, bool, datetime …) - Optional[X] - Enum types - list[X], dict[K, V] - Nested Structs (recursively converted) - Tagged unions (A | B where both have tag) → Pydantic discriminated unions with Literal discriminator field - Annotated metadata is stripped (AutoCRUD-specific markers like Ref, DisplayName, Unique are not meaningful for Pydantic).

Source code in autocrud/resource_manager/pydantic_converter.py

def struct_to_pydantic(struct_cls: type) -> type:
    """Convert a msgspec Struct class to a Pydantic BaseModel class.

    This is the reverse of ``pydantic_to_struct``.  It allows using a
    Struct-based type as a FastAPI request body parameter by generating an
    equivalent Pydantic model that FastAPI can introspect for OpenAPI schema
    generation and validation.

    Usage::

        @app.post("/action")
        async def my_action(body: struct_to_pydantic(MyStruct) = Body(...)): ...

    Handles:
    - Simple scalar types (str, int, float, bool, datetime …)
    - ``Optional[X]``
    - ``Enum`` types
    - ``list[X]``, ``dict[K, V]``
    - Nested Structs (recursively converted)
    - Tagged unions (``A | B`` where both have ``tag``) → Pydantic
      discriminated unions with ``Literal`` discriminator field
    - ``Annotated`` metadata is **stripped** (AutoCRUD-specific markers
      like ``Ref``, ``DisplayName``, ``Unique`` are not meaningful for
      Pydantic).
    """

    if not (isinstance(struct_cls, type) and issubclass(struct_cls, msgspec.Struct)):
        raise TypeError(f"Expected a msgspec Struct class, got {struct_cls}")

    cache: dict[type, type] = {}
    return _struct_to_pydantic_recursive(struct_cls, cache)