demessaging.config.backend module

Configuration classes for the de-messaging backend module.

Classes:

`BaseConfig`(*[, doc, registry, template])	Configuration base class for functions, modules and classes.
`ClassConfig`(*[, doc, registry, template, ...])	Configuration class for a backend module class.
`FunctionConfig`(*[, doc, registry, template, ...])	Configuration class for a backend module function.
`ListenConfig`(*[, doc, registry, template, ...])	A configuration for the listen command
`ModuleConfig`(*[, doc, registry, template, ...])	Configuration class for a backend module.

class demessaging.config.backend.BaseConfig(*, doc: str = '', registry: ApiRegistry = None, template: Template = Template(name='empty', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}))[source]

Bases: BaseModel

Configuration base class for functions, modules and classes.

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 this object for the generated API.

Attributes:

`doc`
`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].
`registry`
`template`

Methods:

render(**context)

Generate the code to call this function in the frontend.

doc: str

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]] = {'doc': FieldInfo(annotation=str, required=False, default='', description='The documentation of the object. If empty, this will be taken from the corresponding ``__doc__`` attribute.'), 'registry': FieldInfo(annotation=ApiRegistry, required=False, default_factory=_get_registry, description='Utilities for imports and encoders.'), 'template': FieldInfo(annotation=Template, required=False, default=Template(name='empty', 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 this object for the generated API.')}

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

This replaces Model.__fields__ from Pydantic V1.

registry: ApiRegistry

render(**context) → str[source]: Generate the code to call this function in the frontend.

template: Template

class demessaging.config.backend.ClassConfig(*, doc: str = '', registry: ApiRegistry = None, template: Template = Template(name='class_.py', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}), name: str = '', init_doc: str = '', signature: Signature | None = None, methods: List[str] = None, validators: Dict[str, Any] = None, serializers: Dict[str, Any] = None, field_params: Dict[str, Dict[str, Any]] = None, annotations: Dict[str, Any] = None, reporter_args: Dict[str, BaseReport] = None, json_schema_extra: Dict[str, Any] = None)[source]

Bases: BaseConfig

Configuration class for a backend module class.

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 class for the generated API.
name (str) – The name of the function. If empty, this will be taken from the classes __name__ attribute.
init_doc (str) – The documentation of the function. If empty, this will be taken from the classes __init__ method.
signature (Optional[inspect.Signature]) – The calling signature for the function. If empty, this will be taken from the function itself.
methods (List[str]) – methods to use within the backend modules
validators (Dict[str, Any]) – custom validators for the constructor parameters
serializers (Dict[str, Any]) – A mapping from function argument to serializer that is of instance pydantic.functional_serializers.PlainSerializer or pydantic.functional_serializers.WrapSerializer.
field_params (Dict[str, Dict[str, Any]]) – custom Field overrides for the constructor parameters. See pydantic.Fields.Field()
annotations (Dict[str, Any]) – custom annotations for constructor 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

Attributes:

`annotations`
`field_params`
`init_doc`
`json_schema_extra`
`methods`
`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`
`reporter_args`
`serializers`
`signature`
`template`
`validators`

annotations: Dict[str, Any]

doc: str

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

init_doc: str

json_schema_extra: Dict[str, Any]

methods: List[str]

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 constructor parameters'), '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`'), 'init_doc': FieldInfo(annotation=str, required=False, default='', description='The documentation of the function. If empty, this will be taken from the classes ``__init__`` method.'), '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'), 'methods': FieldInfo(annotation=List[str], required=False, default_factory=list, description='methods to use within the backend modules'), 'name': FieldInfo(annotation=str, required=False, default='', description='The name of the function. If empty, this will be taken from the classes ``__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'), 'serializers': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='A mapping from function argument to serializer that is of instance :class:`pydantic.functional_serializers.PlainSerializer` or :class:`pydantic.functional_serializers.WrapSerializer`.'), '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='class_.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 class for the generated API.'), 'validators': FieldInfo(annotation=Dict[str, Any], required=False, default_factory=dict, description='custom validators for the constructor parameters')}

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

serializers: Dict[str, Any]

signature: inspect.Signature | None

template: Template

validators: Dict[str, Any]

class demessaging.config.backend.FunctionConfig(*, 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 | Annotated[Callable, PlainSerializer(func=object_to_string, return_type=PydanticUndefined, when_used=always)]]] = None, serializers: Dict[str, ImportString | Annotated[Callable, PlainSerializer(func=object_to_string, return_type=PydanticUndefined, when_used=always)]] = None, return_validators: List[ImportString | Annotated[Callable, PlainSerializer(func=object_to_string, return_type=PydanticUndefined, when_used=always)]] | None = None, return_serializer: ImportString | Annotated[Callable, PlainSerializer(func=object_to_string, return_type=PydanticUndefined, when_used=always)] | 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)[source]

Bases: BaseConfig

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 0x7f08170131c0>, 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 0x7f08170131c0>, 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 0x7f08170131c0>, 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 0x7f08170131c0>, 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

Attributes:

`annotations`
`field_params`
`json_schema_extra`
`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`
`reporter_args`
`return_annotation`
`return_serializer`
`return_validators`
`returns`
`serializers`
`signature`
`template`
`validators`

annotations: Dict[str, Any]

doc: str

field_params: Dict[str, Dict[str, 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'), '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`'), '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: 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

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

class demessaging.config.backend.ListenConfig(*, doc: str = '', registry: ApiRegistry = None, template: Template = Template(name='empty', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}), dump_to: str | None = None, dump_tool: str | None = None, cmd: str | None = None)[source]

Bases: BaseConfig

A configuration for the listen command

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 this object for the generated API.
dump_to (Optional[str]) – None
dump_tool (Optional[str]) – None
cmd (Optional[str]) – None

Attributes:

`cmd`
`dump_to`
`dump_tool`
`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].

cmd: str | None

doc: str

dump_to: str | None

dump_tool: str | None

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]] = {'cmd': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'doc': FieldInfo(annotation=str, required=False, default='', description='The documentation of the object. If empty, this will be taken from the corresponding ``__doc__`` attribute.'), 'dump_to': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'dump_tool': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'registry': FieldInfo(annotation=ApiRegistry, required=False, default_factory=_get_registry, description='Utilities for imports and encoders.'), 'template': FieldInfo(annotation=Template, required=False, default=Template(name='empty', 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 this object for the generated API.')}

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

This replaces Model.__fields__ from Pydantic V1.

registry: ApiRegistry

template: Template

class demessaging.config.backend.ModuleConfig(*, doc: str = '', registry: ApiRegistry = None, template: Template = Template(name='module.py', folder=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/templates'), suffix='.jinja2', context={}), messaging_config: PulsarConfig | WebsocketURLConfig, listen_config: ListenConfig = None, log_config: LoggingConfig = LoggingConfig(config_file=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/config/logging.yaml'), level=None, logfile=None, config_overrides=None, merge_config=False), debug: bool = False, members: List[str | Callable | Type[object] | Any] = None, imports: str = '', json_schema_extra: Dict[str, Any] = None)[source]

Bases: BaseConfig

Configuration class for a backend module.

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 module for the generated API.
messaging_config (Union[demessaging.config.messaging.PulsarConfig, demessaging.config.messaging.WebsocketURLConfig]) – Configuration on how to connect to the message broker.
listen_config (demessaging.config.backend.ListenConfig) – None
log_config (demessaging.config.logging.LoggingConfig) – Configuration for the logging.
debug (bool) – Run the backend module in debug mode (creates more verbose error messages).
members (List[Union[str, Callable, Type[object], Any]]) – List of members for this module
imports (str) – Imports that should be added to the generate API module.
json_schema_extra (Dict[str, Any]) – Any extra parameter for the JSON schema export for the function

Attributes:

`debug`
`imports`
`json_schema_extra`
`listen_config`
`log_config`
`members`
`messaging_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].
`pulsar_config`	DEPRECATED! Get the messaging configuration.
`template`

debug: bool

doc: str

imports: str

json_schema_extra: Dict[str, Any]

listen_config: ListenConfig

log_config: LoggingConfig

members: List[str | Callable | Type[object] | Any]

messaging_config: PulsarConfig | WebsocketURLConfig

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]] = {'debug': FieldInfo(annotation=bool, required=False, default=False, description='Run the backend module in debug mode (creates more verbose error messages).'), 'doc': FieldInfo(annotation=str, required=False, default='', description='The documentation of the object. If empty, this will be taken from the corresponding ``__doc__`` attribute.'), 'imports': FieldInfo(annotation=str, required=False, default='', description='Imports that should be added to the generate API module.'), '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'), 'listen_config': FieldInfo(annotation=ListenConfig, required=False, default_factory=ListenConfig), 'log_config': FieldInfo(annotation=LoggingConfig, required=False, default=LoggingConfig(config_file=PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/dasf/checkouts/latest/demessaging/config/logging.yaml'), level=None, logfile=None, config_overrides=None, merge_config=False), description='Configuration for the logging.'), 'members': FieldInfo(annotation=List[Union[str, Callable, Type[object], Any]], required=False, default_factory=list, description='List of members for this module'), 'messaging_config': FieldInfo(annotation=Union[PulsarConfig, WebsocketURLConfig], required=True, description='Configuration on how to connect to the message broker.'), 'registry': FieldInfo(annotation=ApiRegistry, required=False, default_factory=_get_registry, description='Utilities for imports and encoders.'), 'template': FieldInfo(annotation=Template, required=False, default=Template(name='module.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 module for the generated API.')}

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

This replaces Model.__fields__ from Pydantic V1.

property pulsar_config: PulsarConfig | WebsocketURLConfig

DEPRECATED! Get the messaging configuration.

Please use the messaging_config attribute of this class.

registry: ApiRegistry

template: Template