How to define and use custom data types

In the hello world example from the quick-start guide we only used the string data type. Here we will show you how to define more complex composite data types and use them either as function parameters or return values.

Therefore we extend the previous HelloWorld example.

Exposing classes

In order to support most use case scenarios you can not just expose individual functions via DASF, but entire classes. So before we dive into custom data types, let’s convert our tiny hello_world function into a hello world class.

HelloWorld class
1from typing import List
2
3
4class HelloWorld:
5    def __init__(self, message: str):
6        self.message = message
7
8    def repeat_message(self, repeat: int) -> List[str]:
9        return [self.message] * repeat

The given HelloWorld class defines a constructor that expects a string parameter called message and a single function called repeat_message that takes an integer and returns a list of strings. Now the idea is, that objects of this HelloWorld class are instantiated with a message string, that then will be used in its repeat_message along with the repeat parameter to generate the list of strings (repeating the message parameter repeat times) that are returned.

Now, to expose this class through a DASF backend module all we have to do is import and call the main function from the demessaging package and register the class via __all__. So our example above becomes:

HelloWorld class exposed via a HelloWorld backend module.
 1from typing import List
 2from demessaging import main
 3
 4__all__ = ["HelloWorld"]
 5
 6
 7class HelloWorld:
 8    def __init__(self, message: str):
 9        self.message = message
10
11    def repeat_message(self, repeat: int) -> List[str]:
12        return [self.message] * repeat
13
14
15if __name__ == "__main__":
16    main(
17        messaging_config=dict(topic="hello-world-class-topic")
18    )

Configuring exposed classes and functions through annotations

Sometimes you might not want to expose all functions of a class, like private/internal ones. This can be configured via the @configure annotation. Furthermore you might want to assert a certain value range for the method arguments or returns. This can also be configured via the @configure annotation.

Exposed HelloWorld class configured through @configure annotation.
 1from typing import List
 2from demessaging import main, configure
 3
 4__all__ = ["HelloWorld"]
 5
 6
 7@configure(methods=["repeat_message"])
 8class HelloWorld:
 9    def __init__(self, message: str):
10        self.message = message
11
12    @configure(field_params={"repeat": {"ge": 0}})
13    def repeat_message(self, repeat: int) -> List[str]:
14        return [self.message] * repeat
15
16    def unexposed_method(self) -> str:
17        return self.message
18
19
20if __name__ == "__main__":
21    main(
22        messaging_config=dict(topic="hello-world-class-topic")
23    )

Class and function configuration parameters

For a comprehensive list of configuration parameters see: demessaging.config.backend.ClassConfig

Also refer to https://pydantic-docs.helpmanual.io/usage/schema/#field-customisation

Define custom data types

Let’s extent our HelloWorld class even further by defining a custom data type/class that we are going to return in one of our exposed functions. In order to define the data type class we have to register it by using the @registry.register_type annotation. Let’s register a GreetResponse as in the following example:

Custom data type class defined through @registry.register_type annotation.
 1from typing import List
 2import datetime
 3from pydantic import BaseModel
 4from demessaging import main, configure, registry
 5
 6__all__ = ["HelloWorld"]
 7
 8@registry.register_type
 9class GreetResponse(BaseModel):
10    message: str
11    greetings: List[str]
12    greeting_time: datetime.datetime
13
14@configure(methods=["repeat_message"])
15class HelloWorld:
16    def __init__(self, message: str):
17        self.message = message
18
19    @configure(field_params={"repeat": {"ge": 0}})
20    def repeat_message(self, repeat: int) -> List[str]:
21        return [self.message] * repeat
22
23    def unexposed_method(self) -> str:
24        return self.message
25
26
27if __name__ == "__main__":
28    main(
29        messaging_config=dict(topic="hello-world-topic")
30    )

Note that the registered class has to inherit from the pydantic BaseModel class. Once registered we can use it as a function argument type or a return value type, like in the following greet function example.

Custom data type class defined through @registry.register_type annotation.
 1from typing import List
 2import datetime
 3from pydantic import BaseModel
 4from demessaging import main, configure, registry
 5
 6__all__ = ["HelloWorld"]
 7
 8
 9@registry.register_type
10class GreetResponse(BaseModel):
11    message: str
12    greetings: List[str]
13    greeting_time: datetime.datetime
14
15
16@configure(methods=["repeat_message", "greet"])
17class HelloWorld:
18    def __init__(self, message: str):
19        self.message = message
20
21    @configure(field_params={"repeat": {"ge": 0}})
22    def repeat_message(self, repeat: int) -> List[str]:
23        return [self.message] * repeat
24
25    @configure(field_params={"repeat": {"ge": 0}})
26    def greet(self, repeat: int, greet_message: str) -> GreetResponse:
27        greetings: List[str] = [greet_message] * repeat
28        return GreetResponse(
29            message=self.message,
30            greetings=greetings,
31            greeting_time=datetime.datetime.now(),
32        )
33
34    def unexposed_method(self) -> str:
35        return self.message
36
37
38if __name__ == "__main__":
39    main(
40        messaging_config=dict(topic="hello-world-topic")
41    )