juju.utils
Summary
Reference
- class juju.utils.IdQueue[source]
Bases:
objectWrapper around asyncio.Queue that maintains a separate queue for each ID.
- class juju.utils.RegistrationInfo(**kwargs)[source]
Bases:
SequenceASN.1 representation of:
type RegistrationInfo struct { User string
Addrs []string
SecretKey []byte
ControllerName string }
- juju.utils.base_channel_to_series(channel: str) str[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: Callable[[], bool], timeout: float | None = None, wait_period: float = 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: Callable[[], Awaitable[bool]], timeout: float | None = None, wait_period: float = 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) bool[source]
Wrapper around asyncio.create_subprocess_exec.
- juju.utils.generate_user_controller_access_token(username: str, controller_endpoints, secret_key: str, controller_name)[source]
Implement in python what is currently done in GO.
https://github.com/juju/juju/blob/a5ab92e/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_local_charm_base(series: str, charm_path: str, base_class: type)[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 information
- juju.utils.get_local_charm_data(path: str | pathlib.Path, yaml_file: str) dict[str, Any][source]
Retrieve Metadata of a Charm from its path.
- Parameters:
path (str) – Path of charm directory or .charm file
yaml_file (str) – name of the yaml file, can be either
“metadata.yaml”, or “manifest.yaml”, or “charmcraft.yaml”
- Returns:
Object of charm metadata
- juju.utils.get_series_version(series_name: str) str[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: str) str[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() str[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() tuple[str, str][source]
Resolves and returns the path strings for public and private ssh keys for juju CLI.
- juju.utils.parse_base_arg(base: str) 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() str[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: Event, 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
Nonewill 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: str, supported_series: list[str]) str[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: str = '', charm_url=None, model_config=None, supported_series: list[str] = [], force: bool = False) str[source]
Select series to deploy on.
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: dict[str, str], existing_resources, arg_resources) bool[source]
Determine if the given resource should be upgraded.
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