Commit 293a17b8 authored by Philipp S. Sommer's avatar Philipp S. Sommer
Browse files

split backend_models into backend module

the backend_models module is too long and complex, so we split it into multiple files
parent 2bae613f
Pipeline #28423 failed with stage
......@@ -28,12 +28,16 @@ class GreetResponse(BaseModel):
greetings: List[str]
greeting_time: datetime.datetime
# you can configure how the individual function is called in the backend module.
# You can configure different validations for the parameters.
# Here, we know that `repeat` must at least be 0, so we set this as a field
# parameter (see https://pydantic-docs.helpmanual.io/usage/schema/#field-customisation)
# parameter (see
# https://pydantic-docs.helpmanual.io/usage/schema/#field-customisation)
# for available validators
@configure(field_params={"repeat": {"ge": 0}})
def hello_world(message: str, repeat: int, greet_message: str) -> GreetResponse:
def hello_world(
message: str, repeat: int, greet_message: str
) -> GreetResponse:
"""Greet the hello world module.
Parameters
......@@ -51,7 +55,11 @@ def hello_world(message: str, repeat: int, greet_message: str) -> GreetResponse:
Hello world greet response object
"""
greetings: List[str] = [greet_message] * repeat
return GreetResponse(message, greetings, datetime.datetime.now())
return GreetResponse(
message=message,
greetings=greetings,
greeting_time=datetime.datetime.now(),
)
# You can also use generic python classes as input/output for your model. In
......@@ -76,13 +84,13 @@ def compute_sum(da: DataArray) -> DataArray:
# You can configure how the individual class in the backend module shall be
# interpreted with the `configure` function. See
# :class:`demessaging.backend_models.ClassConfig` for available parameters
# :class:`demessaging.config.ClassConfig` for available parameters
@configure(methods=["repeat_message"])
class HelloWorld:
"""Greet the world from a class.
Classes can define the methods that shall be used for the backend, and they
define a constructor. """
define a constructor."""
def __init__(self, message: str):
"""
......@@ -106,4 +114,4 @@ class HelloWorld:
if __name__ == "__main__":
main(topic='hello_world')
main(topic="hello_world")
"""Main module to start a backend module from the command-line."""
from demessaging.backend_models import main
from demessaging.backend import main, BackendModule # noqa: F401
from demessaging.config import configure, registry
__all__ = ["main", "configure", "registry"]
__all__ = ["main", "configure", "registry", "BackendModule"]
"""Core module for generating a de-messaging backend module.
This module defines the base classes to serve a general python module as a
backend module in the Digital Earth messaging framework.
The most important members are:
.. autosummary::
main
BackendModule
BackendFunction
BackendClass
"""
from __future__ import annotations
from typing import Type
from demessaging.config import ModuleConfig
from demessaging.backend.module import BackendModule
from demessaging.backend.function import BackendFunction # noqa: F401
from demessaging.backend.class_ import BackendClass # noqa: F401
def main(
module_name: str = "__main__", *args, **config_kws
) -> Type[BackendModule]:
"""Main function for starting a backend module from the command line."""
from demessaging.cli import UNKNOWN_TOPIC, get_parser
default_config: ModuleConfig
if "config" in config_kws:
default_config = config_kws.pop("config")
default_config = default_config.copy(update=config_kws)
else:
config_kws.setdefault("topic", UNKNOWN_TOPIC)
default_config = ModuleConfig(**config_kws)
parser = get_parser(module_name, default_config)
if args:
ns = parser.parse_args(args)
else:
ns = parser.parse_args()
ns_d = vars(ns)
config = ModuleConfig(**ns_d)
Model = BackendModule.create_model(ns.module_name, config=config)
if ns.command:
method = getattr(Model, ns_d.get("method_name", ns.command))
kws = {key: ns_d[key] for key in ns_d.get("command_params", [])}
print(method(**kws))
return Model
"""Transform a python class into a corresponding pydantic model.
The :class:`BackendClass` model in this module generates subclasses based upon
a python class (similarly as the
:class:`~demessaging.backend.function.BackendFunction` does it for functions).
"""
from __future__ import annotations
from functools import partial
from typing import (
Any,
List,
Callable,
Type,
Dict,
cast,
Optional,
Union,
ClassVar,
TYPE_CHECKING,
)
import json
from textwrap import dedent
import inspect
import docstring_parser
from pydantic import ( # pylint: disable=no-name-in-module
BaseModel,
Field,
create_model,
)
from pydantic.json import ( # pylint: disable=no-name-in-module
custom_pydantic_encoder,
)
from demessaging.config import ClassConfig
from demessaging.backend.function import BackendFunction, BackendFunctionConfig
from demessaging.backend import utils
class BackendClassConfig(ClassConfig):
"""Configuration class for a backend module class."""
models: Dict[str, Type[BackendFunction]] = Field(
default_factory=dict,
description=(
"Mapping of method name to the function model for the "
"methods of this class"
),
)
Class: Type[object] = Field(
description="The class that corresponds to this config."
)
class_name: str = Field(description="Name of the model class")
@property
def method_configs(self) -> List[BackendFunctionConfig]:
"""Get a list of the method configs."""
return [model.backend_config for model in self.models.values()]
def update_from_cls(self) -> None:
"""Update the config from the corresponding function."""
Class = self.Class
if not self.name:
self.name = Class.__name__ or ""
if not self.doc:
self.doc = dedent(inspect.getdoc(Class) or "")
if not self.init_doc:
self.init_doc = dedent(inspect.getdoc(Class.__init__) or "")
if not self.signature:
self.signature = inspect.signature(Class.__init__)
class BackendClass(BaseModel):
"""A basis for class models
Do not directly instantiate from this class, rather use the
:meth:`create_model` method.
"""
class Config:
json_encoders = {int: json.dumps}
backend_config: ClassVar[BackendClassConfig]
if TYPE_CHECKING:
# added properties for subclasses generated by create_model
function: BackendFunction
class_name: str
def __call__(self) -> BackendFunction:
kws = self.dict()
ret_model_kws = kws.copy()
ret_model_kws["function"] = ret_model_kws["function"].copy()
kws.pop("class_name")
kws.pop("func_returns", None)
func_kws = kws.pop("function")
ini: Any = self.backend_config.Class(**kws) # type: ignore
func_name = func_kws.pop("func_name")
func_kws.pop("func_returns", None)
ret = getattr(ini, func_name)(**func_kws)
# now update the function model and return it
function: BackendFunction = self.function # type: ignore
function.func_returns = ret
return function
@classmethod
def create_model(
cls,
Class,
config: Optional[ClassConfig] = None,
methods: Optional[
List[Union[Type[BackendFunction], Callable, str]]
] = None,
class_name: Optional[str] = None,
**kwargs: Any,
) -> Type[BackendClass]:
"""Generate a pydantic model from a class. Parameters
----------
func: type
A class
config: ClassConfig, optional
The configuration to use. If given, this overrides the
``__pulsar_config__`` of the given `Class`
methods: list of methods, optional
A list of methods or model classes generated with
:func:`FunctionModel`. This overrides the methods in `config` or
the ``__pulsar_config__`` attribute of `Class`
class_name: str, optional
The name for the generated subclass of :class:`pydantic.BaseModel`.
If not given, the name of `Class` is used
``**kwargs``
Any other parameter for the :func:`pydantic.create_model` function
Returns
-------
Subclass of BackendClass
The newly generated model that represents this class.
"""
sig = inspect.signature(Class.__init__)
docstring = docstring_parser.parse(Class.__doc__)
init_docstring = docstring_parser.parse(Class.__init__.__doc__)
docstring.params.extend(init_docstring.params)
docstring.meta.extend(init_docstring.meta)
if config is None:
config = getattr(Class, "__pulsar_config__", ClassConfig())
name = Class.__name__
if not class_name:
class_name = utils.snake_to_camel("Class", name)
config = BackendClassConfig(
Class=Class, class_name=class_name, **config.copy(deep=True).dict()
)
fields = utils.get_fields(name, sig, docstring, config)
fields["class_name"] = fields.pop("func_name")
if "function" in fields:
raise ValueError(
f"`function` must not be an init parameter for {name}!"
)
if methods:
pass
elif config.methods:
methods = list(config.methods)
if not methods:
names_members = inspect.getmembers(
Class, predicate=inspect.isfunction
)
methods = [t[0] for t in names_members if not t[0].startswith("_")]
if not methods:
raise ValueError("No methods of the class have been specified!")
for method in methods:
if inspect.isclass(method) and issubclass(method, BackendFunction): # type: ignore # noqa: E501
method_name: str = method.backend_config.name # type: ignore
FuncModel: Type[BackendFunction] = cast(
Type[BackendFunction], method
)
elif callable(method):
method_name = cast(str, method.__name__)
FuncModel = BackendFunction.create_model(
cast(Callable, method),
class_name=utils.snake_to_camel(
"Meth", class_name, method_name
),
)
else:
method_name = method
FuncModel = BackendFunction.create_model(
getattr(Class, method_name),
class_name=utils.snake_to_camel(
"Meth", class_name, method_name
),
)
if method_name not in config.models:
config.models[method_name] = FuncModel
config.methods = list(config.models)
models = list(config.models.values())
function_types = models[0]
for model in models[1:]:
function_types = Union[function_types, model] # type: ignore
fields["function"] = (
function_types,
Field(description="The method to call."),
)
kwargs.update(fields)
Model: Type[BackendClass] = create_model( # type: ignore
class_name,
__validators__=config.validators,
__module__=Class.__module__,
__base__=cls,
**kwargs, # type: ignore
)
Model.backend_config = config
if config.registry.json_encoders:
# it would be better, to set this via __config__ in create_model,
# but this is not possible if we use `__base__`
Model.__config__.json_encoders = config.registry.json_encoders
Model.__json_encoder__ = partial(
custom_pydantic_encoder, config.registry.json_encoders
)
config.Class = Class
config.update_from_cls()
Model.__doc__ = config.doc
return Model
ClassConfig.update_forward_refs()
"""Transform a python function into a corresponding pydantic model.
The :class:`BackendFunction` model in this module generates subclasses based
upon a python class (similarly as the
:class:`~demessaging.backend.class_.BackendClass` does it for classes).
"""
from __future__ import annotations
from functools import partial
from typing import (
Any,
Callable,
Type,
Dict,
cast,
Optional,
ClassVar,
TYPE_CHECKING,
)
import warnings
from textwrap import dedent
import inspect
import docstring_parser
from pydantic import ( # pylint: disable=no-name-in-module
BaseModel,
Field,
create_model,
)
from pydantic.json import ( # pylint: disable=no-name-in-module
custom_pydantic_encoder,
)
from demessaging.config import FunctionConfig
import demessaging.backend.utils as utils
def get_return_field(
docstring: docstring_parser.Docstring, config: FunctionConfig
) -> Any:
"""Generate field for the return property
Our function models get a ``func_returns`` property to highlight the return
type for the user.
Parameters
----------
docstring : docstring_parser.Docstring
The parser that analyzed the docstring
Returns
-------
Any
The pydantic field
"""
return_description = ""
ret_count: int = 0
for arg in docstring.meta:
if (
isinstance(arg, docstring_parser.DocstringReturns)
and arg.description
):
return_description += "\n- " + arg.description
ret_count += 1
return_description = return_description.strip()
if ret_count == 1:
return_description = return_description[2:]
field_kws: Dict[str, Any] = {"default": None}
if return_description.strip():
field_kws["description"] = return_description
field_kws.update(config.returns)
return Field(**field_kws)
class BackendFunctionConfig(FunctionConfig):
"""Configuration class for a backend module function."""
function: Any = Field(description="The function to call.")
class_name: str = Field(description="Name of the model class")
def update_from_function(self) -> None:
"""Update the config from the corresponding function."""
func = self.function
if not self.name:
self.name = func.__name__ or ""
if not self.doc:
self.doc = dedent(inspect.getdoc(func) or "")
if not self.signature:
self.signature = inspect.signature(func)
class BackendFunction(BaseModel):
"""A base class for a function model.
Don't use this model, rather use :meth:`create_model` method to
generate new models.
"""
class Config:
validate_assignment = True
backend_config: ClassVar[BackendFunctionConfig]
if TYPE_CHECKING:
# added properties for subclasses generated by create_model
func_name: str
func_returns: Any
def __call__(self) -> BackendFunction: # type: ignore
kws = self.dict()
kws.pop("func_name")
kws.pop("func_returns", None)
ret = self.backend_config.function(**kws)
self.func_returns = ret
return self
@classmethod
def create_model(
cls,
func: Callable,
config: Optional[FunctionConfig] = None,
class_name=None,
**kwargs,
) -> Type[BackendFunction]:
"""Create a new pydantic Model from a function.
Parameters
----------
func: callable
A function or method
config: FunctionConfig, optional
The configuration to use. If given, this overrides the
``__pulsar_config__`` of the given `func`
class_name: str, optional
The name for the generated subclass of :class:`pydantic.BaseModel`.
If not given, the name of `func` is used
``**kwargs``
Any other parameter for the :func:`pydantic.create_model` function
Returns
-------
Subclass of BackendFunction
The newly generated class that represents this function.
"""
sig = inspect.signature(func)
docstring = docstring_parser.parse(func.__doc__)
if config is None:
config = getattr(func, "__pulsar_config__", FunctionConfig())
name = cast(str, func.__name__)
if not class_name:
class_name = utils.snake_to_camel("Func", name)
config = BackendFunctionConfig(
function=func,
class_name=class_name,
**config.copy(deep=True).dict(),
)
fields = utils.get_fields(name, sig, docstring, config)
desc = utils.get_desc(docstring)
ret_field = get_return_field(docstring, config)
if sig.return_annotation is not sig.empty:
fields["func_returns"] = sig.return_annotation, ret_field
else:
warnings.warn(
f"Missing return signature for {func.__name__}!",
RuntimeWarning,
)
fields["func_returns"] = Any, ret_field
kwargs.update(fields)
Model: Type[BackendFunction] = create_model( # type: ignore
class_name,
__validators__=config.validators,
__base__=cls,
__module__=func.__module__,
**kwargs, # type: ignore
)
Model.backend_config = config
if config.registry.json_encoders:
# it would be better, to set this via __config__ in create_model,
# but this is not possible if we use `__base__`
Model.__config__.json_encoders = config.registry.json_encoders
Model.__json_encoder__ = partial(
custom_pydantic_encoder, config.registry.json_encoders
)
config.function = func
config.update_from_function()
if desc:
Model.__doc__ = desc
else:
Model.__doc__ = ""
return Model
BackendFunctionConfig.update_forward_refs()
"""Base class for a de-messaging backend module.
"""Backend module to transform a python module into a pydantic model.
This module defines a base class to serve a general python module as a
backend module in the Digital Earth messaging framework.
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 :class:`BackendModule` for
details.
"""
from __future__ import annotations