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, template)	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[source]

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_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
`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][source]

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[source]: Get the API info on the function.

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

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

Generates a JSON schema for a model class.

Parameters:

by_alias – Whether to use attribute aliases or not.
ref_template – The reference template.
union_format –
The format to use when combining schemas from unions together. Can be one of:
- ’any_of’: Use the [anyOf](https://json-schema.org/understanding-json-schema/reference/combining#anyOf)
keyword to combine schemas (the default). - ‘primitive_type_array’: Use the [type](https://json-schema.org/understanding-json-schema/reference/type) keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive type (string, boolean, null, integer or number) or contains constraints/metadata, falls back to any_of.
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: ~demessaging.config.registry.ApiRegistry = <factory>, template: ~demessaging.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: ~inspect.Signature | None = None, validators: ~typing.Dict[str, ~typing.List[~pydantic.types.ImportString | ~typing.Annotated[~typing.Callable, ~pydantic.functional_serializers.PlainSerializer(func=~demessaging.utils.object_to_string, return_type=PydanticUndefined, when_used=always)]]] = <factory>, serializers: ~typing.Dict[str, ~pydantic.types.ImportString | ~typing.Annotated[~typing.Callable, ~pydantic.functional_serializers.PlainSerializer(func=~demessaging.utils.object_to_string, return_type=PydanticUndefined, when_used=always)]] = <factory>, return_validators: ~typing.List[~pydantic.types.ImportString | ~typing.Annotated[~typing.Callable, ~pydantic.functional_serializers.PlainSerializer(func=~demessaging.utils.object_to_string, return_type=PydanticUndefined, when_used=always)]] | None = None, return_serializer: ~pydantic.types.ImportString | ~typing.Annotated[~typing.Callable, ~pydantic.functional_serializers.PlainSerializer(func=~demessaging.utils.object_to_string, return_type=PydanticUndefined, when_used=always)] | None = None, field_params: ~typing.Dict[str, ~typing.Dict[str, ~typing.Any]] = <factory>, returns: ~typing.Dict[str, ~typing.Any] = <factory>, return_annotation: ~typing.Any | None = None, annotations: ~typing.Dict[str, ~typing.Any] = <factory>, reporter_args: ~typing.Dict[str, ~deprogressapi.base.BaseReport] = <factory>, json_schema_extra: ~typing.Dict[str, ~typing.Any] = <factory>, function: ~typing.Any, class_name: str)[source]

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 0x71be7455feb0>, 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 0x71be7455feb0>, 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 0x71be7455feb0>, 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 0x71be7455feb0>, 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_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

registry: ApiRegistry

reporter_args: Dict[str, BaseReport]

return_annotation: Any | None

return_serializer: ImportString | Annotated[Callable, PlainSerializer(object_to_string)] | None

return_validators: List[ImportString | Annotated[Callable, PlainSerializer(object_to_string)]] | None

returns: Dict[str, Any]

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

signature: inspect.Signature | None

template: Template

update_from_function() → None[source]: Update the config from the corresponding function.

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

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

Bases: BaseModel

A class in the API suitable for RPC via DASF

Attributes:

`model_config`	Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
`name`
`return_schema`
`rpc_schema`

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str

return_schema: JsonSchemaValue

rpc_schema: JsonSchemaValue

class demessaging.backend.function.ReturnModel(root: RootModelRootType = PydanticUndefined)[source]

Bases: RootModel

Attributes:

model_config

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

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

root: RootModelRootType

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

Generate field for the return property.

Parameters:: docstring (docstring_parser.Docstring) – The parser that analyzed the docstring
Returns:: The pydantic field
Return type:: Any