Source code for juju.client._client1

# DO NOT CHANGE THIS FILE! This file is auto-generated by facade.py.
# Changes will be overwritten/lost when the file is regenerated.

from juju.client._definitions import *
from juju.client.facade import ReturnMapping, Type


[docs]class CredentialManagerFacade(Type): name = "CredentialManager" version = 1
[docs] @ReturnMapping(ErrorResult) async def InvalidateModelCredential(self, reason=None): """InvalidateModelCredential marks the cloud credential for this model as invalid. reason : str Returns -> ErrorResult """ if reason is not None and not isinstance(reason, (bytes, str)): raise Exception(f"Expected reason to be a str, received: {type(reason)}") # map input types to rpc msg _params = dict() msg = dict( type="CredentialManager", request="InvalidateModelCredential", version=1, params=_params, ) _params["reason"] = reason reply = await self.rpc(msg) return reply
[docs]class FirewallRulesFacade(Type): name = "FirewallRules" version = 1
[docs] @ReturnMapping(ListFirewallRulesResults) async def ListFirewallRules(self): """ListFirewallRules returns all the firewall rules. Returns -> ListFirewallRulesResults """ # map input types to rpc msg _params = dict() msg = dict( type="FirewallRules", request="ListFirewallRules", version=1, params=_params ) reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def SetFirewallRules(self, args=None): """SetFirewallRules creates or updates the specified firewall rules. args : typing.Sequence[~FirewallRule] Returns -> ErrorResults """ if args is not None and not isinstance(args, (bytes, str, list)): raise Exception(f"Expected args to be a Sequence, received: {type(args)}") # map input types to rpc msg _params = dict() msg = dict( type="FirewallRules", request="SetFirewallRules", version=1, params=_params ) _params["args"] = args reply = await self.rpc(msg) return reply
[docs]class ImageMetadataManagerFacade(Type): name = "ImageMetadataManager" version = 1
[docs] @ReturnMapping(ErrorResults) async def Delete(self, image_ids=None): """Delete deletes cloud image metadata for given image ids. It supports bulk calls. image_ids : typing.Sequence[str] Returns -> ErrorResults """ if image_ids is not None and not isinstance(image_ids, (bytes, str, list)): raise Exception( f"Expected image_ids to be a Sequence, received: {type(image_ids)}" ) # map input types to rpc msg _params = dict() msg = dict( type="ImageMetadataManager", request="Delete", version=1, params=_params ) _params["image-ids"] = image_ids reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ListCloudImageMetadataResult) async def List( self, arches=None, region=None, root_storage_type=None, stream=None, versions=None, virt_type=None, ): """List returns all found cloud image metadata that satisfy given filter. Returned list contains metadata ordered by priority. arches : typing.Sequence[str] region : str root_storage_type : str stream : str versions : typing.Sequence[str] virt_type : str Returns -> ListCloudImageMetadataResult """ if arches is not None and not isinstance(arches, (bytes, str, list)): raise Exception( f"Expected arches to be a Sequence, received: {type(arches)}" ) if region is not None and not isinstance(region, (bytes, str)): raise Exception(f"Expected region to be a str, received: {type(region)}") if root_storage_type is not None and not isinstance( root_storage_type, (bytes, str) ): raise Exception( f"Expected root_storage_type to be a str, received: {type(root_storage_type)}" ) if stream is not None and not isinstance(stream, (bytes, str)): raise Exception(f"Expected stream to be a str, received: {type(stream)}") if versions is not None and not isinstance(versions, (bytes, str, list)): raise Exception( f"Expected versions to be a Sequence, received: {type(versions)}" ) if virt_type is not None and not isinstance(virt_type, (bytes, str)): raise Exception( f"Expected virt_type to be a str, received: {type(virt_type)}" ) # map input types to rpc msg _params = dict() msg = dict( type="ImageMetadataManager", request="List", version=1, params=_params ) _params["arches"] = arches _params["region"] = region _params["root-storage-type"] = root_storage_type _params["stream"] = stream _params["versions"] = versions _params["virt-type"] = virt_type reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def Save(self, metadata=None): """Save stores given cloud image metadata. It supports bulk calls. metadata : typing.Sequence[~CloudImageMetadataList] Returns -> ErrorResults """ if metadata is not None and not isinstance(metadata, (bytes, str, list)): raise Exception( f"Expected metadata to be a Sequence, received: {type(metadata)}" ) # map input types to rpc msg _params = dict() msg = dict( type="ImageMetadataManager", request="Save", version=1, params=_params ) _params["metadata"] = metadata reply = await self.rpc(msg) return reply
[docs]class KeyManagerFacade(Type): name = "KeyManager" version = 1
[docs] @ReturnMapping(ErrorResults) async def AddKeys(self, ssh_keys=None, user=None): """AddKeys adds new authorised ssh keys for the specified user. ssh_keys : typing.Sequence[str] user : str Returns -> ErrorResults """ if ssh_keys is not None and not isinstance(ssh_keys, (bytes, str, list)): raise Exception( f"Expected ssh_keys to be a Sequence, received: {type(ssh_keys)}" ) if user is not None and not isinstance(user, (bytes, str)): raise Exception(f"Expected user to be a str, received: {type(user)}") # map input types to rpc msg _params = dict() msg = dict(type="KeyManager", request="AddKeys", version=1, params=_params) _params["ssh-keys"] = ssh_keys _params["user"] = user reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def DeleteKeys(self, ssh_keys=None, user=None): """DeleteKeys deletes the authorised ssh keys for the specified user. ssh_keys : typing.Sequence[str] user : str Returns -> ErrorResults """ if ssh_keys is not None and not isinstance(ssh_keys, (bytes, str, list)): raise Exception( f"Expected ssh_keys to be a Sequence, received: {type(ssh_keys)}" ) if user is not None and not isinstance(user, (bytes, str)): raise Exception(f"Expected user to be a str, received: {type(user)}") # map input types to rpc msg _params = dict() msg = dict(type="KeyManager", request="DeleteKeys", version=1, params=_params) _params["ssh-keys"] = ssh_keys _params["user"] = user reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def ImportKeys(self, ssh_keys=None, user=None): """ImportKeys imports new authorised ssh keys from the specified key ids for the specified user. ssh_keys : typing.Sequence[str] user : str Returns -> ErrorResults """ if ssh_keys is not None and not isinstance(ssh_keys, (bytes, str, list)): raise Exception( f"Expected ssh_keys to be a Sequence, received: {type(ssh_keys)}" ) if user is not None and not isinstance(user, (bytes, str)): raise Exception(f"Expected user to be a str, received: {type(user)}") # map input types to rpc msg _params = dict() msg = dict(type="KeyManager", request="ImportKeys", version=1, params=_params) _params["ssh-keys"] = ssh_keys _params["user"] = user reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(StringsResults) async def ListKeys(self, entities=None, mode=None): """ListKeys returns the authorised ssh keys for the specified users. entities : Entities mode : bool Returns -> StringsResults """ if entities is not None and not isinstance(entities, (dict, Entities)): raise Exception( f"Expected entities to be a Entities, received: {type(entities)}" ) if mode is not None and not isinstance(mode, bool): raise Exception(f"Expected mode to be a bool, received: {type(mode)}") # map input types to rpc msg _params = dict() msg = dict(type="KeyManager", request="ListKeys", version=1, params=_params) _params["entities"] = entities _params["mode"] = mode reply = await self.rpc(msg) return reply
[docs]class ModelUpgraderFacade(Type): name = "ModelUpgrader" version = 1
[docs] @ReturnMapping(None) async def AbortModelUpgrade(self, model_tag=None): """AbortModelUpgrade aborts and archives the model upgrade synchronisation record, if any. model_tag : str Returns -> None """ if model_tag is not None and not isinstance(model_tag, (bytes, str)): raise Exception( f"Expected model_tag to be a str, received: {type(model_tag)}" ) # map input types to rpc msg _params = dict() msg = dict( type="ModelUpgrader", request="AbortModelUpgrade", version=1, params=_params ) _params["model-tag"] = model_tag reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(UpgradeModelResult) async def UpgradeModel( self, agent_stream=None, dry_run=None, ignore_agent_versions=None, model_tag=None, target_version=None, ): """UpgradeModel upgrades a model. agent_stream : str dry_run : bool ignore_agent_versions : bool model_tag : str target_version : Number Returns -> UpgradeModelResult """ if agent_stream is not None and not isinstance(agent_stream, (bytes, str)): raise Exception( f"Expected agent_stream to be a str, received: {type(agent_stream)}" ) if dry_run is not None and not isinstance(dry_run, bool): raise Exception(f"Expected dry_run to be a bool, received: {type(dry_run)}") if ignore_agent_versions is not None and not isinstance( ignore_agent_versions, bool ): raise Exception( f"Expected ignore_agent_versions to be a bool, received: {type(ignore_agent_versions)}" ) if model_tag is not None and not isinstance(model_tag, (bytes, str)): raise Exception( f"Expected model_tag to be a str, received: {type(model_tag)}" ) if target_version is not None and not isinstance( target_version, (dict, Number) ): raise Exception( f"Expected target_version to be a Number, received: {type(target_version)}" ) # map input types to rpc msg _params = dict() msg = dict( type="ModelUpgrader", request="UpgradeModel", version=1, params=_params ) _params["agent-stream"] = agent_stream _params["dry-run"] = dry_run _params["ignore-agent-versions"] = ignore_agent_versions _params["model-tag"] = model_tag _params["target-version"] = target_version reply = await self.rpc(msg) return reply
[docs]class PayloadsFacade(Type): name = "Payloads" version = 1
[docs] @ReturnMapping(PayloadListResults) async def List(self, patterns=None): """List builds the list of payloads being tracked for the given unit and IDs. If no IDs are provided then all tracked payloads for the unit are returned. patterns : typing.Sequence[str] Returns -> PayloadListResults """ if patterns is not None and not isinstance(patterns, (bytes, str, list)): raise Exception( f"Expected patterns to be a Sequence, received: {type(patterns)}" ) # map input types to rpc msg _params = dict() msg = dict(type="Payloads", request="List", version=1, params=_params) _params["patterns"] = patterns reply = await self.rpc(msg) return reply
[docs]class PingerFacade(Type): name = "Pinger" version = 1
[docs] @ReturnMapping(None) async def Ping(self): """Returns -> None""" # map input types to rpc msg _params = dict() msg = dict(type="Pinger", request="Ping", version=1, params=_params) reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(None) async def Stop(self): """Returns -> None""" # map input types to rpc msg _params = dict() msg = dict(type="Pinger", request="Stop", version=1, params=_params) reply = await self.rpc(msg) return reply
[docs]class SecretBackendsFacade(Type): name = "SecretBackends" version = 1
[docs] @ReturnMapping(ErrorResults) async def AddSecretBackends(self, args=None): """AddSecretBackends adds new secret backends. args : typing.Sequence[~AddSecretBackendArg] Returns -> ErrorResults """ if args is not None and not isinstance(args, (bytes, str, list)): raise Exception(f"Expected args to be a Sequence, received: {type(args)}") # map input types to rpc msg _params = dict() msg = dict( type="SecretBackends", request="AddSecretBackends", version=1, params=_params, ) _params["args"] = args reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ListSecretBackendsResults) async def ListSecretBackends(self, names=None, reveal=None): """ListSecretBackends lists available secret backends. names : typing.Sequence[str] reveal : bool Returns -> ListSecretBackendsResults """ if names is not None and not isinstance(names, (bytes, str, list)): raise Exception(f"Expected names to be a Sequence, received: {type(names)}") if reveal is not None and not isinstance(reveal, bool): raise Exception(f"Expected reveal to be a bool, received: {type(reveal)}") # map input types to rpc msg _params = dict() msg = dict( type="SecretBackends", request="ListSecretBackends", version=1, params=_params, ) _params["names"] = names _params["reveal"] = reveal reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def RemoveSecretBackends(self, args=None): """RemoveSecretBackends removes secret backends. args : typing.Sequence[~RemoveSecretBackendArg] Returns -> ErrorResults """ if args is not None and not isinstance(args, (bytes, str, list)): raise Exception(f"Expected args to be a Sequence, received: {type(args)}") # map input types to rpc msg _params = dict() msg = dict( type="SecretBackends", request="RemoveSecretBackends", version=1, params=_params, ) _params["args"] = args reply = await self.rpc(msg) return reply
[docs] @ReturnMapping(ErrorResults) async def UpdateSecretBackends(self, args=None): """UpdateSecretBackends updates secret backends. args : typing.Sequence[~UpdateSecretBackendArg] Returns -> ErrorResults """ if args is not None and not isinstance(args, (bytes, str, list)): raise Exception(f"Expected args to be a Sequence, received: {type(args)}") # map input types to rpc msg _params = dict() msg = dict( type="SecretBackends", request="UpdateSecretBackends", version=1, params=_params, ) _params["args"] = args reply = await self.rpc(msg) return reply
[docs]class SecretsFacade(Type): name = "Secrets" version = 1
[docs] @ReturnMapping(ListSecretResults) async def ListSecrets(self, filter_=None, show_secrets=None): """ListSecrets lists available secrets. filter_ : SecretsFilter show_secrets : bool Returns -> ListSecretResults """ if filter_ is not None and not isinstance(filter_, (dict, SecretsFilter)): raise Exception( f"Expected filter_ to be a SecretsFilter, received: {type(filter_)}" ) if show_secrets is not None and not isinstance(show_secrets, bool): raise Exception( f"Expected show_secrets to be a bool, received: {type(show_secrets)}" ) # map input types to rpc msg _params = dict() msg = dict(type="Secrets", request="ListSecrets", version=1, params=_params) _params["filter"] = filter_ _params["show-secrets"] = show_secrets reply = await self.rpc(msg) return reply