Skip to content

get_model_schema

faststream.asyncapi.message.get_model_schema #

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

Get the schema of a model.

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."""
    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):
            # single argument with useless reference
            if param_body.get("$ref"):
                ref_obj: Dict[str, Any] = next(iter(defs.values()))
                return ref_obj
            else:
                param_body[DEF_KEY] = defs

        original_title = param.title if PYDANTIC_V2 else param.field_info.title

        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