ops.pebble¶

Client for the Pebble API.

This module provides a way to interact with Pebble, including:

Client; communicates directly with the Pebble API.
Layer class to define Pebble configuration layers, including:
- Check class to represent Pebble checks.
- LogTarget class to represent Pebble log targets.
- Service class to represent Pebble service descriptions.

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

See more: Pebble documentation

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.ChangeKind( value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None, )[source]¶

Bases: Enum

Enum of Pebble Change kinds.

EXEC = 'exec'¶

PERFORM_CHECK = 'perform-check'¶

RECOVER_CHECK = 'recover-check'¶

REPLAN = 'replan'¶

RESTART = 'restart'¶

START = 'start'¶

STOP = 'stop'¶

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.ChangeStatus( value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None, )[source]¶

Bases: Enum

Enum of Pebble Change statuses.

ABORT = 'Abort'¶: Abort status means the task should stop doing its activities and then undo.

DO = 'Do'¶: Do status means the change or task is ready to start.

DOING = 'Doing'¶: Doing status means the change or task is running or an attempt was made to run it.

DONE = 'Done'¶: Done status means the change or task was accomplished successfully.

ERROR = 'Error'¶: Error status means the change or task has errored out while running or being undone.

HOLD = 'Hold'¶: Hold status means the task should not run for the moment, perhaps as a consequence of an error on another task.

UNDO = 'Undo'¶: Undo status means the change or task should be undone, probably due to an error.

UNDOING = 'Undoing'¶: UndoingStatus means the change or task is being undone or an attempt was made to undo it.

WAIT = 'Wait'¶: Wait status means the task was accomplished successfully but some external event needs to happen before work can progress further.

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¶

startup: Literal['', 'enabled', 'disabled']¶

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, change_id: ChangeID | None = None, startup: CheckStartup = CheckStartup.ENABLED, )[source]¶

Bases: object

Check status information.

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

change_id: ChangeID | None¶

Change ID of perform-check or recover-check change driving this check.

This will be None on older versions of Pebble, which did not use changes to drive health 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.

startup: CheckStartup¶

Startup mode.

CheckStartup.ENABLED means the check will be started when added, and in a replan. CheckStartup.DISABLED means the check must be manually started.

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), and CheckStatus.INACTIVE means the check is not running.

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.CheckStartup( value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None, )[source]¶

Bases: Enum

Enum of check startup options.

DISABLED = 'disabled'¶

ENABLED = 'enabled'¶

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'¶

INACTIVE = 'inactive'¶

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.

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)

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.

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.

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 checks.

After requesting the replan, also wait for any impacted services 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.

Listed running services will be stopped and restarted, and listed stopped services will 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_checks(checks: Iterable[str]) → List[str][source]¶

Start checks by name.

Added in Juju version 3.6.4.

Parameters:: checks – Non-empty list of checks to start.
Returns:: Set of check names that were started. Checks that were already running will not be included.

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_checks(checks: Iterable[str]) → List[str][source]¶

Stop checks by name.

Added in Juju version 3.6.4.

Parameters:: checks – Non-empty list of checks to stop.
Returns:: Set of check names that were stopped. Checks that were already inactive will not be included.

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 stopped.

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:

ChangeError – if there was an error starting or running the process.
ExecError – if the process exits with a non-zero exit code.
TypeError – if Client.exec() was called with the stdout argument.

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'¶

SYMLINK = 'symlink'¶

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://documentation.ubuntu.com/pebble/reference/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', '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.

CHANGE_UPDATE = 'change-update'¶: Recorded whenever a change’s status is updated. The key for change-update notices is the change ID.

CUSTOM = 'custom'¶: A custom notice reported via the Pebble client API or pebble notify. The key and data fields are provided by the user. The key must be in the format example.com/path to ensure well-namespaced notice keys.

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://documentation.ubuntu.com/pebble/reference/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'¶