juju/client/overrides.py

   1 from collections import namedtuple
   2
   3 from .facade import ReturnMapping, Type, TypeEncoder
   4 from .import _client
   5
   6
   7 __all__ = [
   8     'Delta',
   9 ]
  10
  11 __patches__ = [
  12     'ResourcesFacade',
  13     'AllWatcherFacade'
  14 ]
  15
  16
  17 class Delta(Type):
  18     """A single websocket delta.
  19
  20     :ivar entity: The entity name, e.g. 'unit', 'application'
  21     :vartype entity: str
  22
  23     :ivar type: The delta type, e.g. 'add', 'change', 'remove'
  24     :vartype type: str
  25
  26     :ivar data: The raw delta data
  27     :vartype data: dict
  28
  29     NOTE: The 'data' variable above is being incorrectly cross-linked by a
  30     Sphinx bug: https://github.com/sphinx-doc/sphinx/issues/2549
  31
  32     """
  33     _toSchema = {'deltas': 'deltas'}
  34     _toPy = {'deltas': 'deltas'}
  35
  36     def __init__(self, deltas=None):
  37         """
  38         :param deltas: [str, str, object]
  39
  40         """
  41         self.deltas = deltas
  42
  43         Change = namedtuple('Change', 'entity type data')
  44         change = Change(*self.deltas)
  45
  46         self.entity = change.entity
  47         self.type = change.type
  48         self.data = change.data
  49
  50     @classmethod
  51     def from_json(cls, data):
  52         return cls(deltas=data)
  53
  54
  55 class ResourcesFacade(Type):
  56     """Patch parts of ResourcesFacade to make it work.
  57     """
  58
  59     @ReturnMapping(_client.AddPendingResourcesResult)
  60     async def AddPendingResources(self, application_tag, charm_url, resources):
  61         """Fix the calling signature of AddPendingResources.
  62
  63         The ResourcesFacade doesn't conform to the standard facade pattern in
  64         the Juju source, which leads to the schemagened code not matching up
  65         properly with the actual calling convention in the API.  There is work
  66         planned to fix this in Juju, but we have to work around it for now.
  67
  68         application_tag : str
  69         charm_url : str
  70         resources : typing.Sequence<+T_co>[~CharmResource]<~CharmResource>
  71         Returns -> typing.Union[_ForwardRef('ErrorResult'),
  72                                 typing.Sequence<+T_co>[str]]
  73         """
  74         # map input types to rpc msg
  75         _params = dict()
  76         msg = dict(type='Resources',
  77                    request='AddPendingResources',
  78                    version=1,
  79                    params=_params)
  80         _params['tag'] = application_tag
  81         _params['url'] = charm_url
  82         _params['resources'] = resources
  83         reply = await self.rpc(msg)
  84         return reply
  85
  86 class AllWatcherFacade(Type):
  87     """
  88     Patch rpc method of allwatcher to add in 'id' stuff.
  89
  90     """
  91     async def rpc(self, msg):
  92         if not hasattr(self, 'Id'):
  93             client = _client.ClientFacade.from_connection(self.connection)
  94
  95             result = await client.WatchAll()
  96             self.Id = result.watcher_id
  97
  98         msg['Id'] = self.Id
  99         result =  await self.connection.rpc(msg, encoder=TypeEncoder)
 100         return result