Data Platform Libs

Library to provide simple API for promoting typed, validated and structured dataclass in charms.

Dict-like data structure are often used in charms. They are used for config, action parameters and databag. This library aims at providing simple API for using pydantic BaseModel-derived class in charms, in order to enhance:

• Validation, by embedding custom business logic to validate single parameters or even have validators that acts across different fields
• Parsing, by loading data into pydantic object we can both allow for other types (e.g. float) to be used in configuration/parameters as well as specify even nested complex objects for databags
• Static typing checks, by moving from dict-like object to classes with typed-annotated properties, that can be statically checked using mypy to ensure that the code is correct.

Pydantic models can be used on:

• Charm Configuration (as defined in config.yaml)
• Actions parameters (as defined in actions.yaml)
• Application/Unit Databag Information (thus making it more structured and encoded)

Creating models

Any data-structure can be modeled using dataclasses instead of dict-like objects (e.g. storing config, action parameters and databags). Within pydantic, we can define dataclasses that provides also parsing and validation on standard dataclass implementation:

from charms.data_platform_libs.v0.data_models import BaseConfigModel

class MyConfig(BaseConfigModel):

    my_key: int

    @validator("my_key")
     def is_lower_than_100(cls, v: int):
         if v > 100:
             raise ValueError("Too high")

This should allow to collapse both parsing and validation as the dataclass object is parsed and created:

dataclass = MyConfig(my_key="1")

dataclass.my_key # this returns 1 (int)
dataclass["my_key"] # this returns 1 (int)

dataclass = MyConfig(my_key="102") # this returns a ValueError("Too High")

Charm Configuration Model

Using the class above, we can implement parsing and validation of configuration by simply extending our charms using the TypedCharmBase class, as shown below.

class MyCharm(TypedCharmBase[MyConfig]):
    config_type = MyConfig

     # everywhere in the code you will have config property already parsed and validate
     def my_method(self):
         self.config: MyConfig

Action parameters

In order to parse action parameters, we can use a decorator to be applied to action event callbacks, as shown below.

@validate_params(PullActionModel)
def _pull_site_action(
    self, event: ActionEvent,
    params: Optional[Union[PullActionModel, ValidationError]] = None
):
    if isinstance(params, ValidationError):
        # handle errors
    else:
        # do stuff

Note that this changes the signature of the callbacks by adding an extra parameter with the parsed counterpart of the event.params dict-like field. If validation fails, we return (not throw!) the exception, to be handled (or raised) in the callback.

Databag

In order to parse databag fields, we define a decorator to be applied to base relation event callbacks.

@parse_relation_data(app_model=AppDataModel, unit_model=UnitDataModel)
def _on_cluster_relation_joined(
        self, event: RelationEvent,
        app_data: Optional[Union[AppDataModel, ValidationError]] = None,
        unit_data: Optional[Union[UnitDataModel, ValidationError]] = None
) -> None:
    ...

The parameters app_data and unit_data refers to the databag of the entity which fired the RelationEvent.

When we want to access to a relation databag outsides of an action, it can be useful also to compact multiple databags into a single object (if there are no conflicting fields), e.g.

class ProviderDataBag(BaseClass):
    provider_key: str

class RequirerDataBag(BaseClass):
    requirer_key: str

class MergedDataBag(ProviderDataBag, RequirerDataBag):
    pass

merged_data = get_relation_data_as(
    MergedDataBag, relation.data[self.app], relation.data[relation.app]
)

merged_data.requirer_key
merged_data.provider_key

The above code can be generalized to other kinds of merged objects, e.g. application and unit, and it can be extended to multiple sources beyond 2:

merged_data = get_relation_data_as(
    MergedDataBag, relation.data[self.app], relation.data[relation.app], ...
)