demessaging.backend.module module

Backend module to transform a python module into a pydantic model.

This module defines the main model in the demessaging framework. It takes a list of members, or a module, and creates a new Model that can be used to generate code, connect to the pulsar, and more. See BackendModule for details.

Classes:

BackendModule([root])

A base class for a backend module.

BackendModuleConfig(*[, doc, registry, ...])

Configuration class for a backend module.

ModuleAPIModel(*, classes, functions, rpc_schema)

An model that represants the API of a backend module.
class demessaging.backend.module.BackendModule(root: RootModelRootType = PydanticUndefined)[source]

Bases: RootModel

A base class for a backend module.

Do not directly instantiate from this class, rather use the create_model() method.

Parameters:

root (typing.Union[demessaging.backend.function.BackendFunction, demessaging.backend.class_.BackendClass]) – None

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

pulsar

root

Methods:

compute()

Send this request to the backend module and compute the result.

create_model([module_name, members, config, ...])

Generate a module for a backend module.

generate([line_length, use_formatters, ...])

Generate the code for the frontend module.

get_api_info()

Get the API info on the module.

handle_message(request_msg)

_summary_

listen()

_summary_

model_json_schema(*args, **kwargs)

Generates a JSON schema for a model class.

process_request(request)

Test a request to the backend.

send_request(request)

Test a request to the backend.

shell()

Start a shell with the module defined.

test_connect()

Connect to the message pulsar.
backend_config: ClassVar[BackendModuleConfig]
compute() → BaseModel[source]

Send this request to the backend module and compute the result.

This method updates the model inplace.

classmethod create_model(module_name: str | None = None, members: List[Type[BackendFunction] | Type[BackendClass] | Callable | str | Type[object]] | None = None, config: ModuleConfig | None = None, class_name: str | None = None, **config_kws) → Type[BackendModule][source]

Generate a module for a backend module.

Parameters:

  • module_name (str) – The name of the module to import. If none is given, the members must be specified

  • members (list of members) –

    The list of members that shall be added to this module. It can be a list of

  • config (ModuleConfig, optional) – The configuration for the module. If this is not given, you must provide config_kws or define a backend_config variable within the module corresponding to module_name

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

  • **config_kws – An alternative way to specify the configuration for the backend module.

Returns:

The newly generated class that represents this module.

Return type:

Subclass of BackendFunction

classmethod generate(line_length: int = 79, use_formatters: bool = True, use_autoflake: bool = True, use_black: bool = True, use_isort: bool = True) → str[source]

Generate the code for the frontend module.

classmethod get_api_info() → ModuleAPIModel[source]

Get the API info on the module.

classmethod handle_message(request_msg: dict)[source]

_summary_

Parameters:

request_msg (dict) – The message to handle

classmethod listen()[source]

_summary_

Parameters:

  • dump_to (Optional[str], optional) – Instead of processing the request, dump it as a file to the given location. If you need further customization, use –dump-tool.

  • dump_tool (Optional[str], optional) –

    Instead of using –dump-to, use this option to run a specific command for each request. We will first create a temporary file and then run this command as subprocess. This parameter requires –dump-to and two curly brackets ({}) in the argument that specify where to insert the target path. Or use {path} or {basename} or {directory} for more explicit control in your command.

    If you want to process the dumped file further, combine this option with cmd

  • cmd (Optional[str], optional) – Instead of processing the request here, dump the request as file to the disc and run the dedicated command. The specified command must contain two curly braces ({}) that will be replaced with the path or basename of th file. Or use {path}, or {basename} or {directory} for more explicit control in your command.

Examples

Copy the request to a given location via rsync:

BackendModule.listen(dump_tool='rsync {} .')

Copy the request via SSH to another server:

BackendModule.listen(
    dump_tool='scp {} user@machine:/some/folder/'
)

Print the request to stdout and delete the temporary file:

BackendModule.listen(
    dump_tool='cat {path} && rm {path}'
)

Cat the request (i.e. always return the input to the sender):

BackendModule.listen(cmd='cat {}')

Copy the file via scp and run some command to process it on a remote machine:

BackendModule.listen(
    dump_tool='scp {} user@machine:/some/folder/',
    cmd='some-command /some/folder/{basename}',
)
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]] = {'root': FieldInfo(annotation=Union[BackendFunction, BackendClass], 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.

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.

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

classmethod process_request(request: BackendModule | IO | Dict[str, Any]) → ReturnModel[source]

Test a request to the backend.

Parameters:

request (dict or file-like object) – The request that we should process.

pulsar: ClassVar[MessageConsumer]
root: BackendFunction | BackendClass
classmethod send_request(request: BackendModule | IO | Dict[str, Any]) → BaseModel[source]

Test a request to the backend.

Parameters:

request (dict or file-like object) – A request to the backend module.

classmethod shell()[source]

Start a shell with the module defined.

classmethod test_connect()[source]

Connect to the message pulsar.

class demessaging.backend.module.BackendModuleConfig(*, 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, models: List[Any] = None, module: Any, class_name: str)[source]

Bases: ModuleConfig

Configuration class for a backend module.

Parameters:

Attributes:

class_name

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

models

module

class_name: str
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]] = {'class_name': FieldInfo(annotation=str, required=True, description='Name of the model class'), '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.'), 'models': FieldInfo(annotation=List[Any], required=False, default_factory=list, description='a list of function or class models for the members of the backend module'), 'module': FieldInfo(annotation=Any, required=True, description='The imported backend module (or none, if there is 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='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.

models: List[Any]
module: Any
registry: ApiRegistry
template: Template
class demessaging.backend.module.ModuleAPIModel(*, classes: List[ClassAPIModel], functions: List[FunctionAPIModel], rpc_schema: Dict[str, Any])[source]

Bases: BaseModel

An model that represants the API of a backend module.

Attributes:

classes

functions

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

rpc_schema

classes: List[ClassAPIModel]
functions: List[FunctionAPIModel]
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]] = {'classes': FieldInfo(annotation=List[ClassAPIModel], required=True, description='The RPC-enabled classes that this module contains.'), 'functions': FieldInfo(annotation=List[FunctionAPIModel], required=True, description='The RPC-enabled functions that this module contains.'), 'rpc_schema': FieldInfo(annotation=Dict[str, Any], required=True, description='The aggregated JSON schema for an RPC call to this module.')}

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

This replaces Model.__fields__ from Pydantic V1.

rpc_schema: JsonSchemaValue