demessaging package

Data Analytics Software Framework

python module wrapper for the data analytics software framework DASF

Classes:

BackendModule([root])

A base class for a backend module.

Functions:

configure([js, merge])

Configuration decorator for function or modules.

main([module_name])

Main function for starting a backend module from the command line.
class demessaging.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.

demessaging.configure(js: str | None = None, merge: bool = True, **kwargs) → Callable[[T], T][source]

Configuration decorator for function or modules.

Use this function as a decorator for classes or functions in the backend module like so:

>>> @configure(field_params={"a": {"gt": 0}}, returns={"gt": 0})
... def sqrt(a: float) -> float:
...     import math
...
...     return math.sqrt(a)

The available parameters for this function vary depending on what you are decorating. If you are decorating a class, your parameters must be valid for the ClassConfig. If you are decorating a function, your parameters must be valid for a FunctionConfig.

Parameters:

  • js (Optional[str]) – A JSON-formatted string that can be used to setup the config.

  • merge (bool) – If True (default), then the configuration will be merged with the existing configuration for the function (if existing)

  • **kwargs – Any keyword argument that can be used to setup the config.

Notes

If you are specifying any kwargs, your first argument (js) should be None.

demessaging.main(module_name: str = '__main__', *args, **config_kws) → Type[BackendModule][source]

Main function for starting a backend module from the command line.

Parameters:

Subpackages

Submodules