juju.utils

Summary

Reference

class juju.utils.Addrs(*args, **kwargs)[source]

Bases: SequenceOf

componentType = <PrintableString schema object, tagSet <TagSet object, tags 0:0:19>, encoding us-ascii>

Default PyAsn1Item derivative object representing ASN.1 type allowed within |ASN.1| type

class juju.utils.IdQueue(maxsize=0)[source]

Bases: object

Wrapper around asyncio.Queue that maintains a separate queue for each ID.

coroutine get(id)[source]
coroutine put(id, value)[source]
coroutine put_all(value)[source]
class juju.utils.RegistrationInfo(**kwargs)[source]

Bases: Sequence

ASN.1 representation of:

type RegistrationInfo struct { User string

Addrs []string

SecretKey []byte

ControllerName string

}

juju.utils.base_channel_from_series(track, risk, series)[source]
juju.utils.base_channel_to_series(channel)[source]

Returns the series string using the track inside the base channel

Parameters:

channel (str) – is track/risk (e.g. 20.04/stable)

Returns:

str series (e.g. focal)

coroutine juju.utils.block_until(*conditions, timeout=None, wait_period=0.5)[source]

Return only after all conditions are true. If a timeout occurs, it cancels the task and raises asyncio.TimeoutError.

coroutine juju.utils.block_until_with_coroutine(condition_coroutine, timeout=None, wait_period=0.5)[source]

Return only after the given coroutine returns True. If a timeout occurs, it cancels the task and raises asyncio.TimeoutError.

coroutine juju.utils.execute_process(*cmd, log=None)[source]

Wrapper around asyncio.create_subprocess_exec.

juju.utils.generate_user_controller_access_token(username, controller_endpoints, secret_key, controller_name)[source]

“ Implement in python what is currently done in GO https://github.com/juju/juju/blob/a5ab92ec9b7f5da3678d9ac603fe52d45af24412/cmd/juju/user/utils.go#L16

Parameters:
  • username – name of the user to register

  • controller_endpoints – juju controller endpoints list in the format <ip>:<port>

  • secret_key – base64 encoded string of the secret-key generated by juju

  • controller_name – name of the controller to register to.

juju.utils.get_base_from_origin_or_channel(origin_or_channel, series=None)[source]
juju.utils.get_local_charm_base(series, charm_path, base_class)[source]

Deduce the base [channel/osname] of a local charm based on what we know already

Parameters:
  • series (str) – This may come from the argument or the metadata.yaml

  • charm_path (str) – Path of charm directory/.charm file

  • base_class (class) –

Returns:

Instance of the baseCls with channel/osname informaiton

juju.utils.get_local_charm_charmcraft_yaml(path)[source]
juju.utils.get_local_charm_data(path, yaml_file)[source]

Retrieve Metadata of a Charm from its path

Patam str path:

Path of charm directory or .charm file

Patam str yaml_file:

name of the yaml file, can be either

“metadata.yaml”, or “manifest.yaml”, or “charmcraft.yaml”

Returns:

Object of charm metadata

juju.utils.get_local_charm_manifest(path)[source]
juju.utils.get_local_charm_metadata(path)[source]
juju.utils.get_os_from_series(series)[source]
juju.utils.get_series_version(series_name)[source]

get_series_version outputs the version of the OS based on the given series e.g. jammy -> 22.04, kubernetes -> kubernetes

Parameters:

series_name (str) – name of the series

Return str:

os version

juju.utils.get_version_series(version)[source]

get_version_series is the opposite of the get_series_version. It outputs the series based on given OS version

Parameters:

version (str) – version of the OS

return str: name of the series corresponding to the given version

juju.utils.juju_config_dir()[source]

Resolves and returns the path string to the juju configuration folder for the juju CLI tool. Of the following items, returns the first option that works (top to bottom):

  • $JUJU_DATA

  • $XDG_DATA_HOME/juju

  • ~/.local/share/juju

juju.utils.juju_ssh_key_paths()[source]

Resolves and returns the path strings for public and private ssh keys for juju CLI.

juju.utils.parse_base_arg(base)[source]

Parses a given base into a Client.Base object :param base str : The base to deploy a charm (e.g. ubuntu@22.04)

coroutine juju.utils.read_ssh_key()[source]

Attempt to read the local juju admin’s public ssh key, so that it can be passed on to a model.

coroutine juju.utils.run_with_interrupt(task, *events, log=None)[source]

Awaits a task while allowing it to be interrupted by one or more `asyncio.Event`s.

If the task finishes without the events becoming set, the results of the task will be returned. If the event become set, the task will be cancelled None will be returned.

Parameters:
  • task – Task to run

  • events – One or more asyncio.Event`s which, if set, will interrupt `task and cause it to be cancelled.

juju.utils.series_for_charm(requested_series, supported_series)[source]

series_for_charm takes a requested series and a list of series supported by a charm and returns the series which is relevant. If the requested series is empty, then the first supported series is used, otherwise the requested series is validated against the supported series.

juju.utils.series_selector(series_arg='', charm_url=None, model_config=None, supported_series=[], force=False)[source]

series_selector corresponds to the CharmSeries() in https://github.com/juju/juju/blob/develop/core/charm/series_selector.go

determines what series to use with a charm. Order of preference is: - user requested with –series or defined by bundle when deploying - user requested in charm’s url (e.g. juju deploy jammy/ubuntu) - model default, if set, acts like –series - default from charm metadata supported series / series in url - default LTS

juju.utils.should_upgrade_resource(available_resource, existing_resources, arg_resources={})[source]

Called in the context of upgrade_charm. Given a resource R, takes a look at the resources we already have and decides if we need to refresh R.

Parameters:

available_resource (dict[str]) – The dict representing the client.Resource coming from the

charmhub api. We’re considering if we need to refresh this during upgrade_charm. :param dict[str] existing_resources: The dict coming from resources_facade.ListResources representing the resources of the currently deployed charm. :param dict[str] arg_resources: user provided resources to be refreshed

Result bool:

The decision to refresh the given resource

juju.utils.user_requested(series_arg, supported_series, force)[source]
coroutine juju.utils.wait_for_bundle(model, bundle, **kwargs)[source]

Helper to wait for just the apps in a specific bundle.

Equivalent to loading the bundle, pulling out the app names, and calling:

await model.wait_for_idle(app_names, **kwargs)