Manage libraries

Use a library

In your src/charm.py file, observe the custom events that the library provides. For example, a database library may have provided a ready event – a high-level wrapper around the relevant Juju relation events. You can use the ready event to manage the database relation in your charm:

import ops
from charms.charm_with_lib.v0.database_lib import DatabaseReadyEvent, DatabaseRequirer


class MyCharm(ops.CharmBase):
    def __init__(self, framework: ops.Framework):
        super().__init__(framework)
        self.database = DatabaseRequirer(self, 'db-relation')
        framework.observe(self.database.on.ready, self._on_db_ready)

    def _on_db_ready(self, event: DatabaseReadyEvent):
        secret_content = event.credential_secret.get_content()
        ...

A unit test for the charm that uses the database library looks like:

from ops import testing
from charms.charm_with_lib.v0.database_lib import DatabaseRequirer

def test_ready_event():
    ctx = testing.Context(MyCharm)
    secret = testing.Secret({'username': 'admin', 'password': 'admin'})
    state_in = testing.State(secrets={secret})

    state_out = ctx.run(ctx.on.custom(DatabaseRequirer, credential_secret=secret), state_in)

    assert ...

Write a library

When you’re writing libraries, instead of callbacks, use custom events. This results in a more Ops-native-feeling API. From a technical standpoint, a custom event is an ops.EventBase subclass that can be emitted to Ops at any point throughout the charm’s lifecycle. These events are totally unknown to Juju. They are essentially charm-internal, and can be useful to abstract certain conditional workflows and wrap the top level Juju event so it can be observed independently.

Important

Charms should never define custom events themselves. They have no need for emitting events (custom or otherwise) for their own consumption, and as they lack consumers, they don’t need to emit any for others to consume either. Custom events should only be defined and emitted in a library.

Custom events must inherit from EventBase, but not from an Ops subclass of EventBase, such as RelationEvent. When instantiating the custom event, load any data needed from Juju from the originating event, and explicitly pass that to the custom event object.

For example, suppose you have a charm library wrapping a relation endpoint. The wrapper might want to check that the remote end has sent valid data and, if that is the case, communicate it to the charm. In this example, you have a DatabaseRequirer object, and the charm using it is interested in knowing when the database is ready. In your lib/charms/my_charm/v0/my_lib.py file, the DatabaseRequirer then will be:

class DatabaseReadyEvent(ops.EventBase):
    """Event representing that the database is ready."""

    def __init__(self, handle: ops.Handle, *, credential_secret: ops.Secret):
        super().__init__(handle)
        self.credential_secret = credential_secret

    def snapshot(self) -> dict[str, str]:
        data = super().snapshot()
        data['credential_secret_id'] = self.credential_secret.id
        return data

    def restore(self, snapshot: dict[str, Any]):
        super().restore(snapshot)
        credential_secret_id = snapshot['credential_secret_id']
        self.credential_secret = self.framework.model.get_secret(id=credential_secret_id)


class DatabaseRequirerEvents(ops.ObjectEvents):
    """Container for Database Requirer events."""
    ready = ops.charm.EventSource(DatabaseReadyEvent)


class DatabaseRequirer(ops.Object):
    on = DatabaseRequirerEvents()

    def __init__(self, charm: ops.CharmBase, relation_name: str):
        super().__init__(charm, relation_name)
        self.framework.observe(charm.on['database'].relation_changed, self._on_db_changed)

    def _on_db_changed(self, event: ops.RelationChangedEvent):
        if remote_data_is_valid(event.relation):
            secret = ...
            self.on.ready.emit(credential_secret=secret)

Best practice

Libraries should never mutate the status of a unit or application. Instead, use return values, or raise exceptions and let them bubble back up to the charm for the charm author to handle as they see fit.

Test that the library initialises

In your tests/unit/test_my_lib.py file, add a test that validates that a charm can initialise the library, and that no events are unexpectedly emitted.

import pytest
import ops
from ops import testing
from lib.charms.my_Charm.v0.my_lib import DatabaseRequirer


class MyTestCharm(ops.CharmBase):
    META = {
        "name": "my-charm"
    }
    def __init__(self, framework: ops.Framework):
        super().__init__(framework)
        self.db = DatabaseRequirer(self, 'my-relation')


@pytest.mark.parametrize('event', (
    'start', 'install', 'stop', 'remove', 'update-status', #...
))
def test_charm_runs(event):
    """Verify that the charm can create the library object, and doesn't see unexpected events."""
    ctx = testing.Context(MyTestCharm, meta=MyTestCharm.META)
    state_in = testing.State()
    ctx.run(getattr(ctx.on, event), state_in)
    assert len(ctx.emitted_events) == 0
    assert isinstance(ctx.emitted_events[0], ops.StartEvent)

Test custom endpoint names

If DatabaseRequirer is a relation endpoint wrapper, a frequent pattern is to allow customising the name of the endpoint that the object is wrapping.

Examples: Traefik’s ingress-per-unit lib

In your tests/unit/test_my_lib.py file, add a test that validates that custom names are supported:

import pytest
import ops
from ops import testing
from lib.charms.my_charm.v0.my_lib import DatabaseRequirer


@pytest.fixture(params=["foo", "bar"])
def endpoint(request):
    return request.param


@pytest.fixture
def my_charm_type(endpoint: str):
    class MyTestCharm(ops.CharmBase):
        META = {
            "name": "my-charm",
            "requires":
                {endpoint: {"interface": "my_interface"}}
        }

        def __init__(self, framework: ops.Framework):
            super().__init__(framework)
            self.db = DatabaseRequirer(self, endpoint=endpoint)
            framework.observe(self.on.start, self._on_start)
            self.saw_start = False

        def _on_start(self, _):
            self.saw_start = True

    return MyTestCharm


@pytest.fixture
def context(my_charm_type):
    return testing.Context(my_charm_type, meta=my_charm_type.META)


def test_charm_runs(context):
    """Verify that the charm executes regardless of how we name the requirer endpoint."""
    state_in = testing.State()
    with context(context.on.start(), state_in) as mgr:
        mgr.run()
        assert mgr.charm.saw_start

Test that the custom event is emitted

To verify that the library does emit the custom event appropriately, add a test in your tests/unit/test_my_lib.py file:

def test_ready_event():
    ctx = testing.Context(MyTestCharm, meta=MyTestCharm.META)
    relation = testing.Relation('database')
    secret = testing.Secret({'username': 'admin', 'password': 'admin'})
    state_in = testing.State(relations={relation}, secrets={secret})
    ctx.run(ctx.on.relation_changed(relation), state_in)
    relation_changed_event, custom_event = ctx.emitted_events
    assert isinstance(relation_changed_event, ops.RelationChangedEvent)
    assert isinstance(custom_event, DatabaseReadyEvent)
    assert custom_event.credential_secret.id == secret.id

Write a library that manages a relation interface

First, for a new interface, design the interface and the process that related charms will use to populate the relation data. In the simplest case, the providing charm might populate the local app data when the relation is created, but a conversation between the charms is more common. In a conversation, the requiring charm will populate its local data with a request, and the providing charm will use that to provide a suitable response. In more complex cases, this conversation might have multiple stages.

If the library is implementing an existing interface, find the interface documentation by following links from the Integrations tab on Charmhub, or by navigating to https://charmhub.io/integrations/{integration-name}. Alternatively, the interface documentation can be found in the charm-relation-interfaces repository.

In the interface documentation, find the description of the various databags. For example, for v2 of the tracing interface:

# unit_data: <empty>
application_data:
  receivers:
    - protocol:
        name: otlp_http
        type: http
      url: http://traefik_address:2331
    - protocol:
        name: otlp_grpc
        type: grpc
      url: traefik_address:2331

Use this to create Python classes that model the databags. In many cases, there will already be appropriate classes in the interface documentation. Continuing with the tracing example:

class TransportProtocolType(enum.Enum):
    """Receiver Type."""
    HTTP = "http"
    GRPC = "grpc"


class ProtocolType(pydantic.BaseModel):
    """Protocol Type."""
    name: str = pydantic.Field(
        description="Receiver protocol name. What protocols are supported (and what they are called) "
        "may differ per provider.",
        examples=["otlp_grpc", "otlp_http", "tempo_http", "jaeger_thrift_compact"],
    )
    type: TransportProtocolType = pydantic.Field(
        description="The transport protocol used by this receiver.",
        examples=["http", "grpc"],
    )


class Receiver(pydantic.BaseModel):
    """Specification of an active receiver."""
    protocol: ProtocolType = pydantic.Field(description="Receiver protocol name and type.")
    url: str = pydantic.Field(
        description="""URL at which the receiver is reachable. If there's an ingress, it would be the external URL.
        Otherwise, it would be the service's fqdn or internal IP.
        If the protocol type is grpc, the url will not contain a scheme.""",
        examples=[
            "http://traefik_address:2331",
            "https://traefik_address:2331",
            "http://tempo_public_ip:2331",
            "https://tempo_public_ip:2331",
            "tempo_public_ip:2331",
        ],
    )


class TracingProviderAppData(pydantic.BaseModel):
    receivers: list[Receiver] = pydantic.Field(
        description="A list of enabled receivers in the form of the protocol they use and their resolvable server url.",
    )

Tip

The Ops ops.Relation.load and ops.Relation.save methods serialise and deserialise the values of each field, and default to using JSON, so you do not need to wrap fields in pydantic.Json.

In your src/charm.py file, use the class you created to load and save data from the relation:

receiver_protocol_to_transport_protocol: dict[str, TransportProtocolType] = {
    "zipkin": TransportProtocolType.HTTP,
    "otlp_grpc": TransportProtocolType.GRPC,
    "otlp_http": TransportProtocolType.HTTP,
    "jaeger_thrift_http": TransportProtocolType.HTTP,
    "jaeger_grpc": TransportProtocolType.GRPC,
}

def _publish_provider(self, relation: ops.Relation, receivers: Iterable[tuple[str, str]]):
    data = TracingProviderAppData(
        receivers=[
            Receiver(
                url=url,
                protocol=ProtocolType(
                    name=protocol,
                    type=receiver_protocol_to_transport_protocol[protocol],
                ),
            )
            for protocol, url in receivers
        ],
    )
    relation.save(data, self._charm.app)