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
|