ops library API reference

ops module

The ops library: a Python framework for writing Juju charms.

The ops library is a Python framework (available on PyPI) for developing and testing Juju charms in a consistent way, using standard Python constructs to allow for clean, maintainable, and reusable code.

A charm is an operator – business logic encapsulated in a reusable software package that automates every aspect of an application’s life.

Charms written with ops support Kubernetes using Juju’s “sidecar charm” pattern, as well as charms that deploy to Linux-based machines and containers.

Charms should do one thing and do it well. Each charm drives a single application and can be integrated with other charms to deliver a complex system. A charm handles creating the application in addition to scaling, configuration, optimisation, networking, service mesh, observability, and other day-2 operations specific to the application.

The ops library is part of the Charm SDK (the other part being Charmcraft). Full developer documentation for the Charm SDK is available at https://juju.is/docs/sdk.

To learn more about Juju, visit https://juju.is/docs/olm.

class ops.ActionEvent(handle: Handle, id: str | None = None)[source]

Bases: EventBase

Events raised by Juju when an administrator invokes a Juju Action.

This class is the data type of events triggered when an administrator invokes a Juju Action. Callbacks bound to these events may be used for responding to the administrator’s Juju Action request.

To read the parameters for the action, see the instance variable params. To respond with the result of the action, call set_results(). To add progress messages that are visible as the action is progressing use log().

defer() NoReturn[source]

Action events are not deferrable like other events.

This is because an action runs synchronously and the administrator is waiting for the result.

Raises:

RuntimeError – always.

fail(message: str = '')[source]

Report that this action has failed.

Parameters:

message – Optional message to record why it has failed.

id: str = ''

The Juju ID of the action invocation.

log(message: str)[source]

Send a message that a user will see while the action is running.

Parameters:

message – The message for the user.

params: Dict[str, Any]

The parameters passed to the action.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to record the action.

Not meant to be called directly by charm code.

set_results(results: Dict[str, Any])[source]

Report the result of the action.

Juju eventually only accepts a str:str mapping, so we will attempt to flatten any more complex data structure like so:

>>> {'a': 'b'} # becomes: 'a'='b'
>>> {'a': {'b': 'c'}} # becomes: 'a.b'='c'
>>> {'a': {'b': 'c', 'd': 'e'}} # becomes: 'a.b'='c', 'a.d' = 'e'
>>> {'a.b': 'c', 'a.d': 'e'} # equivalent to previous

Note that duplicate keys are not allowed, so this is invalid:

>>> {'a': {'b': 'c'}, 'a.b': 'c'}

Note that the resulting keys must start and end with lowercase alphanumeric, and can only contain lowercase alphanumeric, hyphens and periods.

Because results are passed to Juju using the command line, the maximum size is around 100KB. However, actions results are designed to be small: a few key-value pairs shown in the Juju CLI. If larger content is needed, store it in a file and use something like juju scp.

If any exceptions occur whilst the action is being handled, juju will gather any stdout/stderr data (and the return code) and inject them into the results object. Thus, the results object might contain the following keys, additionally to those specified by the charm code:

  • Stdout

  • Stderr

  • Stdout-encoding

  • Stderr-encoding

  • ReturnCode

Parameters:

results – The result of the action as a Dict

Raises:
  • ModelError – if a reserved key is used.

  • ValueError – if results has a mix of dotted/non-dotted keys that expand out to result in duplicate keys, for example: {'a': {'b': 1}, 'a.b': 2}. Also raised if a dict is passed with a key that fails to meet the format requirements.

  • OSError – if extremely large (>100KB) results are provided.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

class ops.ActionMeta(name: str, raw: Dict[str, Any] | None = None)[source]

Bases: object

Object containing metadata about an action’s definition.

class ops.ActiveStatus(message: str = '')[source]

Bases: StatusBase

The unit is ready.

The unit believes it is correctly offering all the services it has been asked to offer.

name = 'active'
class ops.Application(name: str, meta: CharmMeta, backend: _ModelBackend, cache: _ModelCache)[source]

Bases: object

Represents a named application in the model.

This might be this charm’s application, or might be an application this charm is integrated with. Charmers should not instantiate Application objects directly, but should use Model.app to get the application this unit is part of, or Model.get_app() if they need a reference to a given application.

add_secret(content: Dict[str, str], *, label: str | None = None, description: str | None = None, expire: datetime | timedelta | None = None, rotate: SecretRotate | None = None) Secret[source]

Create a Secret owned by this application.

Parameters:
  • content – A key-value mapping containing the payload of the secret, for example {"password": "foo123"}.

  • label – Charm-local label (or “name”) to assign to this secret, which can later be used for lookup.

  • description – Description of the secret’s purpose.

  • expire – Time in the future (or timedelta from now) at which the secret is due to expire. When that time elapses, Juju will notify the charm by sending a SecretExpired event. None (the default) means the secret will never expire.

  • rotate – Rotation policy/time. Every time this elapses, Juju will notify the charm by sending a SecretRotate event. None (the default) means to use the Juju default, which is never rotate.

Raises:

ValueError – if the secret is empty, or the secret key is invalid.

name: str

The name of this application (eg, ‘mysql’). This name may differ from the name of the charm, if the user has deployed it to a different name.

planned_units() int[source]

Get the number of units that Juju has “planned” for this application.

E.g., if an admin runs “juju deploy foo”, then “juju add-unit -n 2 foo”, the planned unit count for foo will be 3.

The data comes from the Juju agent, based on data it fetches from the controller. Pending units are included in the count, and scale down events may modify the count before some units have been fully torn down. The information in planned_units is up-to-date as of the start of the current hook invocation.

This method only returns data for this charm’s application – the Juju agent isn’t able to see planned unit counts for other applications in the model.

Raises:

RuntimeError – on trying to get the planned units for a remote application.

property status: StatusBase

Used to report or read the status of the overall application.

Changes to status take effect immediately, unlike other Juju operations such as modifying relation data or secrets, which only take effect after a successful event.

Can only be read and set by the lead unit of the application.

The status of remote units is always Unknown.

Alternatively, use the collect_app_status event to evaluate and set application status consistently at the end of every hook.

Raises:
  • RuntimeError – if setting the status of another application, or if setting the status of this application as a unit that is not the leader.

  • InvalidStatusError – if setting the status to something that is not a StatusBase

Example:

self.model.app.status = ops.BlockedStatus('I need a human to come help me')
class ops.Binding(name: str, relation_id: int | None, backend: _ModelBackend)[source]

Bases: object

Binding to a network space.

name: str

The name of the endpoint this binding represents (eg, ‘db’).

property network: Network

The network information for this binding.

class ops.BindingMapping(backend: _ModelBackend)[source]

Bases: Mapping[str, Binding]

Mapping of endpoints to network bindings.

Charm authors should not instantiate this directly, but access it via Model.get_binding()

get(binding_key: str | Relation) Binding[source]

Get a specific Binding for an endpoint/relation.

Not used directly by Charm authors. See Model.get_binding()

class ops.BlockedStatus(message: str = '')[source]

Bases: StatusBase

The unit requires manual intervention.

An admin has to manually intervene to unblock the unit and let it proceed.

name = 'blocked'
class ops.BoundEvent(emitter: Object, event_type: Type[EventBase], event_kind: str)[source]

Bases: object

Event bound to an Object.

emit(*args: Any, **kwargs: Any)[source]

Emit event to all registered observers.

The current storage state is committed before and after each observer is notified.

Note that the emission of custom events is handled immediately. In other words, custom events are not queued, but rather nested. For example:

1. Main hook handler (emits custom_event_1)
2.   Custom event 1 handler (emits custom_event_2)
3.     Custom event 2 handler
4.   Resume custom event 1 handler
5. Resume main hook handler
class ops.BoundStoredState(parent: Object, attr_name: str)[source]

Bases: object

Stored state data bound to a specific Object.

set_default(**kwargs: Any)[source]

Set the value of any given key if it has not already been set.

class ops.CharmBase(framework: Framework)[source]

Bases: Object

Base class that represents the charm overall.

CharmBase is used to create a charm. This is done by inheriting from CharmBase and customising the subclass as required. So to create a charm called MyCharm, define a charm class and set up the required event handlers (“hooks”) in its constructor:

import logging

import ops

def MyCharm(ops.CharmBase):
    def __init__(self, *args):
        super().__init__(*args)
        self.framework.observe(self.on.config_changed, self._on_config_changed)
        self.framework.observe(self.on.stop, self._on_stop)
        # ...

if __name__ == "__main__":
    ops.main(MyCharm)

As shown in the example above, a charm class is instantiated by ops.main rather than charm authors directly instantiating a charm.

Parameters:

framework – The framework responsible for managing the Model and events for this charm.

property app: Application

Application that this unit is part of.

property charm_dir: Path

Root directory of the charm as it is running.

property config: ConfigData

A mapping containing the charm’s config and current values.

property meta: CharmMeta

Metadata of this charm.

on: CharmEvents[source]

This property is used to create an event handler using Framework.observe(), and can be one of the events listed at CharmEvents.

property unit: Unit

Unit that this execution is responsible for.

class ops.CharmEvents(parent: Object | None = None, key: str | None = None)[source]

Bases: ObjectEvents

Events generated by Juju pertaining to application lifecycle.

By default, the events listed as attributes of this class will be provided via the CharmBase.on attribute. For example:

self.framework.observe(self.on.config_changed, self._on_config_changed)

In addition to the events listed as attributes of this class, dynamically-named events will also be defined based on the charm’s metadata (metadata.yaml) for relations, storage, actions, and containers. These named events may be accessed as self.on[<name>].<event> or using a prefix like self.on.<name>_<event>, for example:

self.framework.observe(self.on["db"].relation_created, self._on_db_relation_created)
self.framework.observe(self.on.workload_pebble_ready, self._on_workload_pebble_ready)
collect_app_status

Triggered on the leader at the end of every hook to collect app statuses for evaluation (see CollectStatusEvent).

collect_metrics

Triggered by Juju to collect metrics (see CollectMetricsEvent).

collect_unit_status

Triggered at the end of every hook to collect unit statuses for evaluation (see CollectStatusEvent).

config_changed

Triggered when a configuration change occurs (see ConfigChangedEvent).

install

Triggered when a charm is installed (see InstallEvent).

leader_elected

Triggered when a new leader has been elected (see LeaderElectedEvent).

leader_settings_changed

DEPRECATED. Triggered when leader changes any settings (see LeaderSettingsChangedEvent).

post_series_upgrade

Triggered after a series upgrade (see PostSeriesUpgradeEvent).

pre_series_upgrade

Triggered to prepare a unit for series upgrade (see PreSeriesUpgradeEvent).

remove

Triggered when a unit is about to be terminated (see RemoveEvent).

secret_changed

Triggered by Juju on the observer when the secret owner changes its contents (see SecretChangedEvent).

secret_expired

Triggered by Juju on the owner when a secret’s expiration time elapses (see SecretExpiredEvent).

secret_remove

Triggered by Juju on the owner when a secret revision can be removed (see SecretRemoveEvent).

secret_rotate

Triggered by Juju on the owner when the secret’s rotation policy elapses (see SecretRotateEvent).

start

Triggered immediately after first configuration change (see StartEvent).

stop

Triggered when a charm is shut down (see StopEvent).

update_status

Triggered periodically by a status update request from Juju (see UpdateStatusEvent).

upgrade_charm

Triggered by request to upgrade the charm (see UpgradeCharmEvent).

class ops.CharmMeta(raw: Dict[str, Any] | None = None, actions_raw: Dict[str, Any] | None = None)[source]

Bases: object

Object containing the metadata for the charm.

This is read from metadata.yaml and actions.yaml. Generally charms will define this information, rather than reading it at runtime. This class is mostly for the framework to understand what the charm has defined.

Parameters:
  • raw – a mapping containing the contents of metadata.yaml

  • actions_raw – a mapping containing the contents of actions.yaml

actions: Dict[str, ActionMeta]

Actions the charm has defined.

assumes: JujuAssumes

Juju features this charm requires.

containers: Dict[str, ContainerMeta]

Container metadata for each defined container.

description: str

Long description for this charm.

extra_bindings: Dict[str, None]

Additional named bindings that a charm can use for network configuration.

static from_charm_root(charm_root: Path | str)[source]

Initialise CharmMeta from the path to a charm repository root folder.

classmethod from_yaml(metadata: str | TextIO, actions: str | TextIO | None = None) CharmMeta[source]

Instantiate a CharmMeta from a YAML description of metadata.yaml.

Parameters:
  • metadata – A YAML description of charm metadata (name, relations, etc.) This can be a simple string, or a file-like object (passed to yaml.safe_load).

  • actions – YAML description of Actions for this charm (e.g., actions.yaml)

Links to more details about the charm.

maintainers: List[str]

List of email addresses of charm maintainers.

min_juju_version: str | None

Indicates the minimum Juju version this charm requires.

name: str

Name of this charm.

payloads: Dict[str, PayloadMeta]

Payload metadata for each defined payload.

peers: Dict[str, RelationMeta]

Peer relations.

provides: Dict[str, RelationMeta]

Relations this charm provides.

relations: Dict[str, RelationMeta]

All RelationMeta instances.

This is merged from requires, provides, and peers. If needed, the role of the relation definition can be obtained from its role attribute.

requires: Dict[str, RelationMeta]

Relations this charm requires.

resources: Dict[str, ResourceMeta]

Resource metadata for each defined resource.

series: List[str]

List of supported OS series that this charm can support.

The first entry in the list is the default series that will be used by deploy if no other series is requested by the user.

storages: Dict[str, StorageMeta]

Storage metadata for each defined storage.

subordinate: bool

Whether this charm is intended to be used as a subordinate charm.

summary: str

Short description of what this charm does.

tags: List[str]

Charmhub tag metadata for categories associated with this charm.

terms: List[str]

Charmhub terms that should be agreed to before this charm can be deployed.

class ops.CheckInfoMapping(checks: Iterable[CheckInfo])[source]

Bases: Mapping[str, CheckInfo]

Map of check names to ops.pebble.CheckInfo objects.

This is done as a mapping object rather than a plain dictionary so that we can extend it later, and so it’s not mutable.

class ops.CloudCredential(auth_type: str, attributes: ~typing.Dict[str, str] = <factory>, redacted: ~typing.List[str] = <factory>)[source]

Bases: object

Credentials for cloud.

Used as the type of attribute credential in CloudSpec.

attributes: Dict[str, str]

A dictionary containing cloud credentials.

For example, for AWS, it contains access-key and secret-key; for Azure, application-id, application-password and subscription-id can be found here.

auth_type: str

Authentication type.

classmethod from_dict(d: Dict[str, Any]) CloudCredential[source]

Create a new CloudCredential object from a dictionary.

redacted: List[str]

A list of redacted secrets.

class ops.CloudSpec(type: str, name: str, region: str | None = None, endpoint: str | None = None, identity_endpoint: str | None = None, storage_endpoint: str | None = None, credential: ~ops.model.CloudCredential | None = None, ca_certificates: ~typing.List[str] = <factory>, skip_tls_verify: bool = False, is_controller_cloud: bool = False)[source]

Bases: object

Cloud specification information (metadata) including credentials.

ca_certificates: List[str]

A list of CA certificates.

credential: CloudCredential | None = None

Cloud credentials with key-value attributes.

endpoint: str | None = None

Endpoint of the cloud.

classmethod from_dict(d: Dict[str, Any]) CloudSpec[source]

Create a new CloudSpec object from a dict parsed from JSON.

identity_endpoint: str | None = None

Identity endpoint of the cloud.

is_controller_cloud: bool = False

If this is the cloud used by the controller, defaults to False.

name: str

Juju cloud name.

region: str | None = None

Region of the cloud.

skip_tls_verify: bool = False

Whether to skip TLS verfication.

storage_endpoint: str | None = None

Storage endpoint of the cloud.

type: str

Type of the cloud.

class ops.CollectMetricsEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered by Juju to collect metrics.

Juju fires this event every five minutes for the lifetime of the unit. Callback methods bound to this event may use the add_metrics() method of this class to send measurements to Juju.

Note that associated callback methods are currently sandboxed in how they can interact with Juju.

add_metrics(metrics: Mapping[str, int | float], labels: Mapping[str, str] | None = None)[source]

Record metrics that have been gathered by the charm for this unit.

Parameters:
  • metrics – Key-value mapping of metrics that have been gathered.

  • labels – Key-value labels applied to the metrics.

Raises:

ModelError – if invalid keys or values are provided.

class ops.CollectStatusEvent(handle: Handle)[source]

Bases: LifecycleEvent

Event triggered at the end of every hook to collect statuses for evaluation.

If the charm wants to provide application or unit status in a consistent way after the end of every hook, it should observe the collect_app_status or collect_unit_status event, respectively.

The framework will trigger these events after the hook code runs successfully (collect_app_status will only be triggered on the leader unit). If any statuses were added by the event handler using add_status(), the framework will choose the highest-priority status and set that as the status (application status for collect_app_status, or unit status for collect_unit_status).

The order of priorities is as follows, from highest to lowest:

  • error

  • blocked

  • maintenance

  • waiting

  • active

  • unknown

If there are multiple statuses with the same priority, the first one added wins (and if an event is observed multiple times, the handlers are called in the order they were observed).

A collect-status event can be observed multiple times, and add_status() can be called multiple times to add multiple statuses for evaluation. This is useful when a charm has multiple components that each have a status. Each code path in a collect-status handler should call add_status at least once.

Below is an example “web app” charm component that observes collect_unit_status to provide the status of the component, which requires a “port” config option set before it can proceed:

class MyCharm(ops.CharmBase):
    def __init__(self, *args):
        super().__init__(*args)
        self.webapp = Webapp(self)
        # initialize other components

class WebApp(ops.Object):
    def __init__(self, charm: ops.CharmBase):
        super().__init__(charm, 'webapp')
        self.framework.observe(charm.on.collect_unit_status, self._on_collect_status)

    def _on_collect_status(self, event: ops.CollectStatusEvent):
        if 'port' not in self.model.config:
            event.add_status(ops.BlockedStatus('please set "port" config'))
            return
        event.add_status(ops.ActiveStatus())
add_status(status: StatusBase)[source]

Add a status for evaluation.

See CollectStatusEvent for a description of how to use this.

class ops.CommitEvent(handle: Handle)[source]

Bases: LifecycleEvent

Event that will be emitted second on commit.

class ops.ConfigChangedEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered when a configuration change occurs.

This event will fire in several situations:

  • When the admin reconfigures the charm using the Juju CLI, for example juju config mycharm foo=bar. This event notifies the charm of its new configuration. (The event itself, however, is not aware of what specifically has changed in the config).

  • Right after the unit starts up for the first time. This event notifies the charm of its initial configuration. Typically, this event will fire between an install and a start during the startup sequence (when a unit is first deployed), but in general it will fire whenever the unit is (re)started, for example after pod churn on Kubernetes, on unit rescheduling, on unit upgrade or refresh, and so on.

  • As a specific instance of the above point: when networking changes (if the machine reboots and comes up with a different IP).

Any callback method bound to this event cannot assume that the software has already been started; it should not start stopped software, but should (if appropriate) restart running software to take configuration changes into account.

class ops.ConfigData(backend: _ModelBackend)[source]

Bases: LazyMapping

Configuration data.

This class should not be instantiated directly. It should be accessed via Model.config.

class ops.Container(name: str, backend: _ModelBackend, pebble_client: Client | None = None)[source]

Bases: object

Represents a named container in a unit.

This class should not be instantiated directly, instead use Unit.get_container() or Unit.containers.

For methods that make changes to the container, if the change fails or times out, then a ops.pebble.ChangeError or ops.pebble.TimeoutError will be raised.

Interactions with the container use Pebble, so all methods may raise exceptions when there are problems communicating with Pebble. Problems connecting to or transferring data with Pebble will raise a ops.pebble.ConnectionError - you can guard against these by first checking can_connect(), but that generally introduces a race condition where problems occur after can_connect() has succeeded. When an error occurs executing the request, such as trying to add an invalid layer or execute a command that does not exist, an ops.pebble.APIError is raised.

add_layer(label: str, layer: str | LayerDict | Layer, *, combine: bool = False)[source]

Dynamically add a new layer onto the Pebble configuration layers.

Parameters:
  • label – Label for new layer (and label of layer to merge with if combining).

  • layer – A YAML string, configuration layer dict, or pebble.Layer object containing the Pebble layer to add.

  • combine – If combine is False (the default), append the new layer as the top layer with the given label (must be unique). If combine is True and the label already exists, the two layers are combined into a single one considering the layer override rules; if the layer doesn’t exist, it is added as usual.

autostart() None[source]

Autostart all services marked as startup: enabled.

can_connect() bool[source]

Report whether the Pebble API is reachable in the container.

This method returns a bool that indicates whether the Pebble API is available at the time the method is called. It does not guard against the Pebble API becoming unavailable, and should be treated as a “point in time” status only.

For example:

# Add status based on any earlier errors communicating with Pebble.
...
# Check that Pebble is still reachable now.
container = self.unit.get_container("example")
if not container.can_connect():
    event.add_status(ops.WaitingStatus("Waiting for Pebble..."))
exec(command: List[str], *, service_context: str | None = None, environment: Dict[str, str] | None = None, working_dir: str | None = None, timeout: float | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None, stdin: str | TextIO | None = None, stdout: TextIO | None = None, stderr: TextIO | None = None, encoding: str = 'utf-8', combine_stderr: bool = False) ExecProcess[str][source]
exec(command: List[str], *, service_context: str | None = None, environment: Dict[str, str] | None = None, working_dir: str | None = None, timeout: float | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None, stdin: bytes | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, encoding: None = None, combine_stderr: bool = False) ExecProcess[bytes]

Execute the given command on the remote system.

See ops.pebble.Client.exec() for documentation of the parameters and return value, as well as examples.

Note that older versions of Juju do not support the service_content parameter, so if the Charm is to be used on those versions, then JujuVersion.supports_exec_service_context() should be used as a guard.

Raises:

ExecError – if the command exits with a non-zero exit code.

exists(path: str | PurePath) bool[source]

Report whether a path exists on the container filesystem.

get_check(check_name: str) CheckInfo[source]

Get check information for a single named check.

Raises:

ModelError – if a check with the given name is not found

get_checks(*check_names: str, level: CheckLevel | None = None) CheckInfoMapping[source]

Fetch and return a mapping of check information indexed by check name.

Parameters:
  • check_names – Optional check names to query for. If no check names are specified, return checks with any name.

  • level – Optional check level to query for. If not specified, fetch all checks.

get_notice(id: str) Notice[source]

Get details about a single notice by ID.

Raises:

ModelError – if a notice with the given ID is not found

get_notices(*, users: NoticesUsers | None = None, user_id: int | None = None, types: Iterable[NoticeType | str] | None = None, keys: Iterable[str] | None = None) List[Notice][source]

Query for notices that match all of the provided filters.

See ops.pebble.Client.get_notices() for documentation of the parameters.

get_plan() Plan[source]

Get the combined Pebble configuration.

This will immediately reflect changes from any previous add_layer() calls, regardless of whether replan() or restart() have been called.

get_service(service_name: str) ServiceInfo[source]

Get status information for a single named service.

Raises:

ModelError – if a service with the given name is not found

get_services(*service_names: str) Mapping[str, ServiceInfo][source]

Fetch and return a mapping of status information indexed by service name.

If no service names are specified, return status information for all services, otherwise return information for only the given services.

isdir(path: str | PurePath) bool[source]

Report whether a directory exists at the given path on the container filesystem.

list_files(path: str | PurePath, *, pattern: str | None = None, itself: bool = False) List[FileInfo][source]

Return list of directory entries from given path on remote system.

Despite the name, this method returns a list of files and directories, similar to os.listdir() or os.scandir().

Parameters:
  • path – Path of the directory to list, or path of the file to return information about.

  • pattern – If specified, filter the list to just the files that match, for example *.txt.

  • itself – If path refers to a directory, return information about the directory itself, rather than its contents.

make_dir(path: str | PurePath, *, make_parents: bool = False, permissions: int | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None)[source]

Create a directory on the remote system with the given attributes.

Parameters:
  • path – Path of the directory to create on the remote system.

  • make_parents – If True, create parent directories if they don’t exist.

  • permissions – Permissions (mode) to create directory with (Pebble default is 0o755).

  • user_id – User ID (UID) for directory.

  • user – Username for directory. User’s UID must match user_id if both are specified.

  • group_id – Group ID (GID) for directory.

  • group – Group name for directory. Group’s GID must match group_id if both are specified.

name: str

The name of the container from metadata.yaml, for example “postgres”.

property pebble: Client

The low-level ops.pebble.Client instance for this container.

pull(path: str | PurePath, *, encoding: None) BinaryIO[source]
pull(path: str | PurePath, *, encoding: str = 'utf-8') TextIO

Read a file’s content from the remote system.

Parameters:
  • path – Path of the file to read from the remote system.

  • encoding – Encoding to use for decoding the file’s bytes to string, or None to specify no decoding.

Returns:

A readable file-like object, whose read() method will return strings decoded according to the specified encoding, or bytes if encoding is None.

Raises:

pebble.PathError – If there was an error reading the file at path, for example, if the file doesn’t exist or is a directory.

pull_path(source_path: str | PurePath | Iterable[str | PurePath], dest_dir: str | Path)[source]

Recursively pull a remote path or files to the local system.

Only regular files and directories are copied; symbolic links, device files, etc. are skipped. Pulling is attempted to completion even if errors occur during the process. All errors are collected incrementally. After copying has completed, if any errors occurred, a single MultiPushPullError is raised containing details for each error.

Assuming the following files exist remotely:

  • /foo/bar/baz.txt

  • /foo/foobar.txt

  • /quux.txt

These are various pull examples:

# copy one file
container.pull_path('/foo/foobar.txt', '/dst')
# Destination results: /dst/foobar.txt

# copy a directory
container.pull_path('/foo', '/dst')
# Destination results: /dst/foo/bar/baz.txt, /dst/foo/foobar.txt

# copy a directory's contents
container.pull_path('/foo/*', '/dst')
# Destination results: /dst/bar/baz.txt, /dst/foobar.txt

# copy multiple files
container.pull_path(['/foo/bar/baz.txt', 'quux.txt'], '/dst')
# Destination results: /dst/baz.txt, /dst/quux.txt

# copy a file and a directory
container.pull_path(['/foo/bar', '/quux.txt'], '/dst')
# Destination results: /dst/bar/baz.txt, /dst/quux.txt
Parameters:
  • source_path – A single path or list of paths to pull from the remote system. The paths can be either a file or a directory but must be absolute paths. If source_path is a directory, the directory base name is attached to the destination directory – that is, the source path directory is placed inside the destination directory. If a source path ends with a trailing /* it will have its contents placed inside the destination directory.

  • dest_dir – Local destination directory inside which the source dir/files will be placed.

push(path: str | PurePath, source: bytes | str | BinaryIO | TextIO, *, encoding: str = 'utf-8', make_dirs: bool = False, permissions: int | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None)[source]

Write content to a given file path on the remote system.

Note that if another process has the file open on the remote system, or if the remote file is a bind mount, pushing will fail with a pebble.PathError. Use Container.exec() for full control.

Parameters:
  • path – Path of the file to write to on the remote system.

  • source – Source of data to write. This is either a concrete str or bytes instance, or a readable file-like object.

  • encoding – Encoding to use for encoding source str to bytes, or strings read from source if it is a TextIO type. Ignored if source is bytes or BinaryIO.

  • make_dirs – If True, create parent directories if they don’t exist.

  • permissions – Permissions (mode) to create file with (Pebble default is 0o644).

  • user_id – User ID (UID) for file.

  • user – Username for file. User’s UID must match user_id if both are specified.

  • group_id – Group ID (GID) for file.

  • group – Group name for file. Group’s GID must match group_id if both are specified.

push_path(source_path: str | Path | Iterable[str | Path], dest_dir: str | PurePath)[source]

Recursively push a local path or files to the remote system.

Only regular files and directories are copied; symbolic links, device files, etc. are skipped. Pushing is attempted to completion even if errors occur during the process. All errors are collected incrementally. After copying has completed, if any errors occurred, a single MultiPushPullError is raised containing details for each error.

Assuming the following files exist locally:

  • /foo/bar/baz.txt

  • /foo/foobar.txt

  • /quux.txt

These are various push examples:

# copy one file
container.push_path('/foo/foobar.txt', '/dst')
# Destination results: /dst/foobar.txt

# copy a directory
container.push_path('/foo', '/dst')
# Destination results: /dst/foo/bar/baz.txt, /dst/foo/foobar.txt

# copy a directory's contents
container.push_path('/foo/*', '/dst')
# Destination results: /dst/bar/baz.txt, /dst/foobar.txt

# copy multiple files
container.push_path(['/foo/bar/baz.txt', 'quux.txt'], '/dst')
# Destination results: /dst/baz.txt, /dst/quux.txt

# copy a file and a directory
container.push_path(['/foo/bar', '/quux.txt'], '/dst')
# Destination results: /dst/bar/baz.txt, /dst/quux.txt
Parameters:
  • source_path – A single path or list of paths to push to the remote system. The paths can be either a file or a directory. If source_path is a directory, the directory base name is attached to the destination directory – that is, the source path directory is placed inside the destination directory. If a source path ends with a trailing /* it will have its contents placed inside the destination directory.

  • dest_dir – Remote destination directory inside which the source dir/files will be placed. This must be an absolute path.

remove_path(path: str | PurePath, *, recursive: bool = False)[source]

Remove a file or directory on the remote system.

Parameters:
  • path – Path of the file or directory to delete from the remote system.

  • recursive – If True, and path is a directory, recursively delete it and everything under it. If path is a file, delete the file. In either case, do nothing if the file or directory does not exist. Behaviourally similar to rm -rf <file|dir>.

Raises:

pebble.PathError – If a relative path is provided, or if recursive is False and the file or directory cannot be removed (it does not exist or is not empty).

replan() None[source]

Replan all services: restart changed services and start startup-enabled services.

restart(*service_names: str)[source]

Restart the given service(s) by name.

send_signal(sig: int | str, *service_names: str)[source]

Send the given signal to one or more services.

Parameters:
  • sig – Name or number of signal to send, for example "SIGHUP", 1, or signal.SIGHUP.

  • service_names – Name(s) of the service(s) to send the signal to.

Raises:

pebble.APIError – If any of the services are not in the plan or are not currently running.

start(*service_names: str)[source]

Start given service(s) by name.

stop(*service_names: str)[source]

Stop given service(s) by name.

class ops.ContainerBase(os_name: str, channel: str, architectures: List[str])[source]

Bases: object

Metadata to resolve a container image.

architectures: List[str]

List of architectures that this charm can run on.

channel: str

Channel of the OS in format track[/risk][/branch] as used by Snaps.

For example: 20.04/stable or 18.04/stable/fips

classmethod from_dict(d: _ContainerBaseDict) ContainerBase[source]

Create new ContainerBase object from dict parsed from YAML.

os_name: str

Name of the OS.

For example: ubuntu

class ops.ContainerMapping(names: Iterable[str], backend: _ModelBackend)[source]

Bases: Mapping[str, Container]

Map of container names to Container objects.

This is done as a mapping object rather than a plain dictionary so that we can extend it later, and so it’s not mutable.

class ops.ContainerMeta(name: str, raw: Dict[str, Any])[source]

Bases: object

Metadata about an individual container.

bases: List[ContainerBase] | None

List of bases for use in resolving a container image.

Sorted by descending order of preference, and must not be present if resource is specified.

property mounts: Dict[str, ContainerStorageMeta]

An accessor for the mounts in a container.

Dict keys match key name in StorageMeta

Example:

storage:
  foo:
    type: filesystem
    location: /test
containers:
  bar:
    mounts:
      - storage: foo
      - location: /test/mount
name: str

Name of the container (key in the YAML).

resource: str | None

Reference for an entry in the resources field.

Specifies the oci-image resource used to create the container. Must not be present if a base/channel is specified.

class ops.ContainerStorageMeta(storage: str, location: str)[source]

Bases: object

Metadata about storage for an individual container.

If multiple locations are specified for the same storage, such as Kubernetes subPath mounts, location will not be an accessible attribute, as it would not be possible to determine which mount point was desired, and locations should be iterated over.

add_location(location: str)[source]

Add an additional mount point to a known storage.

property location: str

The location the storage is mounted at.

Raises:

RuntimeError – if there is more than one mount point with the same backing storage - use locations instead.

property locations: List[str]

An accessor for the list of locations for a mount.

storage: str

Name for the mount point, which should exist in the keys of the charm’s StorageMeta.

class ops.ErrorStatus(message: str = '')[source]

Bases: StatusBase

The unit status is error.

The unit-agent has encountered an error (the application or unit requires human intervention in order to operate correctly).

This status is read-only; trying to set unit or application status to ErrorStatus will raise ModelError.

name = 'error'
class ops.EventBase(handle: Handle)[source]

Bases: object

The base class for all events.

Inherit this and override the snapshot and restore methods to create a custom event.

defer() None[source]

Defer the event to the future.

Deferring an event from a handler puts that handler into a queue, to be called again the next time the charm is invoked. This invocation may be the result of an action, or any event other than metric events. The queue of events will be dispatched before the new event is processed.

Important points that follow from the above:

  • defer() does not interrupt the execution of the current event handler. In almost all cases, a call to defer() should be followed by an explicit return from the handler;

  • the re-execution of the deferred event handler starts from the top of the handler method (not where defer was called);

  • only the handlers that actually called defer() are called again (that is: despite talking about “deferring an event” it is actually the handler/event combination that is deferred); and

  • any deferred events get processed before the event (or action) that caused the current invocation of the charm.

The general desire to call defer() happens when some precondition isn’t yet met. However, care should be exercised as to whether it is better to defer this event so that it is seen again, or whether it is better to just wait for the event that indicates the precondition has been met.

For example, if handling a config change requires that two config values are changed, there’s no reason to defer the first config-changed because there will be a second config-changed event fired when the other config value changes.

Similarly, if two events need to occur before execution can proceed (say event A and B), the event A handler could defer() because B has not been seen yet. However, that leads to:

  1. event A fires, calls defer()

  2. event B fires, event A handler is called first, still hasn’t seen B happen, so is deferred again. Then B happens, which progresses since it has seen A.

  3. At some future time, event C happens, which also checks if A can proceed.

framework: Framework = None

The Framework instance (set by the framework itself).

restore(snapshot: Dict[str, Any])[source]

Restore the value state from the given snapshot.

Subclasses must override to restore their custom state.

snapshot() Dict[str, Any][source]

Return the snapshot data that should be persisted.

Subclasses must override to save any custom state.

class ops.EventSource(event_type: Type[EventBase])[source]

Bases: object

EventSource wraps an event type with a descriptor to facilitate observing and emitting.

It is generally used as:

class SomethingHappened(ops.EventBase):
    pass

class SomeObject(Object):
    something_happened = ops.EventSource(SomethingHappened)

With that, instances of that type will offer the someobj.something_happened attribute which is a BoundEvent, and may be used to emit and observe the event.

class ops.Framework(storage: SQLiteStorage | JujuStorage, charm_dir: str | Path, meta: CharmMeta, model: Model, event_name: str | None = None)[source]

Bases: Object

Main interface from the Charm to the ops library’s infrastructure.

breakpoint(name: str | None = None)[source]

Add breakpoint, optionally named, at the place where this method is called.

For the breakpoint to be activated the JUJU_DEBUG_AT environment variable must be set to “all” or to the specific name parameter provided, if any. In every other situation calling this method does nothing.

The framework also provides a standard breakpoint named “hook”, that will stop execution when a hook event is about to be handled.

For those reasons, the “all” and “hook” breakpoint names are reserved.

Raises:

ValueError – if the breakpoint name is invalid.

charm_dir: Path = None

The charm project root directory.

close() None[source]

Close the underlying backends.

commit() None[source]

Save changes to the underlying backends.

drop_snapshot(handle: Handle)[source]

Discard a persistent snapshot.

load_snapshot(handle: Handle) Serializable[source]

Load a persistent snapshot.

meta: CharmMeta = None

The charm’s metadata.

model: Model = None

The Model instance for this charm.

observe(bound_event: BoundEvent, observer: Callable[[Any], None])[source]

Register observer to be called when bound_event is emitted.

If this is called multiple times for the same event type, the framework calls the observers in the order they were observed.

The bound_event is generally provided as an attribute of the object that emits the event, and is created in this style:

class SomeObject:
    something_happened = Event(SomethingHappened)

That event may be observed as:

framework.observe(someobj.something_happened, self._on_something_happened)
Raises:

RuntimeError – if bound_event or observer are the wrong type.

on[source]

Used for observe()-ing framework-specific events.

reemit() None[source]

Reemit previously deferred events to the observers that deferred them.

Only the specific observers that have previously deferred the event will be notified again. Observers that asked to be notified about events after it’s been first emitted won’t be notified, as that would mean potentially observing events out of order.

register_type(cls: Type[Serializable], parent: Handle | Object | None, kind: str | None = None)[source]

Register a type to a handle.

remove_unreferenced_events() None[source]

Remove events from storage that are not referenced.

In older versions of the framework, events that had no observers would get recorded but never deleted. This makes a best effort to find these events and remove them from the database.

save_snapshot(value: StoredStateData | EventBase)[source]

Save a persistent snapshot of the provided value.

set_breakpointhook() Any | None[source]

Hook into sys.breakpointhook so the builtin breakpoint() works as expected.

This method is called by main, and is not intended to be called by users of the framework itself outside of perhaps some testing scenarios.

The breakpoint() function is a Python >= 3.7 feature.

Returns:

The old value of sys.breakpointhook.

class ops.FrameworkEvents(parent: Object | None = None, key: str | None = None)[source]

Bases: ObjectEvents

Manager of all framework events.

commit

Triggered before event data is committed to storage.

pre_commit

Triggered before the commit event.

class ops.Handle(parent: Handle | Object | None, kind: str, key: str | None)[source]

Bases: object

Handle defines a name for an object in the form of a hierarchical path.

The provided parent is the object (or that object’s handle) that this handle sits under, or None if the object identified by this handle stands by itself as the root of its own hierarchy.

The handle kind is a string that defines a namespace so objects with the same parent and kind will have unique keys.

The handle key is a string uniquely identifying the object. No other objects under the same parent and kind may have the same key.

classmethod from_path(path: str) Handle[source]

Build a handle from the indicated path.

property key: str | None

Return the handle’s key.

property kind: str

Return the handle’s kind.

nest(kind: str, key: str | None) Handle[source]

Create a new handle as child of the current one.

property parent: Handle | None

Return own parent handle.

property path: str

Return the handle’s path.

class ops.HandleKind[source]

Bases: object

Helper descriptor to define the Object.handle_kind field.

The handle_kind for an object defaults to its type name, but it may be explicitly overridden if desired.

class ops.HookEvent(handle: Handle)[source]

Bases: EventBase

Events raised by Juju to progress a charm’s lifecycle.

Hooks are callback methods of a charm class (a subclass of CharmBase) that are invoked in response to events raised by Juju. These callback methods are the means by which a charm governs the lifecycle of its application.

The HookEvent class is the base of a type hierarchy of events related to the charm’s lifecycle.

HookEvent subtypes are grouped into the following categories

  • Core lifecycle events

  • Relation events

  • Storage events

  • Metric events

class ops.InstallEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered when a charm is installed.

This event is triggered at the beginning of a charm’s lifecycle. Any associated callback method should be used to perform one-time setup operations, such as installing prerequisite software.

exception ops.InvalidStatusError[source]

Bases: ModelError

Raised if trying to set an Application or Unit status to something invalid.

class ops.JujuAssumes(features: List[str | JujuAssumes], condition: JujuAssumesCondition = JujuAssumesCondition.ALL)[source]

Bases: object

Juju model features that are required by the charm.

See the Juju docs for a list of available features.

condition: JujuAssumesCondition = 'all-of'
features: List[str | JujuAssumes]
classmethod from_list(raw: List[Any], condition: JujuAssumesCondition = JujuAssumesCondition.ALL) JujuAssumes[source]

Create new JujuAssumes object from list parsed from YAML.

class ops.JujuAssumesCondition(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Distinguishes between JujuAssumes that must match all or any features.

ALL = 'all-of'

All features are required to satisfy the requirement.

ANY = 'any-of'

Any of the features satisfies the requirement.

class ops.JujuVersion(version: str)[source]

Bases: object

Helper to work with the Juju version.

It knows how to parse the JUJU_VERSION environment variable, and exposes different capabilities according to the specific version. It also allows users to compare JujuVersion instances with < and > operators.

classmethod from_environ() JujuVersion[source]

Build a version from the JUJU_VERSION environment variable.

has_app_data() bool[source]

Report whether this Juju version supports app data.

has_controller_storage() bool[source]

Report whether this Juju version supports controller-side storage.

property has_secrets: bool

Report whether this Juju version supports the “secrets” feature.

is_dispatch_aware() bool[source]

Report whether this Juju version supports dispatch.

property supports_exec_service_context: bool

Report whether this Juju version supports exec’s service_context option.

property supports_open_port_on_k8s: bool

Report whether this Juju version supports open-port on Kubernetes.

class ops.LazyMapping[source]

Bases: Mapping[str, str], ABC

Represents a dict that isn’t populated until it is accessed.

Charm authors should generally never need to use this directly, but it forms the basis for many of the dicts that the framework tracks.

class ops.LazyNotice(container: Container, id: str, type: str, key: str)[source]

Bases: object

Provide lazily-loaded access to a Pebble notice’s details.

The attributes provided by this class are the same as those of ops.pebble.Notice, however, the notice details are only fetched from Pebble if necessary (and cached on the instance).

expire_after: timedelta | None
first_occurred: datetime
id: str
key: str
last_data: Dict[str, str]
last_occurred: datetime
last_repeated: datetime
occurrences: int
repeat_after: timedelta | None
type: NoticeType | str
user_id: int | None
class ops.LeaderElectedEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered when a new leader has been elected.

Juju will trigger this event when a new leader unit is chosen for a given application.

This event fires at least once after Juju selects a leader unit. Callback methods bound to this event may take any action required for the elected unit to assert leadership. Note that only the elected leader unit will receive this event.

class ops.LeaderSettingsChangedEvent(handle: Handle)[source]

Bases: HookEvent

DEPRECATED. Event triggered when leader changes any settings.

This event has been deprecated in favor of using a Peer relation, and having the leader set a value in the Application data bag for that peer relation. (see RelationChangedEvent).

class ops.LifecycleEvent(handle: Handle)[source]

Bases: EventBase

Events tied to the lifecycle of the Framework object.

defer() NoReturn[source]

Lifecycle events are not deferrable like other events.

This is because these events are run alongside each event invocation, so deferring would always end up simply doubling the work.

Raises:

RuntimeError – always.

class ops.MaintenanceStatus(message: str = '')[source]

Bases: StatusBase

The unit is performing maintenance tasks.

The unit is not yet providing services, but is actively doing work in preparation for providing those services. This is a “spinning” state, not an error state. It reflects activity on the unit itself, not on peers or related units.

name = 'maintenance'

Bases: object

Links to additional information about a charm.

documentation: str | None

Link to charm documentation.

issues: List[str]

List of links to the charm issue tracker.

sources: List[str]

List of links to the charm source code.

websites: List[str]

List of links to project websites.

class ops.Model(meta: CharmMeta, backend: _ModelBackend, broken_relation_id: int | None = None)[source]

Bases: object

Represents the Juju Model as seen from this unit.

This should not be instantiated directly by Charmers, but can be accessed as self.model from any class that derives from Object.

property app: Application

The application this unit is a part of.

Use get_app() to get an arbitrary application by name.

property config: ConfigData

Return a mapping of config for the current application.

get_app(app_name: str) Application[source]

Get an application by name.

Use app to get this charm’s application.

Internally this uses a cache, so asking for the same application two times will return the same object.

get_binding(binding_key: str | Relation) Binding | None[source]

Get a network space binding.

Parameters:

binding_key – The relation name or instance to obtain bindings for.

Returns:

If binding_key is a relation name, the method returns the default binding for that relation. If a relation instance is provided, the method first looks up a more specific binding for that specific relation ID, and if none is found falls back to the default binding for the relation name.

get_cloud_spec() CloudSpec[source]

Get details of the cloud in which the model is deployed.

Note: This information is only available for machine charms, not Kubernetes sidecar charms.

Returns:

a specification for the cloud in which the model is deployed, including credential information.

Raises:

ModelError – if called in a Kubernetes model.

get_relation(relation_name: str, relation_id: int | None = None) Relation | None[source]

Get a specific Relation instance.

If relation_id is not given, this will return the Relation instance if the relation is established only once or None if it is not established. If this same relation is established multiple times the error TooManyRelatedAppsError is raised.

Parameters:
  • relation_name – The name of the endpoint for this charm

  • relation_id – An identifier for a specific relation. Used to disambiguate when a given application has more than one relation on a given endpoint.

Raises:

TooManyRelatedAppsError – is raised if there is more than one integration with the supplied relation_name and no relation_id was supplied

get_secret(*, id: str | None = None, label: str | None = None) Secret[source]

Get the Secret with the given ID or label.

The caller must provide at least one of id (the secret’s locator ID) or label (the charm-local “name”).

If both are provided, the secret will be fetched by ID, and the secret’s label will be updated to the label provided. Normally secret owners set a label using add_secret, whereas secret observers set a label using get_secret (see an example at Secret.label).

Parameters:
  • id – Secret ID if fetching by ID.

  • label – Secret label if fetching by label (or updating it).

Raises:

SecretNotFoundError – If a secret with this ID or label doesn’t exist.

get_unit(unit_name: str) Unit[source]

Get an arbitrary unit by name.

Use unit to get the current unit.

Internally this uses a cache, so asking for the same unit two times will return the same object.

property name: str

Return the name of the Model that this unit is running in.

This is read from the environment variable JUJU_MODEL_NAME.

property pod: Pod

Represents the definition of a pod spec in legacy Kubernetes models.

DEPRECATED: New charms should use the sidecar pattern with Pebble.

Use Pod.set_spec() to set the container specification for legacy Kubernetes charms.

property relations: RelationMapping

Mapping of endpoint to list of Relation.

Answers the question “what am I currently integrated with”. See also get_relation().

In a relation-broken event, the broken relation is excluded from this list.

property resources: Resources

Access to resources for this charm.

Use model.resources.fetch(resource_name) to get the path on disk where the resource can be found.

property storages: StorageMapping

Mapping of storage_name to Storage as defined in metadata.yaml.

property unit: Unit

The unit that is running this code.

Use get_unit() to get an arbitrary unit by name.

property uuid: str

Return the identifier of the Model that this unit is running in.

This is read from the environment variable JUJU_MODEL_UUID.

exception ops.ModelError[source]

Bases: Exception

Base class for exceptions raised when interacting with the Model.

exception ops.MultiPushPullError(message: str, errors: List[Tuple[str, Exception]])[source]

Bases: Exception

Aggregates multiple push and pull exceptions into one.

This class should not be instantiated directly. It is raised by Container.push_path() and Container.pull_path().

errors: List[Tuple[str, Exception]]

The list of errors.

Each error is represented by a tuple of (<source_path>, <exception>), where source_path is the path being pushed to or pulled from.

message: str

The error message.

class ops.Network(network_info: _NetworkDict)[source]

Bases: object

Network space details.

Charm authors should not instantiate this directly, but should get access to the Network definition from Model.get_binding() and its network attribute.

property bind_address: IPv4Address | IPv6Address | str | None

A single address that the charm’s application should bind() to.

For the common case where there is a single answer. This represents a single address from interfaces that can be used to configure where the charm’s application should bind() and listen().

egress_subnets: List[IPv4Network | IPv6Network]

A list of networks representing the subnets that other units will see the charm connecting from. Due to things like NAT it isn’t always possible to narrow it down to a single address, but when it is clear, the CIDRs will be constrained to a single address (for example, 10.0.0.1/32).

property ingress_address: IPv4Address | IPv6Address | str | None

The address other applications should use to connect to the current unit.

Due to things like public/private addresses, NAT and tunneling, the address the charm will bind() to is not always the address other people can use to connect() to the charm. This is just the first address from ingress_addresses.

ingress_addresses: List[IPv4Address | IPv6Address | str]

A list of IP addresses that other units should use to get in touch with the charm.

interfaces: List[NetworkInterface]

A list of network interface details. This includes the information about how the application should be configured (for example, what IP addresses should be bound to).

Multiple addresses for a single interface are represented as multiple interfaces, for example:

[NetworkInfo('ens1', '10.1.1.1/32'), NetworkInfo('ens1', '10.1.2.1/32'])
class ops.NetworkInterface(name: str, address_info: _AddressDict)[source]

Bases: object

Represents a single network interface that the charm needs to know about.

Charmers should not instantiate this type directly. Instead use Model.get_binding() to get the network information for a given endpoint.

address: IPv4Address | IPv6Address | str | None

The address of the network interface.

name: str

The name of the interface (for example, ‘eth0’ or ‘ens1’).

subnet: IPv4Network | IPv6Network | None

The subnet of the network interface. This may be a single address (for example, ‘10.0.1.2/32’).

exception ops.NoTypeError(handle_path: str)[source]

Bases: Exception

No class to hold it was found when restoring an event.

class ops.Object(parent: Framework | Object, key: str | None)[source]

Bases: object

Initialize an Object as a new leaf in Framework, identified by key.

Parameters:
  • parent – parent node in the tree.

  • key – unique identifier for this object.

Every object belongs to exactly one framework.

Every object has a parent, which might be a framework.

We track a “path to object,” which is the path to the parent, plus the object’s unique identifier. Event handlers use this identity to track the destination of their events, and the Framework uses this id to track persisted state between event executions.

The Framework should raise an error if it ever detects that two objects with the same id have been created.

handle_kind: str

Helper descriptor to define the Object.handle_kind field.

The handle_kind for an object defaults to its type name, but it may be explicitly overridden if desired.

property model: Model

Shortcut for more simple access the model.

class ops.ObjectEvents(parent: Object | None = None, key: str | None = None)[source]

Bases: Object

Convenience type to allow defining .on attributes at class level.

classmethod define_event(event_kind: str, event_type: Type[EventBase])[source]

Define an event on this type at runtime.

Note that attempting to define the same event kind more than once will raise an “overlaps with existing type” runtime error. Ops uses a labeling system to track and reconstruct events between hook executions (each time a hook runs, the Juju Agent invokes a fresh instance of ops; there is no ops process that persists on the host between hooks). Having duplicate Python objects creates duplicate labels. Overwriting a previously created label means that only the latter code path will be run when the current event, if it does get deferred, is re-emitted. This is usually not what is desired, and is error-prone and ambiguous.

Parameters:
  • event_kind – An attribute name that will be used to access the event. Must be a valid Python identifier, not be a keyword or an existing attribute.

  • event_type – A type of the event to define.

Raises:

RuntimeError – if the same event is defined twice, or if event_kind is an invalid name.

events() Dict[str, EventSource][source]

Return a mapping of event_kinds to bound_events for all available events.

handle_kind: str = 'on'
ops.OpenedPort

alias of Port

class ops.PayloadMeta(name: str, raw: Dict[str, Any])[source]

Bases: object

Object containing metadata about a payload definition.

payload_name: str

Name of the payload.

type: str

Payload type.

class ops.PebbleCustomNoticeEvent(handle: Handle, workload: Container, notice_id: str, notice_type: str, notice_key: str)[source]

Bases: PebbleNoticeEvent

Event triggered when a Pebble notice of type “custom” is created or repeats.

class ops.PebbleNoticeEvent(handle: Handle, workload: Container, notice_id: str, notice_type: str, notice_key: str)[source]

Bases: WorkloadEvent

Base class for Pebble notice events (each notice type is a subclass).

notice: LazyNotice

Provide access to the event notice’s details.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

class ops.PebbleReadyEvent(handle: Handle, workload: Container)[source]

Bases: WorkloadEvent

Event triggered when Pebble is ready for a workload.

This event is triggered when the Pebble process for a workload/container starts up, allowing the charm to configure how services should be launched.

Callback methods bound to this event allow the charm to run code after a workload has started its Pebble instance and is ready to receive instructions regarding what services should be started. The name prefix of the hook will depend on the container key defined in the metadata.yaml file.

class ops.Pod(backend: _ModelBackend)[source]

Bases: object

Represents the definition of a pod spec in legacy Kubernetes models.

DEPRECATED: New charms should use the sidecar pattern with Pebble.

Currently only supports simple access to setting the Juju pod spec via set_spec.

set_spec(spec: Mapping[str, Any], k8s_resources: Mapping[str, Any] | None = None)[source]

Set the specification for pods that Juju should start in kubernetes.

See juju help-tool pod-spec-set for details of what should be passed.

Parameters:
  • spec – The mapping defining the pod specification

  • k8s_resources – Additional kubernetes specific specification.

class ops.Port(protocol: Literal['tcp', 'udp', 'icmp'], port: int | None)[source]

Bases: object

Represents a port opened by Unit.open_port() or Unit.set_ports().

port: int | None

The port number. Will be None if protocol is 'icmp'.

protocol: Literal['tcp', 'udp', 'icmp']

The IP protocol.

class ops.PostSeriesUpgradeEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered after a series upgrade.

This event is triggered after the administrator has done a distribution upgrade (or rolled back and kept the same series). It is called in response to juju upgrade-machine <machine> complete. Associated charm callback methods are expected to do whatever steps are necessary to reconfigure their applications for the new series. This may include things like populating the upgraded version of a database. Note however charms are expected to check if the series has actually changed or whether it was rolled back to the original series.

class ops.PreCommitEvent(handle: Handle)[source]

Bases: LifecycleEvent

Event that will be emitted first on commit.

class ops.PreSeriesUpgradeEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered to prepare a unit for series upgrade.

This event triggers when an administrator executes juju upgrade-machine <machine> prepare. The event will fire for each unit that is running on the specified machine. Any callback method bound to this event must prepare the charm for an upgrade to the series. This may include things like exporting database content to a version neutral format, or evacuating running instances to other machines.

It can be assumed that only after all units on a machine have executed the callback method associated with this event, the administrator will initiate steps to actually upgrade the series. After the upgrade has been completed, the PostSeriesUpgradeEvent will fire.

class ops.PrefixedEvents(emitter: Object, key: str)[source]

Bases: object

Events to be found in all events using a specific prefix.

class ops.Relation(relation_name: str, relation_id: int, is_peer: bool, our_unit: Unit, backend: _ModelBackend, cache: _ModelCache, active: bool = True)[source]

Bases: object

Represents an established relation between this application and another application.

This class should not be instantiated directly, instead use Model.get_relation(), Model.relations, or ops.RelationEvent.relation. This is principally used by ops.RelationMeta to represent the relationships between charms.

active: bool

Indicates whether this relation is active.

This is normally True; it will be False if the current event is a relation-broken event associated with this relation.

app: Application

Represents the remote application of this relation.

For peer relations, this will be the local application.

data: RelationData

Holds the data buckets for each entity of a relation.

This is accessed using, for example, Relation.data[unit]['foo'].

id: int

The identifier for a particular relation.

name: str

The name of the local endpoint of the relation (for example, ‘db’).

units: Set[Unit]

A set of units that have started and joined this relation.

For subordinate relations, this set will include only one unit: the principal unit.

class ops.RelationBrokenEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None)[source]

Bases: RelationEvent

Event triggered when a relation is removed.

If a relation is being removed (juju remove-relation or juju remove-application), once all the units have been removed, this event will fire to signal that the relationship has been fully terminated.

The event indicates that the current relation is no longer valid, and that the charm’s software must be configured as though the relation had never existed. It will only be called after every callback method bound to RelationDepartedEvent has been run. If a callback method bound to this event is being executed, it is guaranteed that no remote units are currently known locally.

unit: None

Always None.

class ops.RelationChangedEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None)[source]

Bases: RelationEvent

Event triggered when relation data changes.

This event is triggered whenever there is a change to the data bucket for a related application or unit. Look at event.relation.data[event.unit/app] to see the new information, where event is the event object passed to the callback method bound to this event.

This event always fires once, after RelationJoinedEvent, and will subsequently fire whenever that remote unit changes its data for the relation. Callback methods bound to this event should be the only ones that rely on remote relation data. They should not error if the data is incomplete, since it can be guaranteed that when the remote unit or application changes its data, the event will fire again.

The data that may be queried, or set, are determined by the relation’s interface.

class ops.RelationCreatedEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None)[source]

Bases: RelationEvent

Event triggered when a new relation is created.

This is triggered when a new integration with another app is added in Juju. This can occur before units for those applications have started. All existing relations will trigger RelationCreatedEvent before StartEvent is emitted.

unit: None

Always None.

class ops.RelationData(relation: Relation, our_unit: Unit, backend: _ModelBackend)[source]

Bases: Mapping[Union[Unit, Application], RelationDataContent]

Represents the various data buckets of a given relation.

Each unit and application involved in a relation has their own data bucket. For example, {entity: RelationDataContent}, where entity can be either a Unit or an Application.

Units can read and write their own data, and if they are the leader, they can read and write their application data. They are allowed to read remote unit and application data.

This class should not be instantiated directly, instead use Relation.data

exception ops.RelationDataAccessError[source]

Bases: RelationDataError

Raised by Relation.data[entity][key] = value if unable to access.

This typically means that permission to write read/write the databag is missing, but in some cases it is raised when attempting to read/write from a deceased remote entity.

class ops.RelationDataContent(relation: Relation, entity: Unit | Application, backend: _ModelBackend)[source]

Bases: LazyMapping, MutableMapping[str, str]

Data content of a unit or application in a relation.

exception ops.RelationDataError[source]

Bases: ModelError

Raised when a relation data read/write is invalid.

This is raised either when trying to set a value to something that isn’t a string, or when trying to set a value in a bucket without the required access. (For example, another application/unit, or setting application data without being the leader.)

exception ops.RelationDataTypeError[source]

Bases: RelationDataError

Raised by Relation.data[entity][key] = value if key or value are not strings.

class ops.RelationDepartedEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None, departing_unit_name: str | None = None)[source]

Bases: RelationEvent

Event triggered when a unit leaves a relation.

This is the inverse of the RelationJoinedEvent, representing when a unit is leaving the relation (the unit is being removed, the app is being removed, the relation is being removed). For remaining units, this event is emitted once for each departing unit. For departing units, this event is emitted once for each remaining unit.

Callback methods bound to this event may be used to remove all references to the departing remote unit, because there’s no guarantee that it’s still part of the system; it’s perfectly probable (although not guaranteed) that the system running that unit has already shut down.

Once all callback methods bound to this event have been run for such a relation, the unit agent will fire the RelationBrokenEvent.

property departing_unit: Unit | None

The ops.Unit that is departing, if any.

Use this method to determine (for example) whether this unit is the departing one.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

unit: Unit

The remote unit that has triggered this event.

class ops.RelationEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None)[source]

Bases: HookEvent

A base class representing the various relation lifecycle events.

Relation lifecycle events are generated when application units participate in relations. Units can only participate in relations after they have been “started”, and before they have been “stopped”. Within that time window, the unit may participate in several different relations at a time, including multiple relations with the same name.

app: Application

The remote application that has triggered this event.

relation: Relation

The relation involved in this event.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

unit: Unit | None

The remote unit that has triggered this event.

This will be None if the relation event was triggered as an Application-level event.

class ops.RelationJoinedEvent(handle: Handle, relation: Relation, app: Application | None = None, unit: Unit | None = None)[source]

Bases: RelationEvent

Event triggered when a new unit joins a relation.

This event is triggered whenever a new unit of a related application joins the relation. The event fires only when that remote unit is first observed by the unit. Callback methods bound to this event may set any local unit data that can be determined using no more than the name of the joining unit and the remote private-address setting, which is always available when the relation is created and is by convention not deleted.

unit: Unit

The remote unit that has triggered this event.

class ops.RelationMapping(relations_meta: Dict[str, RelationMeta], our_unit: Unit, backend: _ModelBackend, cache: _ModelCache, broken_relation_id: int | None)[source]

Bases: Mapping[str, List[Relation]]

Map of relation names to lists of Relation instances.

class ops.RelationMeta(role: RelationRole, relation_name: str, raw: _RelationMetaDict)[source]

Bases: object

Object containing metadata about a relation definition.

Should not be constructed directly by charm code, but gotten from one of CharmMeta.peers, CharmMeta.requires, CharmMeta.provides, or CharmMeta.relations.

VALID_SCOPES = ['global', 'container']
interface_name: str | None

Definition of the interface protocol.

limit: int | None

Maximum number of connections to this relation endpoint.

optional: bool

If True, the relation is considered optional.

This value is informational only and is not used by Juju itself (all relations are optional from Juju’s perspective), but it may be set in metadata.yaml and used by the charm code if appropriate.

relation_name: str

Name of this relation.

role: RelationRole

Role this relation takes, one of ‘peer’, ‘requires’, or ‘provides’.

scope: str

Scope based on how this relation should be used.

Will be either "global" or "container".

exception ops.RelationNotFoundError[source]

Bases: ModelError

Raised when querying Juju for a given relation and that relation doesn’t exist.

class ops.RelationRole(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

An annotation for a charm’s role in a relation.

For each relation a charm’s role may be

  • A Peer

  • A service consumer in the relation (‘requires’)

  • A service provider in the relation (‘provides’)

is_peer() bool[source]

Report whether this role is ‘peer’.

role.is_peer() is a shortcut for role == ops.RelationRole.peer.

peer = 'peer'
provides = 'provides'
requires = 'requires'
class ops.RemoveEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered when a unit is about to be terminated.

This event fires prior to Juju removing the charm and terminating its unit.

defer() NoReturn[source]

Remove events are not deferrable like other events.

This is because the unit is about to be torn down, and there will not be an opportunity for the deferred event to run.

Raises:

RuntimeError – always.

class ops.ResourceMeta(name: str, raw: _ResourceMetaDict)[source]

Bases: object

Object containing metadata about a resource definition.

description: str

A description of the resource.

This will be empty string (rather than None) if not set in metadata.yaml.

filename: str | None

Filename of the resource file.

resource_name: str

Name of the resource.

type: str

Type of the resource. One of "file" or "oci-image".

class ops.Resources(names: Iterable[str], backend: _ModelBackend)[source]

Bases: object

Object representing resources for the charm.

fetch(name: str) Path[source]

Fetch the resource from the controller or store.

If successfully fetched, this returns the path where the resource is stored on disk, otherwise it raises a NameError.

Raises:

NameError – if the resource’s path cannot be fetched.

class ops.Secret(backend: _ModelBackend, id: str | None = None, label: str | None = None, content: Dict[str, str] | None = None)[source]

Bases: object

Represents a single secret in the model.

This class should not be instantiated directly, instead use Model.get_secret() (for observers and owners), or Application.add_secret() or Unit.add_secret() (for owners).

All secret events have a .secret attribute which provides the Secret associated with that event.

get_content(*, refresh: bool = False) Dict[str, str][source]

Get the secret’s content.

Returns:

A copy of the secret’s content dictionary.

Parameters:

refresh – If true, fetch the latest revision’s content and tell Juju to update to tracking that revision. The default is to get the content of the currently-tracked revision.

get_info() SecretInfo[source]

Get this secret’s information (metadata).

Only secret owners can fetch this information.

grant(relation: Relation, *, unit: Unit | None = None)[source]

Grant read access to this secret.

If the application or unit has already been granted access to this secret, do nothing.

Parameters:
  • relation – The relation used to scope the life of this secret.

  • unit – If specified, grant access to only this unit, rather than all units in the application.

property id: str | None

Locator ID (URI) for this secret.

This has an unfortunate name for historical reasons, as it’s not really a unique identifier, but the secret’s locator URI, which may or may not include the model UUID (for cross-model secrets).

Charms should treat this as an opaque string for looking up secrets and sharing them via relation data. If a charm-local “name” is needed for a secret, use a label. (If a charm needs a truly unique identifier for identifying one secret in a set of secrets of arbitrary size, use unique_identifier – this should be rare.)

This will be None if the secret was obtained using Model.get_secret() with a label but no ID.

property label: str | None

Label used to reference this secret locally.

This label is effectively a name for the secret that’s local to the charm, for example “db-password” or “tls-cert”. The secret owner sets a label with Application.add_secret() or Unit.add_secret(), and the secret observer sets a label with a call to Model.get_secret().

The label property can be used distinguish between multiple secrets in event handlers like ops.SecretChangedEvent. For example, if a charm is observing two secrets, it might call model.get_secret(id=secret_id, label='db-password') and model.get_secret(id=secret_id, label='tls-cert') in the relevant relation-changed event handlers, and then switch on event.secret.label in secret-changed:

def _on_secret_changed(self, event):
    if event.secret.label == 'db-password':
        content = event.secret.get_content(refresh=True)
        self._configure_db_credentials(content['username'], content['password'])
    elif event.secret.label == 'tls-cert':
        content = event.secret.get_content(refresh=True)
        self._update_tls_cert(content['cert'])
    else:
        pass  # ignore other labels (or log a warning)

Juju will ensure that the entity (the owner or observer) only has one secret with this label at once.

This will be None if the secret was obtained using Model.get_secret() with an ID but no label.

peek_content() Dict[str, str][source]

Get the content of the latest revision of this secret.

This returns the content of the latest revision without updating the tracking.

remove_all_revisions() None[source]

Remove all revisions of this secret.

This is called when the secret is no longer needed, for example when handling ops.RelationBrokenEvent.

remove_revision(revision: int)[source]

Remove the given secret revision.

This is normally called when handling ops.SecretRemoveEvent or ops.SecretExpiredEvent.

Parameters:

revision – The secret revision to remove. If being called from a secret event, this should usually be set to SecretRemoveEvent.revision.

revoke(relation: Relation, *, unit: Unit | None = None)[source]

Revoke read access to this secret.

If the application or unit does not have access to this secret, do nothing.

Parameters:
  • relation – The relation used to scope the life of this secret.

  • unit – If specified, revoke access to only this unit, rather than all units in the application.

set_content(content: Dict[str, str])[source]

Update the content of this secret.

This will create a new secret revision, and notify all units tracking the secret (the “observers”) that a new revision is available with a ops.SecretChangedEvent.

Parameters:

content – A key-value mapping containing the payload of the secret, for example {"password": "foo123"}.

set_info(*, label: str | None = None, description: str | None = None, expire: datetime | timedelta | None = None, rotate: SecretRotate | None = None)[source]

Update this secret’s information (metadata).

This will not create a new secret revision (that applies only to set_content()). Once attributes are set, they cannot be unset.

Parameters:
  • label – New label to apply.

  • description – New description to apply.

  • expire – New expiration time (or timedelta from now) to apply.

  • rotate – New rotation policy to apply. The new policy will take effect only after the currently-scheduled rotation.

property unique_identifier: str | None

Unique identifier of this secret.

This is the secret’s globally-unique identifier (currently a 20-character Xid, for example “9m4e2mr0ui3e8a215n4g”).

Charms should use id (the secret’s locator ID) to send the secret’s ID across relation data, and labels (label) to assign a charm-local “name” to the secret for lookup in this charm. However, unique_identifier can be useful to distinguish secrets in cases where the charm has a set of secrets of arbitrary size, for example, a group of 10 or 20 TLS certificates.

This will be None if the secret was obtained using Model.get_secret() with a label but no ID.

class ops.SecretChangedEvent(handle: Handle, id: str, label: str | None)[source]

Bases: SecretEvent

Event triggered on the secret observer charm when the secret owner changes its contents.

When the owner of a secret changes the secret’s contents, Juju will create a new secret revision, and all applications or units that are tracking this secret will be notified via this event that a new revision is available.

Typically, the charm will fetch the new content by calling event.secret.get_content() with refresh=True to tell Juju to start tracking the new revision.

class ops.SecretEvent(handle: Handle, id: str, label: str | None)[source]

Bases: HookEvent

Base class for all secret events.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

property secret: Secret

The secret instance this event refers to.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

class ops.SecretExpiredEvent(handle: Handle, id: str, label: str | None, revision: int)[source]

Bases: SecretEvent

Event triggered on the secret owner charm when a secret’s expiration time elapses.

This event is fired on the secret owner to inform it that the secret revision must be removed. The event will keep firing until the owner removes the revision by calling event.secret.remove_revision().

defer() NoReturn[source]

Secret expiration events are not deferrable (Juju handles re-invocation).

Raises:

RuntimeError – always.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

property revision: int

The secret revision this event refers to.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

class ops.SecretInfo(id: str, label: str | None, revision: int, expires: datetime | None, rotation: SecretRotate | None, rotates: datetime | None)[source]

Bases: object

Secret information (metadata).

classmethod from_dict(id: str, d: Dict[str, Any]) SecretInfo[source]

Create new SecretInfo object from ID and dict parsed from JSON.

exception ops.SecretNotFoundError[source]

Bases: ModelError

Raised when the specified secret does not exist.

class ops.SecretRemoveEvent(handle: Handle, id: str, label: str | None, revision: int)[source]

Bases: SecretEvent

Event triggered on the secret owner charm when a secret revision can be removed.

When the owner of a secret creates a new revision, and after all observers have updated to that new revision, this event will be fired to inform the secret owner that the old revision can be removed.

Typically, the charm will call event.secret.remove_revision() to remove the now-unused revision.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

property revision: int

The secret revision this event refers to.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

class ops.SecretRotate(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Secret rotation policies.

DAILY = 'daily'
HOURLY = 'hourly'
MONTHLY = 'monthly'
NEVER = 'never'
QUARTERLY = 'quarterly'
WEEKLY = 'weekly'
YEARLY = 'yearly'
class ops.SecretRotateEvent(handle: Handle, id: str, label: str | None)[source]

Bases: SecretEvent

Event triggered on the secret owner charm when the secret’s rotation policy elapses.

This event is fired on the secret owner to inform it that the secret must be rotated. The event will keep firing until the owner creates a new revision by calling event.secret.set_content().

defer() NoReturn[source]

Secret rotation events are not deferrable (Juju handles re-invocation).

Raises:

RuntimeError – always.

class ops.Serializable(*args, **kwargs)[source]

Bases: Protocol

The type returned by Framework.load_snapshot().

property handle: Handle
handle_kind = ''
restore(snapshot: Dict[str, Any]) None[source]
snapshot() Dict[str, Any][source]
class ops.ServiceInfoMapping(services: Iterable[ServiceInfo])[source]

Bases: Mapping[str, ServiceInfo]

Map of service names to pebble.ServiceInfo objects.

This is done as a mapping object rather than a plain dictionary so that we can extend it later, and so it’s not mutable.

class ops.StartEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered immediately after first configuration change.

This event is triggered immediately after the first ConfigChangedEvent. Callback methods bound to the event should be used to ensure that the charm’s software is in a running state. Note that the charm’s software should be configured so as to persist in this state through reboots without further intervention on Juju’s part.

class ops.StatusBase(message: str = '')[source]

Bases: object

Status values specific to applications and units.

To access a status by name, use StatusBase.from_name(). However, most use cases will directly use the child class such as ActiveStatus to indicate their status.

classmethod from_name(name: str, message: str)[source]

Create a status instance from a name and message.

If name is “unknown”, message is ignored, because unknown status does not have an associated message.

Parameters:
  • name – Name of the status, for example “active” or “blocked”.

  • message – Message to include with the status.

Raises:

KeyError – If name is not a registered status.

name = ''
classmethod register(child: Type[StatusBase])[source]

Register a Status for the child’s name.

class ops.StopEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered when a charm is shut down.

This event is triggered when an application’s removal is requested by the client. The event fires immediately before the end of the unit’s destruction sequence. Callback methods bound to this event should be used to ensure that the charm’s software is not running, and that it will not start again on reboot.

defer() NoReturn[source]

Stop events are not deferrable like other events.

This is because the unit is in the process of tearing down, and there will not be an opportunity for the deferred event to run.

Raises:

RuntimeError – always.

class ops.Storage(storage_name: str, storage_index: int, backend: _ModelBackend)[source]

Bases: object

Represents a storage as defined in metadata.yaml.

property full_id: str

Canonical storage name with index, for example “bigdisk/0”.

property id: int

DEPRECATED. Use Storage.index instead.

property index: int

Index associated with the storage (usually 0 for singular storage).

property location: Path

Location of the storage.

name: str

Name of the storage.

class ops.StorageAttachedEvent(handle: Handle, storage: Storage)[source]

Bases: StorageEvent

Event triggered when new storage becomes available.

This event is triggered when new storage is available for the charm to use.

Callback methods bound to this event allow the charm to run code when storage has been added. Such methods will be run before the InstallEvent fires, so that the installation routine may use the storage. The name prefix of this hook will depend on the storage key defined in the metadata.yaml file.

class ops.StorageDetachingEvent(handle: Handle, storage: Storage)[source]

Bases: StorageEvent

Event triggered prior to removal of storage.

This event is triggered when storage a charm has been using is going away.

Callback methods bound to this event allow the charm to run code before storage is removed. Such methods will be run before storage is detached, and always before the StopEvent fires, thereby allowing the charm to gracefully release resources before they are removed and before the unit terminates. The name prefix of the hook will depend on the storage key defined in the metadata.yaml file.

class ops.StorageEvent(handle: Handle, storage: Storage)[source]

Bases: HookEvent

Base class representing storage-related events.

Juju can provide a variety of storage types to a charms. The charms can define several different types of storage that are allocated from Juju. Changes in state of storage trigger sub-types of StorageEvent.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

storage: Storage

Storage instance this event refers to.

class ops.StorageMapping(storage_names: Iterable[str], backend: _ModelBackend)[source]

Bases: Mapping[str, List[Storage]]

Map of storage names to lists of Storage instances.

request(storage_name: str, count: int = 1)[source]

Requests new storage instances of a given name.

Uses storage-add tool to request additional storage. Juju will notify the unit via <storage-name>-storage-attached events when it becomes available.

Raises:

ModelError – if the storage is not in the charm’s metadata.

class ops.StorageMeta(name: str, raw: _StorageMetaDict)[source]

Bases: object

Object containing metadata about a storage definition.

description: str

Text description of the storage.

location: str | None

Mount point of the storage.

minimum_size: str | None

Minimum size of the storage.

multiple_range: Tuple[int, int | None] | None

Range of numeric qualifiers when multiple storage units are used.

properties

List of additional characteristics of the storage.

alias of List[str]

read_only: bool

True if the storage is read-only.

shared: bool

True if all units of the application share the storage.

storage_name: str

Name of storage.

type: str

Storage type, “filesystem” or “block”.

class ops.StoredDict(stored_data: StoredStateData, under: Dict[Hashable, Any])[source]

Bases: MutableMapping[Hashable, Any]

A dict-like object that uses the StoredState as backend.

class ops.StoredList(stored_data: StoredStateData, under: List[Any])[source]

Bases: MutableSequence[Any]

A list-like object that uses the StoredState as backend.

append(value: Any)[source]

Append value to the end of the list.

insert(index: int, value: Any)[source]

Insert value before index.

class ops.StoredSet(stored_data: StoredStateData, under: Set[Any])[source]

Bases: MutableSet[Any]

A set-like object that uses the StoredState as backend.

add(key: Any)[source]

Add a key to a set.

This has no effect if the key is already present.

discard(key: Any)[source]

Remove a key from a set if it is a member.

If the key is not a member, do nothing.

class ops.StoredState[source]

Bases: object

A class used to store data the charm needs, persisted across invocations.

Example:

class MyClass(ops.Object):
    _stored = ops.StoredState()

Instances of MyClass can transparently save state between invocations by setting attributes on _stored. Initial state should be set with set_default on the bound object, that is:

class MyClass(ops.Object):
    _stored = ops.StoredState()

    def __init__(self, parent, key):
        super().__init__(parent, key)
        self._stored.set_default(seen=set())
        self.framework.observe(self.on.seen, self._on_seen)

    def _on_seen(self, event):
        self._stored.seen.add(event.uuid)
class ops.StoredStateData(parent: Object, attr_name: str)[source]

Bases: Object

Manager of the stored data.

on_commit(event: EventBase) None[source]

Save changes to the storage backend.

restore(snapshot: Dict[str, Any])[source]

Restore current state to the given snapshot.

snapshot() Dict[str, Any][source]

Return the current state.

exception ops.TooManyRelatedAppsError(relation_name: str, num_related: int, max_supported: int)[source]

Bases: ModelError

Raised by Model.get_relation() if there is more than one integrated application.

class ops.Unit(name: str, meta: CharmMeta, backend: _ModelBackend, cache: _ModelCache)[source]

Bases: object

Represents a named unit in the model.

This might be the current unit, another unit of the charm’s application, or a unit of another application that the charm is integrated with.

add_secret(content: Dict[str, str], *, label: str | None = None, description: str | None = None, expire: datetime | timedelta | None = None, rotate: SecretRotate | None = None) Secret[source]

Create a Secret owned by this unit.

See Application.add_secret() for parameter details.

Raises:

ValueError – if the secret is empty, or the secret key is invalid.

app: Application

Application the unit is part of.

close_port(protocol: Literal['tcp', 'udp', 'icmp'], port: int | None = None) None[source]

Close a port with the given protocol for this unit.

Some behaviour, such as whether the port is closed externally without using “juju unexpose”, differs between Kubernetes and machine charms. See the Juju documentation for more detail.

Use set_ports() for a more declarative approach where all of the ports that should be open are provided in a single call. For example, set_ports() will close all open ports.

Parameters:
  • protocol – String representing the protocol; must be one of ‘tcp’, ‘udp’, or ‘icmp’ (lowercase is recommended, but uppercase is also supported).

  • port – The port to open. Required for TCP and UDP; not allowed for ICMP.

Raises:

ModelError – If port is provided when protocol is ‘icmp’ or port is not provided when protocol is ‘tcp’ or ‘udp’.

property containers: Mapping[str, Container]

Return a mapping of containers indexed by name.

Raises:

RuntimeError – if called for another unit

get_container(container_name: str) Container[source]

Get a single container by name.

Raises:

ModelError – if the named container doesn’t exist

is_leader() bool[source]

Return whether this unit is the leader of its application.

This can only be called for the current unit.

Raises:

RuntimeError – if called for another unit

name: str

Name of the unit, for example “mysql/0”.

open_port(protocol: Literal['tcp', 'udp', 'icmp'], port: int | None = None) None[source]

Open a port with the given protocol for this unit.

Some behaviour, such as whether the port is opened externally without using “juju expose” and whether the opened ports are per-unit, differs between Kubernetes and machine charms. See the Juju documentation for more detail.

Use set_ports() for a more declarative approach where all of the ports that should be open are provided in a single call.

Parameters:
  • protocol – String representing the protocol; must be one of ‘tcp’, ‘udp’, or ‘icmp’ (lowercase is recommended, but uppercase is also supported).

  • port – The port to open. Required for TCP and UDP; not allowed for ICMP.

Raises:

ModelError – If port is provided when protocol is ‘icmp’ or port is not provided when protocol is ‘tcp’ or ‘udp’.

opened_ports() Set[Port][source]

Return a list of opened ports for this unit.

reboot(now: bool = False) None[source]

Reboot the host machine.

Normally, the reboot will only take place after the current hook successfully completes. Use now=True to reboot immediately without waiting for the hook to complete; this is useful when multiple restarts are required (Juju will re-run the hook after rebooting).

This is not supported on Kubernetes charms, can only be called for the current unit, and cannot be used in an action hook.

Parameters:

now – terminate immediately without waiting for the current hook to complete, restarting the hook after reboot.

Raises:
set_ports(*ports: int | Port) None[source]

Set the open ports for this unit, closing any others that are open.

Some behaviour, such as whether the port is opened or closed externally without using Juju’s expose and unexpose commands, differs between Kubernetes and machine charms. See the Juju documentation for more detail.

Use open_port() and close_port() to manage ports individually.

Parameters:

ports – The ports to open. Provide an int to open a TCP port, or a Port to open a port for another protocol.

Raises:

ModelError – if a Port is provided where protocol is ‘icmp’ but port is not None, or where protocol is ‘tcp’ or ‘udp’ and port is None.

set_workload_version(version: str) None[source]

Record the version of the software running as the workload.

This shouldn’t be confused with the revision of the charm. This is informative only; shown in the output of ‘juju status’.

property status: StatusBase

Used to report or read the status of a specific unit.

Changes to status take effect immediately, unlike other Juju operations such as modifying relation data or secrets, which only take effect after a successful event.

The status of any unit other than the current unit is always Unknown.

Alternatively, use the collect_unit_status event to evaluate and set unit status consistently at the end of every hook.

Raises:

Example:

self.model.unit.status = ops.MaintenanceStatus('reconfiguring the frobnicators')
class ops.UnknownStatus[source]

Bases: StatusBase

The unit status is unknown.

A unit-agent has finished calling install, config-changed and start, but the charm has not called status-set yet.

This status is read-only; trying to set unit or application status to UnknownStatus will raise ModelError.

name = 'unknown'
class ops.UpdateStatusEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered by a status update request from Juju.

This event is periodically triggered by Juju so that it can provide constant feedback to the administrator about the status of the application the charm is modeling. Any callback method bound to this event should determine the “health” of the application and set the status appropriately.

The interval between update-status events can be configured model-wide, e.g. juju model-config update-status-hook-interval=1m.

class ops.UpgradeCharmEvent(handle: Handle)[source]

Bases: HookEvent

Event triggered by request to upgrade the charm.

This event will be triggered when an administrator executes juju upgrade-charm. The event fires after Juju has unpacked the upgraded charm code, and so this event will be handled by the callback method bound to the event in the new codebase. The associated callback method is invoked provided there is no existing error state. The callback method should be used to reconcile current state written by an older version of the charm into whatever form that is needed by the current charm version.

class ops.WaitingStatus(message: str = '')[source]

Bases: StatusBase

A unit is unable to progress.

The unit is unable to progress to an active state because an application with which it is integrated is not running.

name = 'waiting'
class ops.WorkloadEvent(handle: Handle, workload: Container)[source]

Bases: HookEvent

Base class representing workload-related events.

Workload events are generated for all containers that the charm expects in metadata.

restore(snapshot: Dict[str, Any])[source]

Used by the framework to deserialize the event from disk.

Not meant to be called by charm code.

snapshot() Dict[str, Any][source]

Used by the framework to serialize the event to disk.

Not meant to be called by charm code.

workload: Container

The workload involved in this event.

Workload currently only can be a Container, but in future may be other types that represent the specific workload type, for example a machine.

ops.pebble module

Client for the Pebble API (HTTP over Unix socket).

For a command-line interface for local testing, see test/pebble_cli.py.

exception ops.pebble.APIError(body: Dict[str, Any], code: int, status: str, message: str)[source]

Bases: Error

Raised when an HTTP API error occurs talking to the Pebble server.

body: Dict[str, Any]

Body of the HTTP response, parsed as JSON.

code: int

HTTP status code.

message: str

Human-readable error message from the API.

status: str

HTTP status string (reason).

class ops.pebble.Change(id: ChangeID, kind: str, summary: str, status: str, tasks: List[Task], ready: bool, err: str | None, spawn_time: datetime, ready_time: datetime | None, data: Dict[str, Any] | None = None)[source]

Bases: object

Change object.

classmethod from_dict(d: _ChangeDict) Change[source]

Create new Change object from dict parsed from JSON.

exception ops.pebble.ChangeError(err: str, change: Change)[source]

Bases: Error

Raised by actions when a change is ready but has an error.

change: Change

Change object associated with this error.

err: str

Human-readable error message.

class ops.pebble.ChangeID[source]

Bases: str

Change ID (a more strongly-typed string).

class ops.pebble.ChangeState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of states for get_changes() select parameter.

ALL = 'all'
IN_PROGRESS = 'in-progress'
READY = 'ready'
class ops.pebble.Check(name: str, raw: CheckDict | None = None)[source]

Bases: object

Represents a check in a Pebble configuration layer.

to_dict() CheckDict[source]

Convert this check object to its dict representation.

class ops.pebble.CheckDict

Bases: dict

exec: ExecDict | None
http: HttpDict | None
level: CheckLevel | str
override: str
period: str | None
tcp: TcpDict | None
threshold: int | None
timeout: str | None
class ops.pebble.CheckInfo(name: str, level: CheckLevel | str | None, status: CheckStatus | str, failures: int = 0, threshold: int = 0)[source]

Bases: object

Check status information.

A list of these objects is returned from Client.get_checks().

failures: int

Number of failures since the check last succeeded.

This is reset to zero if the check succeeds.

classmethod from_dict(d: _CheckInfoDict) CheckInfo[source]

Create new CheckInfo object from dict parsed from JSON.

level: CheckLevel | str | None

Check level.

This can be CheckLevel.ALIVE, CheckLevel.READY, or None (level not set).

name: str

Name of the check.

status: CheckStatus | str

Status of the check.

CheckStatus.UP means the check is healthy (the number of failures is less than the threshold), CheckStatus.DOWN means the check is unhealthy (the number of failures has reached the threshold).

threshold: int

Failure threshold.

This is how many consecutive failures for the check to be considered “down”.

class ops.pebble.CheckLevel(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of check levels.

ALIVE = 'alive'
READY = 'ready'
UNSET = ''
class ops.pebble.CheckStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of check statuses.

DOWN = 'down'
UP = 'up'
class ops.pebble.Client(socket_path: str, opener: OpenerDirector | None = None, base_url: str = 'http://localhost', timeout: float = 5.0)[source]

Bases: object

Pebble API client.

Defaults to using a Unix socket at socket_path (which must be specified unless a custom opener is provided).

For methods that wait for changes, such as start_services() and replan_services(), if the change fails or times out, then a ChangeError or TimeoutError will be raised.

All methods may raise exceptions when there are problems communicating with Pebble. Problems connecting to or transferring data with Pebble will raise a ConnectionError. When an error occurs executing the request, such as trying to add an invalid layer or execute a command that does not exist, an APIError is raised.

The timeout parameter specifies a timeout in seconds for blocking operations like the connection attempt to Pebble; used by urllib.request.OpenerDirector.open. It’s not for methods like start_services() and replan_services() mentioned above, and it’s not for the command execution timeout defined in method Client.exec().

abort_change(change_id: ChangeID) Change[source]

Abort change with given ID.

ack_warnings(timestamp: datetime) int[source]

Acknowledge warnings up to given timestamp, return number acknowledged.

add_layer(label: str, layer: str | LayerDict | Layer, *, combine: bool = False)[source]

Dynamically add a new layer onto the Pebble configuration layers.

If combine is False (the default), append the new layer as the top layer with the given label. If combine is True and the label already exists, the two layers are combined into a single one considering the layer override rules; if the layer doesn’t exist, it is added as usual.

autostart_services(timeout: float = 30.0, delay: float = 0.1) ChangeID[source]

Start the startup-enabled services and wait (poll) for them to be started.

Parameters:
  • timeout – Seconds before autostart change is considered timed out (float). If timeout is 0, submit the action but don’t wait; just return the change ID immediately.

  • delay – Seconds before executing the autostart change (float).

Returns:

ChangeID of the autostart change.

Raises:

ChangeError – if one or more of the services didn’t start, and timeout is non-zero.

exec(command: List[str], *, service_context: str | None = None, environment: Dict[str, str] | None = None, working_dir: str | None = None, timeout: float | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None, stdin: str | TextIO | None = None, stdout: TextIO | None = None, stderr: TextIO | None = None, encoding: str = 'utf-8', combine_stderr: bool = False) ExecProcess[str][source]
exec(command: List[str], *, service_context: str | None = None, environment: Dict[str, str] | None = None, working_dir: str | None = None, timeout: float | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None, stdin: bytes | BinaryIO | None = None, stdout: BinaryIO | None = None, stderr: BinaryIO | None = None, encoding: None = None, combine_stderr: bool = False) ExecProcess[bytes]

Execute the given command on the remote system.

Two method signatures are shown because this method returns an ExecProcess that deals with strings if encoding is specified (the default ), or one that deals with bytes if encoding is set to None.

Most of the parameters are explained in the “Parameters” section below, however, input/output handling is a bit more complex. Some examples are shown below:

# Simple command with no output; just check exit code
>>> process = client.exec(['send-emails'])
>>> process.wait()

# Fetch output as string
>>> process = client.exec(['python3', '--version'])
>>> version, _ = process.wait_output()
>>> print(version)
Python 3.8.10

# Fetch both stdout and stderr as strings
>>> process = client.exec(['pg_dump', '-s', ...])
>>> schema, logs = process.wait_output()

# Stream input from a string and write output to files
>>> stdin = 'foo\nbar\n'
>>> with open('out.txt', 'w') as out, open('err.txt', 'w') as err:
...     process = client.exec(['awk', '{ print toupper($0) }'],
...                           stdin=stdin, stdout=out, stderr=err)
...     process.wait()
>>> open('out.txt').read()
'FOO\nBAR\n'
>>> open('err.txt').read()
''

# Real-time streaming using ExecProcess.stdin and ExecProcess.stdout
>>> process = client.exec(['cat'])
>>> def stdin_thread():
...     for line in ['one\n', '2\n', 'THREE\n']:
...         process.stdin.write(line)
...         process.stdin.flush()
...         time.sleep(1)
...     process.stdin.close()
...
>>> threading.Thread(target=stdin_thread).start()
>>> for line in process.stdout:
...     print(datetime.datetime.now().strftime('%H:%M:%S'), repr(line))
...
16:20:26 'one\n'
16:20:27 '2\n'
16:20:28 'THREE\n'
>>> process.wait()  # will return immediately as stdin was closed

# Show exception raised for non-zero return code
>>> process = client.exec(['ls', 'notexist'])
>>> out, err = process.wait_output()
Traceback (most recent call last):
  ...
ExecError: "ls" returned exit code 2
>>> exc = sys.last_value
>>> exc.exit_code
2
>>> exc.stdout
''
>>> exc.stderr
"ls: cannot access 'notfound': No such file or directory\n"
Parameters:
  • command – Command to execute: the first item is the name (or path) of the executable, the rest of the items are the arguments.

  • service_context – If specified, run the command in the context of this service. Specifically, inherit its environment variables, user/group settings, and working directory. The other exec options will override the service context; environment will be merged on top of the service’s.

  • environment – Environment variables to pass to the process.

  • working_dir – Working directory to run the command in. If not set, Pebble uses the target user’s $HOME directory (and if the user argument is not set, $HOME of the user Pebble is running as).

  • timeout – Timeout in seconds for the command execution, after which the process will be terminated. If not specified, the execution never times out.

  • user_id – User ID (UID) to run the process as.

  • user – Username to run the process as. User’s UID must match user_id if both are specified.

  • group_id – Group ID (GID) to run the process as.

  • group – Group name to run the process as. Group’s GID must match group_id if both are specified.

  • stdin – A string or readable file-like object that is sent to the process’s standard input. If not set, the caller can write input to ExecProcess.stdin to stream input to the process.

  • stdout – A writable file-like object that the process’s standard output is written to. If not set, the caller can use ExecProcess.wait_output() to capture output as a string, or read from ExecProcess.stdout() to stream output from the process.

  • stderr – A writable file-like object that the process’s standard error is written to. If not set, the caller can use ExecProcess.wait_output() to capture error output as a string, or read from ExecProcess.stderr() to stream error output from the process. Must be None if combine_stderr is True.

  • encoding – If encoding is set (the default is UTF-8), the types read or written to stdin/stdout/stderr are str, and encoding is used to encode them to bytes. If encoding is None, the types read or written are raw bytes.

  • combine_stderr – If True, process’s stderr output is combined into its stdout (the stderr argument must be None). If False, separate streams are used for stdout and stderr.

Returns:

A Process object representing the state of the running process. To wait for the command to finish, the caller will typically call ExecProcess.wait() if stdout/stderr were provided as arguments to exec(), or ExecProcess.wait_output() if not.

Raises:
  • APIError – if an error occurred communicating with pebble, or if the command is not found.

  • ExecError – if the command exits with a non-zero exit code.

get_change(change_id: ChangeID) Change[source]

Get single change by ID.

get_changes(select: ChangeState = ChangeState.IN_PROGRESS, service: str | None = None) List[Change][source]

Get list of changes in given state, filter by service name if given.

get_checks(level: CheckLevel | None = None, names: Iterable[str] | None = None) List[CheckInfo][source]

Get the check status for the configured checks.

Parameters:
  • level – Optional check level to query for (default is to fetch checks with any level).

  • names – Optional list of check names to query for (default is to fetch all checks).

Returns:

List of CheckInfo objects.

get_notice(id: str) Notice[source]

Get details about a single notice by ID.

Raises:

APIError – if a notice with the given ID is not found (code 404)

get_notices(*, users: NoticesUsers | None = None, user_id: int | None = None, types: Iterable[NoticeType | str] | None = None, keys: Iterable[str] | None = None) List[Notice][source]

Query for notices that match all of the provided filters.

Pebble returns notices that match all of the filters, for example, if called with types=[NoticeType.CUSTOM], keys=["example.com/a"], Pebble will only return custom notices that also have key “example.com/a”.

If no filters are specified, return notices viewable by the requesting user (notices whose user_id matches the requester UID as well as public notices).

Note that the “after” filter is not yet implemented, as it’s not needed right now and it’s hard to implement correctly with Python’s datetime type, which only has microsecond precision (and Go’s Time type has nanosecond precision).

Parameters:
  • users – Change which users’ notices to return (instead of returning notices for the current user).

  • user_id – Filter for notices for the specified user, including public notices (only works for Pebble admins).

  • types – Filter for notices with any of the specified types.

  • keys – Filter for notices with any of the specified keys.

get_plan() Plan[source]

Get the Pebble plan (contains combined layer configuration).

get_services(names: Iterable[str] | None = None) List[ServiceInfo][source]

Get the service status for the configured services.

If names is specified, only fetch the service status for the services named.

get_system_info() SystemInfo[source]

Get system info.

get_warnings(select: WarningState = WarningState.PENDING) List[Warning][source]

Get list of warnings in given state (pending or all).

list_files(path: str, *, pattern: str | None = None, itself: bool = False) List[FileInfo][source]

Return list of directory entries from given path on remote system.

Despite the name, this method returns a list of files and directories, similar to os.listdir() or os.scandir().

Parameters:
  • path – Path of the directory to list, or path of the file to return information about.

  • pattern – If specified, filter the list to just the files that match, for example *.txt.

  • itself – If path refers to a directory, return information about the directory itself, rather than its contents.

Raises:

PathError – if there was an error listing the directory; for example, if the directory does not exist.

make_dir(path: str, *, make_parents: bool = False, permissions: int | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None)[source]

Create a directory on the remote system with the given attributes.

Parameters:
  • path – Path of the directory to create on the remote system.

  • make_parents – If True, create parent directories if they don’t exist.

  • permissions – Permissions (mode) to create directory with (Pebble default is 0o755).

  • user_id – User ID (UID) for directory.

  • user – Username for directory. User’s UID must match user_id if both are specified.

  • group_id – Group ID (GID) for directory.

  • group – Group name for directory. Group’s GID must match group_id if both are specified.

Raises:

PathError – if there was an error making the directory; for example, if the parent path does not exist, and make_parents is not used.

notify(type: NoticeType, key: str, *, data: Dict[str, str] | None = None, repeat_after: timedelta | None = None) str[source]

Record an occurrence of a notice with the specified options.

Parameters:
  • type – Notice type (currently only “custom” notices are supported).

  • key – Notice key; must be in “example.com/path” format.

  • data – Data fields for this notice.

  • repeat_after – Only allow this notice to repeat after this duration has elapsed (the default is to always repeat).

Returns:

The notice’s ID.

pull(path: str, *, encoding: None) BinaryIO[source]
pull(path: str, *, encoding: str = 'utf-8') TextIO

Read a file’s content from the remote system.

Parameters:
  • path – Path of the file to read from the remote system.

  • encoding – Encoding to use for decoding the file’s bytes to str, or None to specify no decoding.

Returns:

A readable file-like object, whose read() method will return str objects decoded according to the specified encoding, or bytes if encoding is None.

Raises:

PathError – If there was an error reading the file at path, for example, if the file doesn’t exist or is a directory.

push(path: str, source: str | bytes | _FileLikeIO[bytes] | _FileLikeIO[str], *, encoding: str = 'utf-8', make_dirs: bool = False, permissions: int | None = None, user_id: int | None = None, user: str | None = None, group_id: int | None = None, group: str | None = None)[source]

Write content to a given file path on the remote system.

Parameters:
  • path – Path of the file to write to on the remote system.

  • source – Source of data to write. This is either a concrete str or bytes instance, or a readable file-like object.

  • encoding – Encoding to use for encoding source str to bytes, or strings read from source if it is a TextIO type. Ignored if source is bytes or BinaryIO.

  • make_dirs – If True, create parent directories if they don’t exist.

  • permissions – Permissions (mode) to create file with (Pebble default is 0o644).

  • user_id – User ID (UID) for file.

  • user – Username for file. User’s UID must match user_id if both are specified.

  • group_id – Group ID (GID) for file.

  • group – Group name for file. Group’s GID must match group_id if both are specified.

Raises:

PathError – If there was an error writing the file to the path; for example, if the destination path doesn’t exist and make_dirs is not used.

remove_path(path: str, *, recursive: bool = False)[source]

Remove a file or directory on the remote system.

Parameters:
  • path – Path of the file or directory to delete from the remote system.

  • recursive – If True, and path is a directory, recursively delete it and everything under it. If path is a file, delete the file. In either case, do nothing if the file or directory does not exist. Behaviourally similar to rm -rf <file|dir>.

Raises:

pebble.PathError – If a relative path is provided, or if recursive is False and the file or directory cannot be removed (it does not exist or is not empty).

replan_services(timeout: float = 30.0, delay: float = 0.1) ChangeID[source]

Replan by (re)starting changed and startup-enabled services and wait for them to start.

Parameters:
  • timeout – Seconds before replan change is considered timed out (float). If timeout is 0, submit the action but don’t wait; just return the change ID immediately.

  • delay – Seconds before executing the replan change (float).

Returns:

ChangeID of the replan change.

Raises:

ChangeError – if one or more of the services didn’t stop/start, and timeout is non-zero.

restart_services(services: Iterable[str], timeout: float = 30.0, delay: float = 0.1) ChangeID[source]

Restart services by name and wait (poll) for them to be started.

Parameters:
  • services – Non-empty list of services to restart.

  • timeout – Seconds before restart change is considered timed out (float). If timeout is 0, submit the action but don’t wait; just return the change ID immediately.

  • delay – Seconds before executing the restart change (float).

Returns:

ChangeID of the restart change.

Raises:

ChangeError – if one or more of the services didn’t stop/start and timeout is non-zero.

send_signal(sig: int | str, services: Iterable[str])[source]

Send the given signal to the list of services named.

Parameters:
  • sig – Name or number of signal to send, for example "SIGHUP", 1, or signal.SIGHUP.

  • services – Non-empty list of service names to send the signal to.

Raises:

APIError – If any of the services are not in the plan or are not currently running.

start_services(services: Iterable[str], timeout: float = 30.0, delay: float = 0.1) ChangeID[source]

Start services by name and wait (poll) for them to be started.

Parameters:
  • services – Non-empty list of services to start.

  • timeout – Seconds before start change is considered timed out (float). If timeout is 0, submit the action but don’t wait; just return the change ID immediately.

  • delay – Seconds before executing the start change (float).

Returns:

ChangeID of the start change.

Raises:

ChangeError – if one or more of the services didn’t stop/start, and timeout is non-zero.

stop_services(services: Iterable[str], timeout: float = 30.0, delay: float = 0.1) ChangeID[source]

Stop services by name and wait (poll) for them to be started.

Parameters:
  • services – Non-empty list of services to stop.

  • timeout – Seconds before stop change is considered timed out (float). If timeout is 0, submit the action but don’t wait; just return the change ID immediately.

  • delay – Seconds before executing the stop change (float).

Returns:

ChangeID of the stop change.

Raises:

ChangeError – if one or more of the services didn’t stop/start and timeout is non-zero.

wait_change(change_id: ChangeID, timeout: float | None = 30.0, delay: float = 0.1) Change[source]

Wait for the given change to be ready.

If the Pebble server supports the /v1/changes/{id}/wait API endpoint, use that to avoid polling, otherwise poll /v1/changes/{id} every delay seconds.

Parameters:
  • change_id – Change ID of change to wait for.

  • timeout – Maximum time in seconds to wait for the change to be ready. It may be None, in which case wait_change never times out.

  • delay – If polling, this is the delay in seconds between attempts.

Returns:

The Change object being waited on.

Raises:

TimeoutError – If the maximum timeout is reached.

exception ops.pebble.ConnectionError[source]

Bases: Error

Raised when the Pebble client can’t connect to the socket.

exception ops.pebble.Error[source]

Bases: Exception

Base class of most errors raised by the Pebble client.

class ops.pebble.ExecDict

Bases: dict

command: str
environment: Dict[str, str]
group: str
ExecDict.group-id
ExecDict.service-context
user: str
ExecDict.user-id
ExecDict.working-dir
exception ops.pebble.ExecError(command: List[str], exit_code: int, stdout: AnyStr | None, stderr: AnyStr | None)[source]

Bases: Error, Generic

Raised when a Client.exec() command returns a non-zero exit code.

STR_MAX_OUTPUT = 1024

Maximum number of characters that stdout/stderr are truncated to in __str__.

command: List[str]

Command line of command being executed.

exit_code: int

The process’s exit code. Because this is an error, this will always be non-zero.

stderr: AnyStr | None

Standard error from the process.

If ExecProcess.wait_output() was being called and combine_stderr was False, this is the captured stderr as a str (or bytes if encoding was None). If ExecProcess.wait() was being called or combine_stderr was True, this is None.

stdout: AnyStr | None

Standard output from the process.

If ExecProcess.wait_output() was being called, this is the captured stdout as a str (or bytes if encoding was None). If ExecProcess.wait() was being called, this is None.

class ops.pebble.ExecProcess(stdin: IO | None, stdout: IO | None, stderr: IO | None, client: Client, timeout: float | None, control_ws: _WebSocket, stdio_ws: _WebSocket, stderr_ws: _WebSocket | None, command: List[str], encoding: str | None, change_id: ChangeID, cancel_stdin: Callable[[], None] | None, cancel_reader: int | None, threads: List[Thread])[source]

Bases: Generic

Represents a process started by Client.exec().

To avoid deadlocks, most users should use wait_output() instead of reading and writing the stdin, stdout, and stderr attributes directly. Alternatively, users can pass stdin/stdout/stderr to Client.exec().

This class should not be instantiated directly, only via Client.exec().

send_signal(sig: int | str)[source]

Send the given signal to the running process.

Parameters:

sig – Name or number of signal to send, e.g., “SIGHUP”, 1, or signal.SIGHUP.

stderr: IO | None

Standard error from the process.

If the stderr argument was not passed to Client.exec() and combine_stderr was False, this is a readable file-like object the caller can use to stream error output from the process. It is None if stderr was passed to Client.exec() or combine_stderr was True.

stdin: IO | None

Standard input for the process.

If the stdin argument was not passed to Client.exec(), this is a writable file-like object the caller can use to stream input to the process. It is None if stdin was passed to Client.exec().

stdout: IO | None

Standard output from the process.

If the stdout argument was not passed to Client.exec(), this is a readable file-like object the caller can use to stream output from the process. It is None if stdout was passed to Client.exec().

wait()[source]

Wait for the process to finish.

If a timeout was specified to the Client.exec() call, this waits at most that duration.

Raises:
  • ChangeError – if there was an error starting or running the process.

  • ExecError – if the process exits with a non-zero exit code.

wait_output() Tuple[AnyStr, AnyStr | None][source]

Wait for the process to finish and return tuple of (stdout, stderr).

If a timeout was specified to the Client.exec() call, this waits at most that duration. If combine_stderr was True, stdout will include the process’s standard error, and stderr will be None.

Raises:
class ops.pebble.FileInfo(path: str, name: str, type: FileType | str, size: int | None, permissions: int, last_modified: datetime, user_id: int | None, user: str | None, group_id: int | None, group: str | None)[source]

Bases: object

Stat-like information about a single file or directory.

classmethod from_dict(d: _FileInfoDict) FileInfo[source]

Create new FileInfo object from dict parsed from JSON.

group: str | None

Group name of the file.

group_id: int | None

Group ID of the file.

last_modified: datetime

Time file was last modified.

name: str

Base name of the file.

path: str

Full path of the file.

permissions: int

Unix permissions of the file.

size: int | None

Size of the file (will be 0 if type is not “file”).

type: FileType | str

Type of the file (“file”, “directory”, “symlink”, etc).

user: str | None

Username of the file.

user_id: int | None

User ID of the file.

class ops.pebble.FileType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of file types.

DEVICE = 'device'
DIRECTORY = 'directory'
FILE = 'file'
NAMED_PIPE = 'named-pipe'
SOCKET = 'socket'
UNKNOWN = 'unknown'
class ops.pebble.HttpDict

Bases: dict

headers: Dict[str, str]
url: str
class ops.pebble.Layer(raw: str | LayerDict | None = None)[source]

Bases: object

Represents a Pebble configuration layer.

The format of this is documented at https://github.com/canonical/pebble/#layer-specification.

checks: Dict[str, Check]

Mapping of check to Check defined by this layer.

description: str

Long-form description of this layer.

log_targets: Dict[str, LogTarget]

Mapping of target to LogTarget defined by this layer.

services: Dict[str, Service]

Mapping of name to Service defined by this layer.

summary: str

Summary of the purpose of this layer.

to_dict() LayerDict[source]

Convert this layer to its dict representation.

to_yaml() str[source]

Convert this layer to its YAML representation.

class ops.pebble.LayerDict

Bases: dict

checks: Dict[str, CheckDict]
description: str
LayerDict.log-targets
services: Dict[str, ServiceDict]
summary: str
class ops.pebble.LogTarget(name: str, raw: LogTargetDict | None = None)[source]

Bases: object

Represents a log target in a Pebble configuration layer.

to_dict() LogTargetDict[source]

Convert this log target object to its dict representation.

class ops.pebble.LogTargetDict

Bases: dict

labels: Dict[str, str]
location: str
override: Literal['merge'] | Literal['replace']
services: List[str]
type: Literal['loki']
class ops.pebble.Notice(id: str, user_id: int | None, type: ~ops.pebble.NoticeType | str, key: str, first_occurred: ~datetime.datetime, last_occurred: ~datetime.datetime, last_repeated: ~datetime.datetime, occurrences: int, last_data: ~typing.Dict[str, str] = <factory>, repeat_after: ~datetime.timedelta | None = None, expire_after: ~datetime.timedelta | None = None)[source]

Bases: object

Information about a single notice.

expire_after: timedelta | None = None

How long since one of these last occurred until Pebble will drop the notice.

first_occurred: datetime

The first time one of these notices (type and key combination) occurs.

classmethod from_dict(d: _NoticeDict) Notice[source]

Create new Notice object from dict parsed from JSON.

id: str

Server-generated unique ID for this notice.

key: str

The notice key, a string that differentiates notices of this type.

This is in the format example.com/path.

last_data: Dict[str, str]

Additional data captured from the last occurrence of one of these notices.

last_occurred: datetime

The last time one of these notices occurred.

last_repeated: datetime

The time this notice was last repeated.

See Pebble’s Notices documentation for an explanation of what “repeated” means.

occurrences: int

The number of times one of these notices has occurred.

repeat_after: timedelta | None = None

Minimum time after one of these was last repeated before Pebble will repeat it again.

type: NoticeType | str

Type of the notice.

user_id: int | None

UID of the user who may view this notice (None means notice is public).

class ops.pebble.NoticeType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of notice types.

CUSTOM = 'custom'
class ops.pebble.NoticesUsers(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of Client.get_notices() users values.

ALL = 'all'

Return notices from all users (any user ID, including public notices).

This only works for Pebble admins (for example, root).

exception ops.pebble.PathError(kind: str, message: str)[source]

Bases: Error

Raised when there’s an error with a specific path.

kind: Literal['not-found', 'permission-denied', 'generic-file-error']

Short string representing the kind of error.

message: str

Human-readable error message from the API.

class ops.pebble.Plan(raw: str | PlanDict | None = None)[source]

Bases: object

Represents the effective Pebble configuration.

A plan is the combined layer configuration. The layer configuration is documented at https://github.com/canonical/pebble/#layer-specification.

property checks: Dict[str, Check]

This plan’s checks mapping (maps check name to Check).

This property is currently read-only.

property log_targets: Dict[str, LogTarget]

This plan’s log targets mapping (maps log target name to LogTarget).

This property is currently read-only.

property services: Dict[str, Service]

This plan’s services mapping (maps service name to Service).

This property is currently read-only.

to_dict() PlanDict[source]

Convert this plan to its dict representation.

to_yaml() str[source]

Return this plan’s YAML representation.

class ops.pebble.PlanDict

Bases: dict

checks: Dict[str, CheckDict]
PlanDict.log-targets
services: Dict[str, ServiceDict]
exception ops.pebble.ProtocolError[source]

Bases: Error

Raised when there’s a higher-level protocol error talking to Pebble.

class ops.pebble.Service(name: str, raw: ServiceDict | None = None)[source]

Bases: object

Represents a service description in a Pebble configuration layer.

to_dict() ServiceDict[source]

Convert this service object to its dict representation.

class ops.pebble.ServiceDict

Bases: dict

after: Sequence[str]
ServiceDict.backoff-delay
ServiceDict.backoff-factor
ServiceDict.backoff-limit
before: Sequence[str]
command: str
description: str
environment: Dict[str, str]
group: str
ServiceDict.group-id
ServiceDict.kill-delay
ServiceDict.on-check-failure
ServiceDict.on-failure
ServiceDict.on-success
override: str
requires: Sequence[str]
startup: str
summary: str
user: str
ServiceDict.user-id
ServiceDict.working-dir
class ops.pebble.ServiceInfo(name: str, startup: ServiceStartup | str, current: ServiceStatus | str)[source]

Bases: object

Service status information.

classmethod from_dict(d: _ServiceInfoDict) ServiceInfo[source]

Create new ServiceInfo object from dict parsed from JSON.

is_running() bool[source]

Return True if this service is running (in the active state).

class ops.pebble.ServiceStartup(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of service startup options.

DISABLED = 'disabled'
ENABLED = 'enabled'
class ops.pebble.ServiceStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of service statuses.

ACTIVE = 'active'
ERROR = 'error'
INACTIVE = 'inactive'
class ops.pebble.SystemInfo(version: str)[source]

Bases: object

System information object.

classmethod from_dict(d: _SystemInfoDict) SystemInfo[source]

Create new SystemInfo object from dict parsed from JSON.

class ops.pebble.Task(id: TaskID, kind: str, summary: str, status: str, log: List[str], progress: TaskProgress, spawn_time: datetime, ready_time: datetime | None, data: Dict[str, Any] | None = None)[source]

Bases: object

Task object.

classmethod from_dict(d: _TaskDict) Task[source]

Create new Task object from dict parsed from JSON.

class ops.pebble.TaskID[source]

Bases: str

Task ID (a more strongly-typed string).

class ops.pebble.TaskProgress(label: str, done: int, total: int)[source]

Bases: object

Task progress object.

classmethod from_dict(d: _ProgressDict) TaskProgress[source]

Create new TaskProgress object from dict parsed from JSON.

class ops.pebble.TcpDict

Bases: dict

host: str
port: int
exception ops.pebble.TimeoutError[source]

Bases: TimeoutError, Error

Raised when a polling timeout occurs.

class ops.pebble.Warning(message: str, first_added: datetime, last_added: datetime, last_shown: datetime | None, expire_after: str, repeat_after: str)[source]

Bases: object

Warning object.

classmethod from_dict(d: _WarningDict) Warning[source]

Create new Warning object from dict parsed from JSON.

class ops.pebble.WarningState(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Enum of states for get_warnings() select parameter.

ALL = 'all'
PENDING = 'pending'

ops.testing module

Infrastructure to build unit tests for charms using the ops library.

exception ops.testing.ActionFailed(message: str, output: ActionOutput)[source]

Bases: Exception

Raised when event.fail() is called during a Harness.run_action() call.

message: str

Optional details of the failure, as provided by ops.ActionEvent.fail().

output: ActionOutput

Any logs and results set by the Charm.

class ops.testing.ActionOutput(logs: List[str], results: Dict[str, Any])[source]

Bases: object

Contains the logs and results from a Harness.run_action() call.

logs: List[str]

Messages generated by the Charm using ops.ActionEvent.log().

results: Dict[str, Any]

The action’s results, as set or updated by ops.ActionEvent.set_results().

class ops.testing.ExecArgs(command: List[str], environment: Dict[str, str], working_dir: str | None, timeout: float | None, user_id: int | None, user: str | None, group_id: int | None, group: str | None, stdin: str | bytes | None, encoding: str | None, combine_stderr: bool)[source]

Bases: object

Represent arguments captured from the ops.Container.exec() method call.

These arguments will be passed to the Harness.handle_exec() handler function. See ops.pebble.Client.exec() for documentation of properties.

combine_stderr: bool
command: List[str]
encoding: str | None
environment: Dict[str, str]
group: str | None
group_id: int | None
stdin: str | bytes | None
timeout: float | None
user: str | None
user_id: int | None
working_dir: str | None
class ops.testing.ExecResult(exit_code: int = 0, stdout: str | bytes = b'', stderr: str | bytes = b'')[source]

Bases: object

Represents the result of a simulated process execution.

This class is typically used to return the output and exit code from the Harness.handle_exec() result or handler function.

exit_code: int = 0
stderr: str | bytes = b''
stdout: str | bytes = b''
class ops.testing.Harness(charm_cls: Type[CharmType], *, meta: str | TextIO | None = None, actions: str | TextIO | None = None, config: str | TextIO | None = None)[source]

Bases: Generic[CharmType]

This class represents a way to build up the model that will drive a test suite.

The model created is from the viewpoint of the charm that is being tested.

Always call harness.cleanup() after creating a Harness:

@pytest.fixture()
def harness():
    harness = Harness(MyCharm)
    yield harness
    harness.cleanup()

Below is an example test using begin_with_initial_hooks() that ensures the charm responds correctly to config changes (the parameter harness in the test function is a pytest fixture that does setup/teardown, see Harness):

def test_foo(harness):
    # Instantiate the charm and trigger events that Juju would on startup
    harness.begin_with_initial_hooks()

    # Update charm config and trigger config-changed
    harness.update_config({'log_level': 'warn'})

    # Check that charm properly handled config-changed, for example,
    # the charm added the correct Pebble layer
    plan = harness.get_container_pebble_plan('prometheus')
    assert '--log.level=warn' in plan.services['prometheus'].command

To set up the model without triggering events (or calling charm code), perform the harness actions before calling begin(). Below is an example that adds a relation before calling begin, and then updates config to trigger the config-changed event in the charm (the parameter harness in the test function is a pytest fixture that does setup/teardown, see Harness):

def test_bar(harness):
    # Set up model before "begin" (no events triggered)
    harness.set_leader(True)
    harness.add_relation('db', 'postgresql', unit_data={'key': 'val'})

    # Now instantiate the charm to start triggering events as the model changes
    harness.begin()
    harness.update_config({'some': 'config'})

    # Check that charm has properly handled config-changed, for example,
    # has written the app's config file
    root = harness.get_filesystem_root('container')
    assert (root / 'etc' / 'app.conf').exists()
Parameters:
  • charm_cls – The Charm class to test.

  • meta – A string or file-like object containing the contents of metadata.yaml. If not supplied, we will look for a metadata.yaml file in the parent directory of the Charm, and if not found fall back to a trivial name: test-charm metadata.

  • actions – A string or file-like object containing the contents of actions.yaml. If not supplied, we will look for an actions.yaml file in the parent directory of the Charm.

  • config – A string or file-like object containing the contents of config.yaml. If not supplied, we will look for a config.yaml file in the parent directory of the Charm.

add_model_secret(owner: str | Application | Unit, content: Dict[str, str]) str[source]

Add a secret owned by the remote application or unit specified.

This is named add_model_secret instead of add_secret to avoid confusion with the ops.Application.add_secret() and ops.Unit.add_secret() methods used by secret owner charms.

Parameters:
  • owner – The name of the remote application (or specific remote unit) that will own the secret.

  • content – A key-value mapping containing the payload of the secret, for example {"password": "foo123"}.

Returns:

The ID of the newly-secret added.

add_network(address: str, *, endpoint: str | None = None, relation_id: int | None = None, cidr: str | None = None, interface: str = 'eth0', ingress_addresses: Iterable[str] | None = None, egress_subnets: Iterable[str] | None = None)[source]

Add simulated network data for the given relation endpoint (binding).

Calling this multiple times with the same (binding, relation_id) combination will replace the associated network data.

Example:

# Set network info for default binding
harness.add_network('10.0.0.10')

# Or set network info for specific endpoint
harness.add_network('10.0.0.10', endpoint='db')

After either of those calls, the following will be true (in the first case, the simulated network-get will fall back to the default binding):

binding = harness.model.get_binding('db')
assert binding.network.bind_address == ipaddress.IPv4Address('10.0.0.10'))
Parameters:
  • address – Binding’s IPv4 or IPv6 address.

  • endpoint – Name of relation endpoint (binding) to add network data for. If not provided, add info for the default binding.

  • relation_id – Relation ID for the binding. If provided, the endpoint argument must be provided and correspond. If not provided, add network data for the endpoint’s default binding.

  • cidr – Binding’s CIDR. Defaults to “<address>/24” if address is an IPv4 address, or “<address>/64” if address is IPv6 (the host bits are cleared).

  • interface – Name of network interface.

  • ingress_addresses – List of ingress addresses. Defaults to [address].

  • egress_subnets – List of egress subnets. Defaults to [cidr].

Raises:
  • ModelError – If the endpoint is not a known relation name, or the relation_id is incorrect or doesn’t match the endpoint.

  • ValueError – If address is not an IPv4 or IPv6 address.

add_oci_resource(resource_name: str, contents: Mapping[str, str] | None = None) None[source]

Add OCI resources to the backend.

This will register an OCI resource and create a temporary file for processing metadata about the resource. A default set of values will be used for all the file contents unless a specific contents dict is provided.

Parameters:
  • resource_name – Name of the resource to add custom contents to.

  • contents – Optional custom dict to write for the named resource.

add_relation(relation_name: str, remote_app: str, *, app_data: Mapping[str, str] | None = None, unit_data: Mapping[str, str] | None = None) int[source]

Declare that there is a new relation between this application and remote_app.

This function creates a relation with an application and triggers a RelationCreatedEvent. To match Juju’s behaviour, it also creates a default network binding on this endpoint. If you want to associate a custom network to this binding (or a global default network), provide one using add_network() before calling this function.

If app_data or unit_data are provided, also add a new unit (<remote_app>/0) to the relation and trigger RelationJoinedEvent. Then update the application data if app_data is provided and the unit data if unit_data is provided, triggering RelationChangedEvent after each update. Alternatively, charm tests can call add_relation_unit() and update_relation_data() explicitly.

For peer relations defined in the charm’s metadata, begin_with_initial_hooks() will create them automatically, so the caller doesn’t need to call add_relation(). If the caller chooses to add a peer relation by themselves, make sure to call add_relation() before begin_with_initial_hooks() so that Harness won’t create it again.

Example usage:

secret_id = harness.add_model_secret('mysql', {'password': 'SECRET'})
harness.add_relation('db', 'mysql', unit_data={
    'host': 'mysql.localhost,
    'username': 'appuser',
    'secret-id': secret_id,
})
Parameters:
  • relation_name – The relation on the charm that is being integrated with.

  • remote_app – The name of the application that is being integrated with. To add a peer relation, set to the name of this application.

  • app_data – If provided, also add a new unit to the relation (triggering relation-joined) and set the application relation data (triggering relation-changed).

  • unit_data – If provided, also add a new unit to the relation (triggering relation-joined) and set the unit relation data (triggering relation-changed).

Returns:

The ID of the relation created.

add_relation_unit(relation_id: int, remote_unit_name: str) None[source]

Add a new unit to a relation.

This will trigger a relation_joined event. This would naturally be followed by a relation_changed event, which can be triggered with update_relation_data(). This separation is artificial in the sense that Juju will always fire the two, but is intended to make testing relations and their data bags slightly more natural.

Unless finer-grained control is needed, most charm tests can call add_relation() with the app_data or unit_data argument instead of using this function.

Example:

rel_id = harness.add_relation('db', 'postgresql')
harness.add_relation_unit(rel_id, 'postgresql/0')
Parameters:
  • relation_id – The integer relation identifier (as returned by add_relation()).

  • remote_unit_name – A string representing the remote unit that is being added.

add_resource(resource_name: str, content: AnyStr) None[source]

Add content for a resource to the backend.

This will register the content, so that a call to model.resources.fetch(resource_name) will return a path to a file containing that content.

Parameters:
  • resource_name – The name of the resource being added

  • content – Either string or bytes content, which will be the content of the filename returned by resource-get. If contents is a string, it will be encoded in utf-8

add_storage(storage_name: str, count: int = 1, *, attach: bool = False) List[str][source]

Create a new storage device and attach it to this unit.

To have repeatable tests, each device will be initialized with location set to /[tmpdir]/<storage_name>N, where N is the counter and will be a number from [0,total_num_disks-1].

The test harness uses symbolic links to imitate storage mounts, which may lead to some inconsistencies compared to the actual charm.

Parameters:
  • storage_name – The storage backend name on the Charm

  • count – Number of disks being added

  • attach – True to also attach the storage mount; if begin() has been called a True value will also emit storage-attached

Returns:

A list of storage IDs, e.g. [“my-storage/1”, “my-storage/2”].

attach_storage(storage_id: str) None[source]

Attach a storage device.

The intent of this function is to simulate a juju attach-storage call. If called after begin() and hooks are not disabled, it will trigger a storage-attached hook if the storage unit in question exists and is presently marked as detached.

The test harness uses symbolic links to imitate storage mounts, which may lead to some inconsistencies compared to the actual charm.

Parameters:

storage_id – The full storage ID of the storage unit being attached, including the storage key, e.g. my-storage/0.

begin() None[source]

Instantiate the Charm and start handling events.

Before calling begin(), there is no Charm instance, so changes to the Model won’t emit events. Call begin() for charm to be valid.

Should only be called once.

begin_with_initial_hooks() None[source]

Fire the same hooks that Juju would fire at startup.

This triggers install, relation-created, config-changed, start, pebble-ready (for any containers), and any relation-joined hooks based on what relations have been added before begin was called. Note that all of these are fired before returning control to the test suite, so to introspect what happens at each step, fire them directly (for example, Charm.on.install.emit()).

To use this with all the normal hooks, instantiate the harness, setup any relations that should be active when the charm starts, and then call this method. This method will automatically create and add peer relations that are specified in metadata.yaml.

If the charm metadata specifies containers, this sets can_connect to True for all containers (in addition to triggering pebble-ready for each).

Example:

harness = Harness(MyCharm)
# Do initial setup here
# Add storage if needed before begin_with_initial_hooks() is called
storage_ids = harness.add_storage('data', count=1)[0]
storage_id = storage_id[0] # we only added one storage instance
harness.add_relation('db', 'postgresql', unit_data={'key': 'val'})
harness.set_leader(True)
harness.update_config({'initial': 'config'})
harness.begin_with_initial_hooks()
# This will cause
# install, db-relation-created('postgresql'), leader-elected, config-changed, start
# db-relation-joined('postgresql/0'), db-relation-changed('postgresql/0')
# To be fired.
property charm: CharmType

Return the instance of the charm class that was passed to __init__.

Note that the Charm is not instantiated until begin() is called. Until then, attempting to access this property will raise an exception.

cleanup() None[source]

Called by the test infrastructure to clean up any temporary directories/files/etc.

Always call self.addCleanup(harness.cleanup) after creating a Harness.

container_pebble_ready(container_name: str)[source]

Fire the pebble_ready hook for the associated container.

This will switch the given container’s can_connect state to True before the hook function is called.

It will do nothing if begin() has not been called.

detach_storage(storage_id: str) None[source]

Detach a storage device.

The intent of this function is to simulate a juju detach-storage call. It will trigger a storage-detaching hook if the storage unit in question exists and is presently marked as attached.

Note that the Charm is not instantiated until begin() is called. Until then, attempting to use this method will raise an exception.

Parameters:

storage_id – The full storage ID of the storage unit being detached, including the storage key, e.g. my-storage/0.

disable_hooks() None[source]

Stop emitting hook events when the model changes.

This can be used by developers to stop changes to the model from emitting events that the charm will react to. Call enable_hooks() to re-enable them.

enable_hooks() None[source]

Re-enable hook events from charm.on when the model is changed.

By default, hook events are enabled once begin() is called, but if disable_hooks() is used, this method will enable them again.

evaluate_status() None[source]

Trigger the collect-status events and set application and/or unit status.

This will always trigger collect_unit_status, and set the unit status if any statuses were added.

If running on the leader unit (set_leader() has been called with True), this will trigger collect_app_status, and set the application status if any statuses were added.

Tests should normally call this and then assert that self.model.app.status or self.model.unit.status is the value expected.

Evaluation is not “additive”; this method resets the added statuses before triggering each collect-status event.

property framework: Framework

Return the Framework that is being driven by this Harness.

get_container_pebble_plan(container_name: str) Plan[source]

Return the current plan that Pebble is executing for the given container.

Parameters:

container_name – The simple name of the associated container

Returns:

The Pebble plan for this container. Use Plan.to_yaml to get a string form for the content.

Raises:

KeyError – if no Pebble client exists for that container name (should only happen if container is not present in metadata.yaml).

get_filesystem_root(container: str | Container) Path[source]

Return the temp directory path harness will use to simulate the container filesystem.

In a real container runtime, each container has an isolated root filesystem. To simulate this behaviour, the testing harness manages a temporary directory for each container. Any Pebble filesystem API calls will be translated and mapped to this directory, as if the directory was the container’s filesystem root.

This process is quite similar to the chroot command. Charm tests should treat the returned directory as the container’s root directory (/). The testing harness will not create any files or directories inside the simulated container’s root directory; it’s up to the test to populate the container’s root directory with any files or directories the charm needs.

Regarding the file ownership: unprivileged users are unable to create files with distinct ownership. To circumvent this limitation, the testing harness maps all user and group options related to file operations to match the current user and group.

Example usage (the parameter harness in the test function is a pytest fixture that does setup/teardown, see Harness):

# charm.py
class ExampleCharm(ops.CharmBase):
    def __init__(self, *args):
        super().__init__(*args)
        self.framework.observe(self.on["mycontainer"].pebble_ready,
                               self._on_pebble_ready)

    def _on_pebble_ready(self, event: ops.PebbleReadyEvent):
        self.hostname = event.workload.pull("/etc/hostname").read()

# test_charm.py
def test_hostname(harness):
    root = harness.get_filesystem_root("mycontainer")
    (root / "etc").mkdir()
    (root / "etc" / "hostname").write_text("hostname.example.com")
    harness.begin_with_initial_hooks()
    assert harness.charm.hostname == "hostname.example.com"
Parameters:

container – The name of the container or the container instance.

Returns:

The path of the temporary directory associated with the specified container.

get_pod_spec() Tuple[Mapping[Any, Any], Mapping[Any, Any]][source]

Return the content of the pod spec as last set by the charm.

This returns both the pod spec and any k8s_resources that were supplied. See the signature of Pod.set_spec.

get_relation_data(relation_id: int, app_or_unit: str | Application | Unit) Mapping[str, str][source]

Get the relation data bucket for a single app or unit in a given relation.

This ignores all of the safety checks of who can and can’t see data in relations (eg, non-leaders can’t read their own application’s relation data because there are no events that keep that data up-to-date for the unit).

Parameters:
  • relation_id – The relation whose content we want to look at.

  • app_or_unit – An Application or Unit instance, or its name, whose data we want to read.

Returns:

A dict containing the relation data for app_or_unit or None.

Raises:

KeyError – if relation_id doesn’t exist

get_secret_grants(secret_id: str, relation_id: int) Set[str][source]

Return the set of app and unit names granted to secret for this relation.

Parameters:
  • secret_id – The ID of the secret to get grants for.

  • relation_id – The ID of the relation granted access.

get_secret_revisions(secret_id: str) List[int][source]

Return the list of revision IDs for the given secret, oldest first.

Parameters:

secret_id – The ID of the secret to get revisions for.

get_workload_version() str[source]

Read the workload version that was set by the unit.

grant_secret(secret_id: str, observer: str | Application | Unit)[source]

Grant read access to this secret for the given observer application or unit.

If the given application or unit has already been granted access to this secret, do nothing.

Parameters:
  • secret_id – The ID of the secret to grant access to. This should normally be the return value of add_model_secret().

  • observer – The name of the application (or specific unit) to grant access to. A relation between this application and the charm under test must already have been created.

handle_exec(container: str | Container, command_prefix: Sequence[str], *, handler: Callable[[ExecArgs], None | ExecResult] | None = None, result: int | str | bytes | ExecResult | None = None)[source]

Register a handler to simulate the Pebble command execution.

This allows a test harness to simulate the behavior of running commands in a container. When ops.Container.exec() is triggered, the registered handler is used to generate stdout and stderr for the simulated execution.

A handler or a result may be provided, but not both:

  • A handler is a function accepting ops.testing.ExecArgs and returning ops.testing.ExecResult as the simulated process outcome. For cases that have side effects but don’t return output, the handler can return None, which is equivalent to returning ExecResult().

  • A result is for simulations that don’t need to inspect the exec arguments; the output or exit code is provided directly. Setting result to str or bytes means use that string as stdout (with exit code 0); setting result to int means return that exit code (and no stdout).

If handle_exec is called more than once with overlapping command prefixes, the longest match takes precedence. The registration of an execution handler can be updated by re-registering with the same command prefix.

The execution handler receives the timeout value in the ExecArgs. If needed, it can raise a TimeoutError to inform the harness that a timeout occurred.

If ops.Container.exec() is called with combine_stderr=True, the execution handler should, if required, weave the simulated standard error into the standard output. The harness checks the result and will raise an exception if stderr is non-empty.

Parameters:
  • container – The specified container or its name.

  • command_prefix – The command prefix to register against.

  • handler – A handler function that simulates the command’s execution.

  • result – A simplified form to specify the command’s simulated result.

Example usage:

# produce no output and return 0 for every command
harness.handle_exec('container', [], result=0)

# simple example that just produces output (exit code 0)
harness.handle_exec('webserver', ['ls', '/etc'], result='passwd\nprofile\n')

# slightly more complex (use stdin)
harness.handle_exec(
    'c1', ['sha1sum'],
    handler=lambda args: ExecResult(stdout=hashlib.sha1(args.stdin).hexdigest()))

# more complex example using args.command
def docker_handler(args: testing.ExecArgs) -> testing.ExecResult:
    match args.command:
        case ['docker', 'run', image]:
            return testing.ExecResult(stdout=f'running {image}')
        case ['docker', 'ps']:
            return testing.ExecResult(stdout='CONTAINER ID   IMAGE ...')
        case _:
            return testing.ExecResult(exit_code=1, stderr='unknown command')

harness.handle_exec('database', ['docker'], handler=docker_handler)

# handle timeout
def handle_timeout(args: testing.ExecArgs) -> int:
    if args.timeout is not None and args.timeout < 10:
        raise TimeoutError
    return 0

harness.handle_exec('database', ['foo'], handler=handle_timeout)
hooks_disabled()[source]

A context manager to run code with hooks disabled.

Example:

with harness.hooks_disabled():
    # things in here don't fire events
    harness.set_leader(True)
    harness.update_config(unset=['foo', 'bar'])
# things here will again fire events
property model: Model

Return the Model that is being driven by this Harness.

pebble_notify(container_name: str, key: str, *, data: Dict[str, str] | None = None, repeat_after: timedelta | None = None, type: NoticeType = NoticeType.CUSTOM) str[source]

Record a Pebble notice with the specified key and data.

If begin() has been called and the notice is new or was repeated, this will trigger a notice event of the appropriate type, for example ops.PebbleCustomNoticeEvent.

Parameters:
  • container_name – Name of workload container.

  • key – Notice key; must be in “example.com/path” format.

  • data – Data fields for this notice.

  • repeat_after – Only allow this notice to repeat after this duration has elapsed (the default is to always repeat).

  • type – Notice type (currently only “custom” notices are supported).

Returns:

The notice’s ID.

populate_oci_resources() None[source]

Populate all OCI resources.

property reboot_count: int

Number of times the charm has called ops.Unit.reboot().

remove_relation(relation_id: int) None[source]

Remove a relation.

Parameters:

relation_id – The relation ID for the relation to be removed.

Raises:

RelationNotFoundError – if relation id is not valid

remove_relation_unit(relation_id: int, remote_unit_name: str) None[source]

Remove a unit from a relation.

Example:

rel_id = harness.add_relation('db', 'postgresql')
harness.add_relation_unit(rel_id, 'postgresql/0')
...
harness.remove_relation_unit(rel_id, 'postgresql/0')

This will trigger a relation_departed event. This would normally be followed by a relation_changed event triggered by Juju. However, when using the test harness, a relation_changed event must be triggered using update_relation_data(). This deviation from normal Juju behaviour facilitates testing by making each step in the charm life cycle explicit.

Parameters:
  • relation_id – The integer relation identifier (as returned by add_relation()).

  • remote_unit_name – A string representing the remote unit that is being removed.

remove_storage(storage_id: str) None[source]

Detach a storage device.

The intent of this function is to simulate a juju remove-storage call. It will trigger a storage-detaching hook if the storage unit in question exists and is presently marked as attached. Then it will remove the storage unit from the testing backend.

Parameters:

storage_id – The full storage ID of the storage unit being removed, including the storage key, e.g. my-storage/0.

Raises:

RuntimeError – if the storage is not in the metadata.

reset_planned_units() None[source]

Reset the planned units override.

This allows the harness to fall through to the built in methods that will try to guess at a value for planned units, based on the number of peer relations that have been setup in the testing harness.

revoke_secret(secret_id: str, observer: str | Application | Unit)[source]

Revoke read access to this secret for the given observer application or unit.

If the given application or unit does not have access to this secret, do nothing.

Parameters:
  • secret_id – The ID of the secret to revoke access for. This should normally be the return value of add_model_secret().

  • observer – The name of the application (or specific unit) to revoke access to. A relation between this application and the charm under test must have already been created.

run_action(action_name: str, params: Dict[str, Any] | None = None) ActionOutput[source]

Simulates running a charm action, as with juju run.

Use this only after calling begin().

Validates that no required parameters are missing, and that additional parameters are not provided if that is not permitted. Does not validate the types of the parameters - you can use the jsonschema package to do this in your tests; for example:

schema = harness.charm.meta.actions["action-name"].parameters
try:
    jsonschema.validate(instance=params, schema=schema)
except jsonschema.ValidationError:
    # Do something about the invalid params.
    ...
harness.run_action("action-name", params)
Parameters:
  • action_name – the name of the action to run, as found in actions.yaml.

  • params – override the default parameter values found in actions.yaml. If a parameter is not in params, or params is None, then the default value from actions.yaml will be used.

Raises:

ActionFailed – if ops.ActionEvent.fail() is called. Note that this will be raised at the end of the run_action call, not immediately when fail() is called, to match the run-time behaviour.

set_can_connect(container: str | Container, val: bool)[source]

Change the simulated connection status of a container’s underlying Pebble client.

After calling this, ops.Container.can_connect() will return val.

set_cloud_spec(spec: CloudSpec)[source]

Set cloud specification (metadata) including credentials.

Call this method before the charm calls ops.Model.get_cloud_spec().

Example usage (the parameter harness in the test function is a pytest fixture that does setup/teardown, see Harness):

# charm.py
class MyVMCharm(ops.CharmBase):
    def __init__(self, framework: ops.Framework):
        super().__init__(framework)
        framework.observe(self.on.start, self._on_start)

    def _on_start(self, event: ops.StartEvent):
        self.cloud_spec = self.model.get_cloud_spec()

# test_charm.py
def test_start(harness):
    cloud_spec = ops.model.CloudSpec.from_dict({
        'name': 'localhost',
        'type': 'lxd',
        'endpoint': 'https://127.0.0.1:8443',
        'credential': {
            'auth-type': 'certificate',
            'attrs': {
                'client-cert': 'foo',
                'client-key': 'bar',
                'server-cert': 'baz'
            },
        },
    })
    harness.set_cloud_spec(cloud_spec)
    harness.begin()
    harness.charm.on.start.emit()
    assert harness.charm.cloud_spec == cloud_spec
set_leader(is_leader: bool = True) None[source]

Set whether this unit is the leader or not.

If this charm becomes a leader then leader_elected will be triggered. If begin() has already been called, then the charm’s peer relation should usually be added prior to calling this method (with add_relation()) to properly initialise and make available relation data that leader elected hooks may want to access.

Parameters:

is_leader – Whether this unit is the leader.

set_model_info(name: str | None = None, uuid: str | None = None) None[source]

Set the name and UUID of the model that this is representing.

Cannot be called once begin() has been called. Use it to set the value that will be returned by Model.name and Model.uuid.

This is a convenience method to invoke both set_model_name() and set_model_uuid() at once.

set_model_name(name: str) None[source]

Set the name of the Model that this is representing.

Cannot be called once begin() has been called. Use it to set the value that will be returned by Model.name.

set_model_uuid(uuid: str) None[source]

Set the uuid of the Model that this is representing.

Cannot be called once begin() has been called. Use it to set the value that will be returned by Model.uuid.

set_planned_units(num_units: int) None[source]

Set the number of “planned” units.

This is the value that Application.planned_units should return.

In real world circumstances, this number will be the number of units in the application. That is, this number will be the number of peers this unit has, plus one, as we count our own unit in the total.

A change to the return from planned_units will not generate an event. Typically, a charm author would check planned units during a config or install hook, or after receiving a peer relation joined event.

set_secret_content(secret_id: str, content: Dict[str, str])[source]

Update a secret’s content, add a new revision, and fire secret-changed.

Parameters:
  • secret_id – The ID of the secret to update. This should normally be the return value of add_model_secret().

  • content – A key-value mapping containing the new payload.

trigger_secret_expiration(secret_id: str, revision: int, *, label: str | None = None)[source]

Trigger a secret-expired event for the given secret.

This event is fired by Juju when a secret’s expiration time elapses, however, time-based events cannot be simulated appropriately in the harness, so this fires it manually.

Parameters:
  • secret_id – The ID of the secret associated with the event.

  • revision – Revision number to provide to the event. This should be an item from the list returned by get_secret_revisions().

  • label – Label value to send to the event. If None, the secret’s label is used.

trigger_secret_removal(secret_id: str, revision: int, *, label: str | None = None)[source]

Trigger a secret-remove event for the given secret and revision.

This event is fired by Juju for a specific revision when all the secret’s observers have refreshed to a later revision, however, in the harness call this method to fire the event manually.

Parameters:
  • secret_id – The ID of the secret associated with the event.

  • revision – Revision number to provide to the event. This should be an item from the list returned by get_secret_revisions().

  • label – Label value to send to the event. If None, the secret’s label is used.

trigger_secret_rotation(secret_id: str, *, label: str | None = None)[source]

Trigger a secret-rotate event for the given secret.

This event is fired by Juju when a secret’s rotation time elapses, however, time-based events cannot be simulated appropriately in the harness, so this fires it manually.

Parameters:
  • secret_id – The ID of the secret associated with the event.

  • label – Label value to send to the event. If None, the secret’s label is used.

update_config(key_values: Mapping[str, str | int | float | bool] | None = None, unset: Iterable[str] = ()) None[source]

Update the config as seen by the charm.

This will trigger a config_changed event.

Note that the key_values mapping will only add or update configuration items. To remove existing ones, see the unset parameter.

Parameters:
  • key_values – A Mapping of key:value pairs to update in config.

  • unset – An iterable of keys to remove from config. This sets the value to the default if defined, otherwise removes the key altogether.

Raises:

ValueError – if the key is not present in the config.

update_relation_data(relation_id: int, app_or_unit: str, key_values: Mapping[str, str]) None[source]

Update the relation data for a given unit or application in a given relation.

This also triggers the relation_changed event for the given relation_id.

Unless finer-grained control is needed, most charm tests can call add_relation() with the app_data or unit_data argument instead of using this function.

Parameters:
  • relation_id – The integer relation ID representing this relation.

  • app_or_unit – The unit or application name that is being updated. This can be the local or remote application.

  • key_values – Each key/value will be updated in the relation data.

Indices