# 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 ApplicationOffersFacade(Type):
name = "ApplicationOffers"
version = 5
[docs] @ReturnMapping(ApplicationOffersResults)
async def ApplicationOffers(self, bakery_version=None, offer_urls=None):
"""ApplicationOffers gets details about remote applications that match given URLs.
bakery_version : int
offer_urls : typing.Sequence[str]
Returns -> ApplicationOffersResults
"""
if bakery_version is not None and not isinstance(bakery_version, int):
raise Exception(
f"Expected bakery_version to be a int, received: {type(bakery_version)}"
)
if offer_urls is not None and not isinstance(offer_urls, (bytes, str, list)):
raise Exception(
f"Expected offer_urls to be a Sequence, received: {type(offer_urls)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="ApplicationOffers",
version=5,
params=_params,
)
_params["bakery-version"] = bakery_version
_params["offer-urls"] = offer_urls
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(ErrorResults)
async def DestroyOffers(self, force=None, offer_urls=None):
"""DestroyOffers removes the offers specified by the given URLs, forcing if necessary.
force : bool
offer_urls : typing.Sequence[str]
Returns -> ErrorResults
"""
if force is not None and not isinstance(force, bool):
raise Exception(f"Expected force to be a bool, received: {type(force)}")
if offer_urls is not None and not isinstance(offer_urls, (bytes, str, list)):
raise Exception(
f"Expected offer_urls to be a Sequence, received: {type(offer_urls)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers", request="DestroyOffers", version=5, params=_params
)
_params["force"] = force
_params["offer-urls"] = offer_urls
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(QueryApplicationOffersResultsV5)
async def FindApplicationOffers(self, filters=None):
"""FindApplicationOffers gets details about remote applications that match given filter.
filters : typing.Sequence[~OfferFilter]
Returns -> QueryApplicationOffersResultsV5
"""
if filters is not None and not isinstance(filters, (bytes, str, list)):
raise Exception(
f"Expected filters to be a Sequence, received: {type(filters)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="FindApplicationOffers",
version=5,
params=_params,
)
_params["Filters"] = filters
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(ConsumeOfferDetailsResults)
async def GetConsumeDetails(self, offer_urls=None, user_tag=None):
"""GetConsumeDetails returns the details necessary to pass to another model
to allow the specified args user to consume the offers represented by the args URLs.
offer_urls : OfferURLs
user_tag : str
Returns -> ConsumeOfferDetailsResults
"""
if offer_urls is not None and not isinstance(offer_urls, (dict, OfferURLs)):
raise Exception(
f"Expected offer_urls to be a OfferURLs, received: {type(offer_urls)}"
)
if user_tag is not None and not isinstance(user_tag, (bytes, str)):
raise Exception(
f"Expected user_tag to be a str, received: {type(user_tag)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="GetConsumeDetails",
version=5,
params=_params,
)
_params["offer-urls"] = offer_urls
_params["user-tag"] = user_tag
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(QueryApplicationOffersResultsV5)
async def ListApplicationOffers(self, filters=None):
"""ListApplicationOffers gets deployed details about application offers that match given filter.
The results contain details about the deployed applications such as connection count.
filters : typing.Sequence[~OfferFilter]
Returns -> QueryApplicationOffersResultsV5
"""
if filters is not None and not isinstance(filters, (bytes, str, list)):
raise Exception(
f"Expected filters to be a Sequence, received: {type(filters)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="ListApplicationOffers",
version=5,
params=_params,
)
_params["Filters"] = filters
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(ErrorResults)
async def ModifyOfferAccess(self, changes=None):
"""ModifyOfferAccess changes the application offer access granted to users.
changes : typing.Sequence[~ModifyOfferAccess]
Returns -> ErrorResults
"""
if changes is not None and not isinstance(changes, (bytes, str, list)):
raise Exception(
f"Expected changes to be a Sequence, received: {type(changes)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="ModifyOfferAccess",
version=5,
params=_params,
)
_params["changes"] = changes
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(ErrorResults)
async def Offer(self, offers=None):
"""Offer makes application endpoints available for consumption at a specified URL.
offers : typing.Sequence[~AddApplicationOffer]
Returns -> ErrorResults
"""
if offers is not None and not isinstance(offers, (bytes, str, list)):
raise Exception(
f"Expected offers to be a Sequence, received: {type(offers)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(type="ApplicationOffers", request="Offer", version=5, params=_params)
_params["Offers"] = offers
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(RemoteApplicationInfoResults)
async def RemoteApplicationInfo(self, bakery_version=None, offer_urls=None):
"""RemoteApplicationInfo returns information about the requested remote application.
This call currently has no client side API, only there for the Dashboard at this stage.
bakery_version : int
offer_urls : typing.Sequence[str]
Returns -> RemoteApplicationInfoResults
"""
if bakery_version is not None and not isinstance(bakery_version, int):
raise Exception(
f"Expected bakery_version to be a int, received: {type(bakery_version)}"
)
if offer_urls is not None and not isinstance(offer_urls, (bytes, str, list)):
raise Exception(
f"Expected offer_urls to be a Sequence, received: {type(offer_urls)}"
)
# map input types to rpc msg
_params = dict()
msg = dict(
type="ApplicationOffers",
request="RemoteApplicationInfo",
version=5,
params=_params,
)
_params["bakery-version"] = bakery_version
_params["offer-urls"] = offer_urls
reply = await self.rpc(msg)
return reply
[docs]class SubnetsFacade(Type):
name = "Subnets"
version = 5
[docs] @ReturnMapping(ZoneResults)
async def AllZones(self):
"""AllZones returns all availability zones known to Juju. If a
zone is unusable, unavailable, or deprecated the Available
field will be false.
Returns -> ZoneResults
"""
# map input types to rpc msg
_params = dict()
msg = dict(type="Subnets", request="AllZones", version=5, params=_params)
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(ListSubnetsResults)
async def ListSubnets(self, space_tag=None, zone=None):
"""ListSubnets returns the matching subnets after applying
optional filters.
space_tag : str
zone : str
Returns -> ListSubnetsResults
"""
if space_tag is not None and not isinstance(space_tag, (bytes, str)):
raise Exception(
f"Expected space_tag to be a str, received: {type(space_tag)}"
)
if zone is not None and not isinstance(zone, (bytes, str)):
raise Exception(f"Expected zone to be a str, received: {type(zone)}")
# map input types to rpc msg
_params = dict()
msg = dict(type="Subnets", request="ListSubnets", version=5, params=_params)
_params["space-tag"] = space_tag
_params["zone"] = zone
reply = await self.rpc(msg)
return reply
[docs] @ReturnMapping(SubnetsResults)
async def SubnetsByCIDR(self, cidrs=None):
"""SubnetsByCIDR returns the collection of subnets matching each CIDR in the input.
cidrs : typing.Sequence[str]
Returns -> SubnetsResults
"""
if cidrs is not None and not isinstance(cidrs, (bytes, str, list)):
raise Exception(f"Expected cidrs to be a Sequence, received: {type(cidrs)}")
# map input types to rpc msg
_params = dict()
msg = dict(type="Subnets", request="SubnetsByCIDR", version=5, params=_params)
_params["cidrs"] = cidrs
reply = await self.rpc(msg)
return reply