Module `adcp.types.versioned`

Version-scoped Pydantic models backed by bundled AdCP JSON Schemas.

The primary :mod:adcp.types surface represents the SDK's current generated release. Use this module (or adcp.types.v30 / v31 / v32) when an application must construct and validate the exact public shape negotiated with an older peer in the same SDK process. Generated .pyi files expose exact field, nested-value, and direct constructor requiredness information to type checkers; conditional JSON Schema constraints remain runtime-only. At runtime these remain exact-schema RootModel[dict[str, Any]] boundary validators. Use :func:make_versioned_base() when one normal Pydantic model must combine a pinned protocol shape with adopter-defined, excluded internal fields.

Functions

def make_versioned_base(version: str, model_name: str) ‑> type[pydantic.main.BaseModel]

Expand source code

@cache
def make_versioned_base(version: str, model_name: str) -> type[BaseModel]:
    """Create a subclassable Pydantic base for one bundled protocol model.

    Example::

        ListCreatives31 = make_versioned_base("3.1", "ListCreativesRequest")

        class SellerListCreativesRequest(ListCreatives31):
            internal_tenant_id: str = Field(exclude=True)

    The returned class has real top-level Pydantic fields, reuses nested
    runtime annotations from the SDK's current public model where available,
    and validates its serialized protocol payload against the requested
    bundled schema. Adopter subclasses may add fields declared with
    ``Field(exclude=True)``; those fields never enter schema validation or the
    wire payload. Unknown undeclared fields are rejected even when a protocol
    schema permits extension keys, keeping version-only fields explicit.
    """
    tool_name, direction = _schema_key_for_model_name(model_name)
    schema = get_portable_schema(tool_name, direction, version=version)
    if schema is None:
        raise LookupError(f"no {version} schema for {tool_name}::{direction}")

    shapes = _object_shapes(schema, schema)
    properties = {
        name: field_schema
        for shape_properties, _required in shapes
        for name, field_schema in shape_properties.items()
    }
    guaranteed_fields = set.intersection(*(required for _properties, required in shapes))
    current_annotations = _current_model_annotations(model_name)
    fields: dict[str, Any] = {}
    omit_none_fields = frozenset(
        name
        for name, field_schema in properties.items()
        if name not in guaranteed_fields and not _schema_admits_null(field_schema, schema)
    )
    for name, field_schema in properties.items():
        annotation = current_annotations.get(
            name,
            _fallback_annotation(field_schema, schema),
        )
        if name not in guaranteed_fields:
            annotation = annotation | None
        description = field_schema.get("description") if isinstance(field_schema, dict) else None
        if isinstance(field_schema, dict) and "default" in field_schema:
            default = Field(
                default=copy.deepcopy(field_schema["default"]),
                description=description,
            )
        elif name in guaranteed_fields:
            default = Field(description=description)
        else:
            default = Field(default=None, description=description)
        fields[name] = (annotation, default)

    version_token = re.sub(r"[^A-Za-z0-9]+", "_", version).strip("_")
    generated_name = f"{model_name}V{version_token}Base"
    model: type[_VersionedExtensionModel] = create_model(
        generated_name,
        __base__=_VersionedExtensionModel,
        __module__=__name__,
        **fields,
    )
    model.schema_version = version
    model.schema_tool_name = tool_name
    model.schema_direction = direction
    model.schema_document = schema
    model._omit_none_fields = omit_none_fields
    return model

Create a subclassable Pydantic base for one bundled protocol model.

Example::

ListCreatives31 = make_versioned_base("3.1", "ListCreativesRequest")

class SellerListCreativesRequest(ListCreatives31):
    internal_tenant_id: str = Field(exclude=True)

The returned class has real top-level Pydantic fields, reuses nested runtime annotations from the SDK's current public model where available, and validates its serialized protocol payload against the requested bundled schema. Adopter subclasses may add fields declared with Field(exclude=True); those fields never enter schema validation or the wire payload. Unknown undeclared fields are rejected even when a protocol schema permits extension keys, keeping version-only fields explicit.

def model_for_version(version: str, model_name: str) ‑> type[VersionedSchemaModel]

Expand source code

def model_for_version(version: str, model_name: str) -> type[VersionedSchemaModel]:
    """Resolve ``ListCreativesRequest``-style names for a protocol release."""
    tool_name, direction = _schema_key_for_model_name(model_name)
    return schema_model_for_version(
        version,
        tool_name,
        direction,
        model_name=model_name,
    )

Resolve ListCreativesRequest-style names for a protocol release.

def schema_model_for_version(version: str, tool_name: str, direction: VersionedDirection = 'request', *, model_name: str | None = None) ‑> type[VersionedSchemaModel]

Expand source code

@cache
def schema_model_for_version(
    version: str,
    tool_name: str,
    direction: VersionedDirection = "request",
    *,
    model_name: str | None = None,
) -> type[VersionedSchemaModel]:
    """Return a cached Pydantic model for one version/tool/direction."""
    schema = get_portable_schema(tool_name, direction, version=version)
    if schema is None:
        raise LookupError(f"no {version} schema for {tool_name}::{direction}")
    suffix = {
        "request": "Request",
        "sync": "Response",
        "submitted": "SubmittedResponse",
        "working": "WorkingResponse",
        "input-required": "InputRequiredResponse",
    }[direction]
    name = model_name or f"{_pascal_case(tool_name)}{suffix}"
    shapes = _object_shapes(schema, schema)
    known_fields = {field for properties, _required in shapes for field in properties}
    guaranteed_fields = set.intersection(*(required for _properties, required in shapes))
    properties = schema.get("properties", {})
    if isinstance(properties, dict):
        guaranteed_fields.update(
            field
            for field, field_schema in properties.items()
            if isinstance(field_schema, dict) and "default" in field_schema
        )
    return type(
        name,
        (VersionedSchemaModel,),
        {
            "__module__": __name__,
            "schema_version": version,
            "schema_tool_name": tool_name,
            "schema_direction": direction,
            "schema_document": schema,
            "optional_fields": frozenset(known_fields - guaranteed_fields),
        },
    )

Return a cached Pydantic model for one version/tool/direction.

def versioned_surface(version: str, module_name: str) ‑> tuple[typing.Any, typing.Any, list[str]]

Expand source code

def versioned_surface(
    version: str,
    module_name: str,
) -> tuple[Any, Any, list[str]]:
    """Build PEP 562 hooks for a version shorthand module."""
    names: list[str] = []
    for key in list_validator_keys(version=version):
        tool_name, direction = key.split("::", 1)
        if direction == "request":
            names.append(f"{_pascal_case(tool_name)}Request")
        elif direction == "sync":
            names.append(f"{_pascal_case(tool_name)}Response")
        elif direction == "submitted":
            names.append(f"{_pascal_case(tool_name)}SubmittedResponse")
        elif direction == "working":
            names.append(f"{_pascal_case(tool_name)}WorkingResponse")
        elif direction == "input-required":
            names.append(f"{_pascal_case(tool_name)}InputRequiredResponse")
    exported = sorted(set(names))

    def resolve(name: str) -> Any:
        if name not in exported:
            raise AttributeError(f"module {module_name!r} has no attribute {name!r}")
        model = model_for_version(version, name)
        model.__module__ = module_name
        return model

    def directory() -> list[str]:
        return list(exported)

    return resolve, directory, exported

Build PEP 562 hooks for a version shorthand module.

Classes

class VersionedSchemaModel (root: dict[str, Any] | None = None, **data: Any)

Expand source code

class VersionedSchemaModel(RootModel[dict[str, Any]]):
    """Dict-shaped Pydantic model that enforces one bundled schema version.

    Keyword construction and attribute access intentionally mirror ordinary
    generated request models while retaining the exact JSON Schema as the
    validation authority. The companion module stubs provide static field
    information. This runtime wrapper remains intended for boundary validation
    and schema generation, not as a base class for adopter-defined models.
    """

    schema_version: ClassVar[str]
    schema_tool_name: ClassVar[str]
    schema_direction: ClassVar[VersionedDirection]
    schema_document: ClassVar[dict[str, Any]]
    optional_fields: ClassVar[frozenset[str]]

    def __init__(self, root: dict[str, Any] | None = None, **data: Any) -> None:
        if root is not None and data:
            raise TypeError("pass either root or keyword fields, not both")
        super().__init__(root=root if root is not None else data)

    @model_validator(mode="before")
    @classmethod
    def _apply_top_level_defaults(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        result = dict(value)
        properties = cls.schema_document.get("properties", {})
        if isinstance(properties, dict):
            for name, field_schema in properties.items():
                if (
                    name not in result
                    and isinstance(field_schema, dict)
                    and "default" in field_schema
                ):
                    result[name] = copy.deepcopy(field_schema["default"])
        return result

    @model_validator(mode="after")
    def _validate_bundled_schema(self) -> VersionedSchemaModel:
        validator = get_validator(
            self.schema_tool_name,
            self.schema_direction,
            version=self.schema_version,
        )
        if validator is None:
            raise ValueError(
                f"no {self.schema_version} schema for "
                f"{self.schema_tool_name}::{self.schema_direction}"
            )
        issues = sorted(validator.iter_errors(self.root), key=lambda error: list(error.path))
        if issues:
            issue = issues[0]
            path = ".".join(str(part) for part in issue.absolute_path) or "<root>"
            raise ValueError(f"{path}: {issue.message}")
        return self

    def __getattr__(self, name: str) -> Any:
        root = object.__getattribute__(self, "root")
        if name in root:
            return root[name]
        if name in type(self).optional_fields:
            return None
        raise AttributeError(f"{type(self).__name__!s} has no attribute {name!r}")

    def __getitem__(self, name: str) -> Any:
        return self.root[name]

    @classmethod
    def model_json_schema(cls, *args: Any, **kwargs: Any) -> dict[str, Any]:
        del args, kwargs
        return copy.deepcopy(cls.schema_document)

    @classmethod
    def __get_pydantic_json_schema__(
        cls,
        core_schema: CoreSchema,
        handler: GetJsonSchemaHandler,
    ) -> JsonSchemaValue:
        """Expose the negotiated contract to TypeAdapter/FastAPI consumers."""
        del core_schema, handler
        return _inline_local_refs(cls.schema_document)

Dict-shaped Pydantic model that enforces one bundled schema version.

Keyword construction and attribute access intentionally mirror ordinary generated request models while retaining the exact JSON Schema as the validation authority. The companion module stubs provide static field information. This runtime wrapper remains intended for boundary validation and schema generation, not as a base class for adopter-defined models.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

pydantic.root_model.RootModel[dict[str, Any]]
pydantic.root_model.RootModel
pydantic.main.BaseModel
typing.Generic

Subclasses

Class variables

var model_config
var optional_fields : ClassVar[frozenset[str]]
var schema_direction : ClassVar[Literal['request', 'sync', 'submitted', 'working', 'input-required']]
var schema_document : ClassVar[dict[str, Any]]
var schema_tool_name : ClassVar[str]
var schema_version : ClassVar[str]

Static methods

def model_json_schema(*args: Any, **kwargs: Any) ‑> dict[str, typing.Any]

Generates a JSON schema for a model class.

Args

-----=

by_alias

Whether to use attribute aliases or not.

ref_template

The reference template.

union_format

The format to use when combining schemas from unions together. Can be one of:

'any_of': Use the anyOf keyword to combine schemas (the default).
'primitive_type_array': Use the type keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive type (string, boolean, null, integer or number) or contains constraints/metadata, falls back to any_of.

schema_generator

To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

mode

The mode in which to generate the schema.

Returns -----= The JSON schema for the given model class.