demessaging.backend.function module

Transform a python function into a corresponding pydantic model.

The BackendFunction model in this module generates subclasses based upon a python class (similarly as the BackendClass does it for classes).

Classes:

BackendFunction()	A base class for a function model.
BackendFunctionConfig(*[, doc, registry, ...])	Configuration class for a backend module function.
FunctionAPIModel(*, name, rpc_schema, ...)	A class in the API suitable for RPC via DASF
ReturnModel([root])

Functions:

get_return_model(docstring, config)

Generate field for the return property.

class demessaging.backend.function.BackendFunction

Bases: BaseModel

A base class for a function model.

Don’t use this model, rather use create_model() method to generate new models.

Attributes:

backend_config
model_computed_fields	A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
model_config	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
model_fields	Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].
return_model

Methods:

create_model(func[, config, class_name])	Create a new pydantic Model from a function.
get_api_info()	Get the API info on the function.
model_json_schema(*args, **kwargs)	Generates a JSON schema for a model class.

backend_config: ClassVar[BackendFunctionConfig]

classmethod create_model(func: Callable, config: FunctionConfig | None = None, class_name=None, **kwargs) → Type[BackendFunction]

Create a new pydantic Model from a function.

Parameters:

• func (callable) – A function or method

• config (FunctionConfig, optional) – The configuration to use. If given, this overrides the __pulsar_config__ of the given func

• class_name (str, optional) – The name for the generated subclass of pydantic.BaseModel. If not given, the name of func is used

• **kwargs – Any other parameter for the pydantic.create_model() function

Returns:

The newly generated class that represents this function.

Return type:

Subclass of BackendFunction

func_name: str

classmethod get_api_info() → FunctionAPIModel

Get the API info on the function.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

classmethod model_json_schema(*args, **kwargs) → Dict[str, Any]

Generates a JSON schema for a model class.

Parameters:

• by_alias – Whether to use attribute aliases or not.

• ref_template – The reference template.

• schema_generator – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

• mode – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

return_model: ClassVar[Type[BaseModel]]

class demessaging.backend.function.BackendFunctionConfig(*, doc: str = '', registry: ApiRegistry = None, template: Template = Template(name='function.py', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}), name: str = '', signature: Signature | None = None, validators: Dict[str, List[ImportString | Callable]] = None, serializers: Dict[str, ImportString | Callable] = None, return_validators: List[ImportString | Callable] | None = None, return_serializer: ImportString | Callable | None = None, field_params: Dict[str, Dict[str, Any]] = None, returns: Dict[str, Any] = None, return_annotation: Any | None = None, annotations: Dict[str, Any] = None, reporter_args: Dict[str, BaseReport] = None, json_schema_extra: Dict[str, Any] = None, function: Any, class_name: str)

Bases: FunctionConfig

Configuration class for a backend module function.

Parameters:

• doc (str) – The documentation of the object. If empty, this will be taken from the corresponding __doc__ attribute.

• registry (demessaging.config.registry.ApiRegistry) – Utilities for imports and encoders.

• template (demessaging.template.Template) – The demessaging.template.Template that is used to render the function for the generated API.

• name (str) – The name of the function. If empty, this will be taken from the functions __name__ attribute.

• signature (Optional[inspect.Signature]) – The calling signature for the function. If empty, this will be taken from the function itself.

• validators (Dict[str, List[Union[pydantic.types.ImportString, Annotated[Callable, PlainSerializer(func=<function object_to_string at 0x7f740a889360>, return_type=PydanticUndefined, when_used='always')]]]]) – Custom validators for function arguments. This parameter is a mapping from function argument name to a list of callables that can be used as validator.

• serializers (Dict[str, Union[pydantic.types.ImportString, Annotated[Callable, PlainSerializer(func=<function object_to_string at 0x7f740a889360>, return_type=PydanticUndefined, when_used='always')]]]) – A mapping from function argument to serializing function that is then used for the pydantic.functional_serializers.PlainSerializer.

• return_validators (Optional[List[Union[pydantic.types.ImportString, Annotated[Callable, PlainSerializer(func=<function object_to_string at 0x7f740a889360>, return_type=PydanticUndefined, when_used='always')]]]]) – Validators for the return value. This parameter is a list of callables that can be used as validator for the return value.

• return_serializer (Union[pydantic.types.ImportString, Annotated[Callable, PlainSerializer(func=<function object_to_string at 0x7f740a889360>, return_type=PydanticUndefined, when_used='always')], NoneType]) – A function that is used to serialize the return value.

• field_params (Dict[str, Dict[str, Any]]) – custom Field overrides for the constructor parameters. See pydantic.Fields.Field()

• returns (Dict[str, Any]) – custom returns overrides.

• return_annotation (Optional[Any]) – The annotation for the return value.

• annotations (Dict[str, Any]) – custom annotations for function parameters

• reporter_args (Dict[str, deprogressapi.base.BaseReport]) – Arguments that use the dasf-progress-api

• json_schema_extra (Dict[str, Any]) – Any extra parameter for the JSON schema export for the function

• function (Any) – The function to call.

• class_name (str) – Name of the model class

Attributes:

class_name
function
model_computed_fields	A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
model_config	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
model_fields	Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

Methods:

update_from_function()

Update the config from the corresponding function.

annotations: Dict[str, Any]

class_name: str

doc: str

field_params: Dict[str, Dict[str, Any]]

function: Any

json_schema_extra: Dict[str, Any]

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'annotations': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='custom annotations for function parameters'), 'class_name': FieldInfo(annotation=str, required=True, description='Name of the model class'), 'doc': FieldInfo(annotation=str, required=False, default='', description='The documentation of the object. If empty, this will be taken from the corresponding ``__doc__`` attribute.'), 'field_params': FieldInfo(annotation=Dict[str, Dict[str, Any]], required=False, default_factory=dict, description='custom Field overrides for the constructor parameters. See :func:`pydantic.Fields.Field`'), 'function': FieldInfo(annotation=Any, required=True, description='The function to call.'), 'json_schema_extra': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='Any extra parameter for the JSON schema export for the function'), 'name': FieldInfo(annotation=str, required=False, default='', description='The name of the function. If empty, this will be taken from the functions ``__name__`` attribute.'), 'registry': FieldInfo(annotation=ApiRegistry, required=False, default_factory=_get_registry, description='Utilities for imports and encoders.'), 'reporter_args': FieldInfo(annotation=Dict[str, BaseReport], required=False, default_factory=dict, description='Arguments that use the dasf-progress-api'), 'return_annotation': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None, description='The annotation for the return value.'), 'return_serializer': FieldInfo(annotation=Union[ImportString, Annotated[Callable, PlainSerializer], NoneType], required=False, default=None, description='A function that is used to serialize the return value.'), 'return_validators': FieldInfo(annotation=Union[List[Union[ImportString, Annotated[Callable, PlainSerializer]]], NoneType], required=False, default=None, description='Validators for the return value. This parameter is a list of callables that can be used as validator for the return value.'), 'returns': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='custom returns overrides.'), 'serializers': FieldInfo(annotation=Dict[str, Union[ImportString, Annotated[Callable, PlainSerializer]]], required=False, default_factory=dict, description='A mapping from function argument to serializing function that is then used for the :class:`pydantic.functional_serializers.PlainSerializer`.'), 'signature': FieldInfo(annotation=Union[Signature, NoneType], required=False, default=None, description='The calling signature for the function. If empty, this will be taken from the function itself.'), 'template': FieldInfo(annotation=Template, required=False, default=Template(name='function.py', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}), description='The :class:`demessaging.template.Template` that is used to render the function for the generated API.'), 'validators': FieldInfo(annotation=Dict[str, List[Union[ImportString, Annotated[Callable, PlainSerializer]]]], required=False, default_factory=dict, description='Custom validators for function arguments. This parameter is a mapping from function argument name to a list of callables that can be used as validator.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

registry: ApiRegistry

reporter_args: Dict[str, BaseReport]

return_annotation: Any | None

return_serializer: Optional[Union[ImportString, Annotated[Callable, PlainSerializer(object_to_string)]]]

return_validators: Optional[List[Union[ImportString, Annotated[Callable, PlainSerializer(object_to_string)]]]]

returns: Dict[str, Any]

serializers: Dict[str, Union[ImportString, Annotated[Callable, PlainSerializer(object_to_string)]]]

signature: inspect.Signature | None

template: Template

update_from_function() → None

Update the config from the corresponding function.

validators: Dict[str, List[Union[ImportString, Annotated[Callable, PlainSerializer(object_to_string)]]]]

class demessaging.backend.function.FunctionAPIModel(*, name: str, rpc_schema: Dict[str, Any], return_schema: Dict[str, Any])

Bases: BaseModel

A class in the API suitable for RPC via DASF

Attributes:

model_computed_fields	A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
model_config	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
model_fields	Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].
name
return_schema
rpc_schema

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'name': FieldInfo(annotation=str, required=True, description='The name of the function that is used as identifier in the RPC.'), 'return_schema': FieldInfo(annotation=Dict[str, Any], required=True, description='The JSON Schema for the return value.'), 'rpc_schema': FieldInfo(annotation=Dict[str, Any], required=True, description='The JSON Schema for the function.')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

return_schema: JsonSchemaValue

rpc_schema: JsonSchemaValue

class demessaging.backend.function.ReturnModel(root: RootModelRootType = PydanticUndefined)

Bases: RootModel

Attributes:

model_computed_fields	A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
model_config	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
model_fields	Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'root': FieldInfo(annotation=~RootModelRootType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

root: RootModelRootType

demessaging.backend.function.get_return_model(docstring: Docstring, config: BackendFunctionConfig) → Type[BaseModel]

Generate field for the return property.

Parameters:

docstring (docstring_parser.Docstring) – The parser that analyzed the docstring

Returns:

The pydantic field

Return type:

Any