1 from collections
import namedtuple
3 from .facade
import ReturnMapping
, Type
16 """A single websocket delta.
18 :ivar entity: The entity name, e.g. 'unit', 'application'
21 :ivar type: The delta type, e.g. 'add', 'change', 'remove'
24 :ivar data: The raw delta data
27 NOTE: The 'data' variable above is being incorrectly cross-linked by a
28 Sphinx bug: https://github.com/sphinx-doc/sphinx/issues/2549
31 _toSchema
= {'deltas': 'deltas'}
32 _toPy
= {'deltas': 'deltas'}
34 def __init__(self
, deltas
=None):
36 :param deltas: [str, str, object]
41 Change
= namedtuple('Change', 'entity type data')
42 change
= Change(*self
.deltas
)
44 self
.entity
= change
.entity
45 self
.type = change
.type
46 self
.data
= change
.data
49 def from_json(cls
, data
):
50 return cls(deltas
=data
)
53 class ResourcesFacade(Type
):
54 """Patch parts of ResourcesFacade to make it work.
57 @ReturnMapping(_client
.AddPendingResourcesResult
)
58 async def AddPendingResources(self
, application_tag
, charm_url
, resources
):
59 """Fix the calling signature of AddPendingResources.
61 The ResourcesFacade doesn't conform to the standard facade pattern in
62 the Juju source, which leads to the schemagened code not matching up
63 properly with the actual calling convention in the API. There is work
64 planned to fix this in Juju, but we have to work around it for now.
68 resources : typing.Sequence<+T_co>[~CharmResource]<~CharmResource>
69 Returns -> typing.Union[_ForwardRef('ErrorResult'),
70 typing.Sequence<+T_co>[str]]
72 # map input types to rpc msg
74 msg
= dict(type='Resources',
75 request
='AddPendingResources',
78 _params
['tag'] = application_tag
79 _params
['url'] = charm_url
80 _params
['resources'] = resources
81 reply
= await self
.rpc(msg
)