Skip to content

get_model_schema

faststream.asyncapi.message.get_model_schema #

get_model_schema(
    call: Optional[Type[BaseModel]],
    prefix: str = "",
    exclude: Sequence[str] = (),
) -> Optional[Dict[str, Any]]

Get the schema of a model.

PARAMETER DESCRIPTION
call

The model class to get the schema for.

TYPE: Optional[Type[BaseModel]]

prefix

A prefix to add to the schema title.

TYPE: str DEFAULT: ''

exclude

A sequence of field names to exclude from the schema.

TYPE: Sequence[str] DEFAULT: ()

RETURNS DESCRIPTION
Optional[Dict[str, Any]]

The schema of the model as a dictionary, or None if the model has no fields.

RAISES DESCRIPTION
NotImplementedError

If the model is a silent animal.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/asyncapi/message.py
def get_model_schema(
    call: Optional[Type[BaseModel]],
    prefix: str = "",
    exclude: Sequence[str] = (),
) -> Optional[Dict[str, Any]]:
    """Get the schema of a model.

    Args:
        call: The model class to get the schema for.
        prefix: A prefix to add to the schema title.
        exclude: A sequence of field names to exclude from the schema.

    Returns:
        The schema of the model as a dictionary, or None if the model has no fields.

    Raises:
        NotImplementedError: If the model is a silent animal.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    if call is None:
        return None

    params = {k: v for k, v in get_model_fields(call).items() if k not in exclude}
    params_number = len(params)

    if params_number == 0:
        return None

    model = None
    use_original_model = False
    if params_number == 1:
        name, param = tuple(params.items())[0]

        if (
            param.annotation
            and isclass(param.annotation)
            and issubclass(param.annotation, BaseModel)  # NOTE: 3.7-3.10 compatibility
        ):
            model = param.annotation
            use_original_model = True

    if model is None:
        model = call

    body = model_schema(model)

    if params_number == 1 and not use_original_model:
        param_body = body.get("properties", {})
        param_body = param_body[name]

        if PYDANTIC_V2:
            original_title = param.title
        else:
            original_title = param.field_info.title  # type: ignore[attr-defined]

        if original_title:
            use_original_model = True
            param_body["title"] = original_title
        else:
            param_body["title"] = name

        body = param_body

    if not use_original_model:
        body["title"] = f"{prefix}:Payload"

    return body

Last update: 2023-11-13