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)

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)
listen()	Connect to the message pulsar.
model_json_schema(*args, **kwargs)	Generates a JSON schema for a model class.
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

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]

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

Generate the code for the frontend module.

classmethod get_api_info() → ModuleAPIModel

Get the API info on the module.

classmethod handle_message(request_msg)

classmethod listen()

Connect to the message pulsar.

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]

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.

pulsar: ClassVar[MessageConsumer]

root: BackendFunction | BackendClass

classmethod send_request(request: BackendModule | IO | Dict[str, Any]) → BaseModel

Test a request to the backend.

Parameters:

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

classmethod shell()

Start a shell with the module defined.

classmethod test_connect()

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, 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)

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.

• 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_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]

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'), '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])

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