juju/client/overrides.py

   1 from collections import namedtuple
   2
   3 from .facade import ReturnMapping, Type
   4 from .import _client
   5
   6 __all__ = [
   7     'Delta',
   8 ]
   9
  10 __patches__ = [
  11     'ResourcesFacade',
  12 ]
  13
  14
  15 class Delta(Type):
  16     """A single websocket delta.
  17
  18     :ivar entity: The entity name, e.g. 'unit', 'application'
  19     :vartype entity: str
  20
  21     :ivar type: The delta type, e.g. 'add', 'change', 'remove'
  22     :vartype type: str
  23
  24     :ivar data: The raw delta data
  25     :vartype data: dict
  26
  27     NOTE: The 'data' variable above is being incorrectly cross-linked by a
  28     Sphinx bug: https://github.com/sphinx-doc/sphinx/issues/2549
  29
  30     """
  31     _toSchema = {'deltas': 'deltas'}
  32     _toPy = {'deltas': 'deltas'}
  33
  34     def __init__(self, deltas=None):
  35         """
  36         :param deltas: [str, str, object]
  37
  38         """
  39         self.deltas = deltas
  40
  41         Change = namedtuple('Change', 'entity type data')
  42         change = Change(*self.deltas)
  43
  44         self.entity = change.entity
  45         self.type = change.type
  46         self.data = change.data
  47
  48     @classmethod
  49     def from_json(cls, data):
  50         return cls(deltas=data)
  51
  52
  53 class ResourcesFacade(Type):
  54     """Patch parts of ResourcesFacade to make it work.
  55     """
  56
  57     @ReturnMapping(_client.AddPendingResourcesResult)
  58     async def AddPendingResources(self, application_tag, charm_url, resources):
  59         """Fix the calling signature of AddPendingResources.
  60
  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.
  65
  66         application_tag : str
  67         charm_url : str
  68         resources : typing.Sequence<+T_co>[~CharmResource]<~CharmResource>
  69         Returns -> typing.Union[_ForwardRef('ErrorResult'),
  70                                 typing.Sequence<+T_co>[str]]
  71         """
  72         # map input types to rpc msg
  73         _params = dict()
  74         msg = dict(type='Resources',
  75                    request='AddPendingResources',
  76                    version=1,
  77                    params=_params)
  78         _params['tag'] = application_tag
  79         _params['url'] = charm_url
  80         _params['resources'] = resources
  81         reply = await self.rpc(msg)
  82         return reply