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

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
- BackendFunction classes ( generated with create_model())
- BackendClass classes ( generated with create_model())
- functions (that will then be transformed using create_model())
- classes (that will then be transformed using create_model())
- strings, in which case they point to the member of the given module_name
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_config: ClassVar[ConfigDict] = {}: 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.
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: ~demessaging.config.registry.ApiRegistry = <factory>, template: ~demessaging.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: ~demessaging.config.messaging.PulsarConfig | ~demessaging.config.messaging.WebsocketURLConfig, listen_config: ~demessaging.config.backend.ListenConfig = <factory>, log_config: ~demessaging.config.logging.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: ~typing.List[str | ~typing.Callable | ~typing.Type[object] | ~typing.Any] = <factory>, imports: str = '', json_schema_extra: ~typing.Dict[str, ~typing.Any] = <factory>, models: ~typing.List[~typing.Any] = <factory>, module: ~typing.Any, class_name: str)[source]

Bases: ModuleConfig

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
models (List[Any]) – a list of function or class models for the members of the backend module
module (Any) – The imported backend module (or none, if there is none)
class_name (str) – Name of the model class

Attributes:

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

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

classes: List[ClassAPIModel]

functions: List[FunctionAPIModel]

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

rpc_schema: JsonSchemaValue