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.

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.

    """
    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 = next(iter(params.items()))
        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: Dict[str, Any] = model_schema(model)
    body["properties"] = body.get("properties", {})
    for i in exclude:
        body["properties"].pop(i, None)
    if required := body.get("required"):
        body["required"] = list(filter(lambda x: x not in exclude, required))

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

        if defs := body.get(DEF_KEY):
            param_body[DEF_KEY] = defs

        original_title = param.title if PYDANTIC_V2 else 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