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. |
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
|