--- /dev/null
+import re
+from collections import namedtuple
+
+from . import _client, _definitions
+from .facade import ReturnMapping, Type, TypeEncoder
+
+__all__ = [
+ 'Delta',
+ 'Number',
+ 'Binary',
+ 'ConfigValue',
+ 'Resource',
+]
+
+__patches__ = [
+ 'ResourcesFacade',
+ 'AllWatcherFacade',
+ 'ActionFacade',
+]
+
+
+class Delta(Type):
+ """A single websocket delta.
+
+ :ivar entity: The entity name, e.g. 'unit', 'application'
+ :vartype entity: str
+
+ :ivar type: The delta type, e.g. 'add', 'change', 'remove'
+ :vartype type: str
+
+ :ivar data: The raw delta data
+ :vartype data: dict
+
+ NOTE: The 'data' variable above is being incorrectly cross-linked by a
+ Sphinx bug: https://github.com/sphinx-doc/sphinx/issues/2549
+
+ """
+ _toSchema = {'deltas': 'deltas'}
+ _toPy = {'deltas': 'deltas'}
+
+ def __init__(self, deltas=None):
+ """
+ :param deltas: [str, str, object]
+
+ """
+ self.deltas = deltas
+
+ Change = namedtuple('Change', 'entity type data')
+ change = Change(*self.deltas)
+
+ self.entity = change.entity
+ self.type = change.type
+ self.data = change.data
+
+ @classmethod
+ def from_json(cls, data):
+ return cls(deltas=data)
+
+
+class ResourcesFacade(Type):
+ """Patch parts of ResourcesFacade to make it work.
+ """
+
+ @ReturnMapping(_client.AddPendingResourcesResult)
+ async def AddPendingResources(self, application_tag, charm_url, resources):
+ """Fix the calling signature of AddPendingResources.
+
+ The ResourcesFacade doesn't conform to the standard facade pattern in
+ the Juju source, which leads to the schemagened code not matching up
+ properly with the actual calling convention in the API. There is work
+ planned to fix this in Juju, but we have to work around it for now.
+
+ application_tag : str
+ charm_url : str
+ resources : typing.Sequence<+T_co>[~CharmResource]<~CharmResource>
+ Returns -> typing.Union[_ForwardRef('ErrorResult'),
+ typing.Sequence<+T_co>[str]]
+ """
+ # map input types to rpc msg
+ _params = dict()
+ msg = dict(type='Resources',
+ request='AddPendingResources',
+ version=1,
+ params=_params)
+ _params['tag'] = application_tag
+ _params['url'] = charm_url
+ _params['resources'] = resources
+ reply = await self.rpc(msg)
+ return reply
+
+
+class AllWatcherFacade(Type):
+ """
+ Patch rpc method of allwatcher to add in 'id' stuff.
+
+ """
+ async def rpc(self, msg):
+ if not hasattr(self, 'Id'):
+ client = _client.ClientFacade.from_connection(self.connection)
+
+ result = await client.WatchAll()
+ self.Id = result.watcher_id
+
+ msg['Id'] = self.Id
+ result = await self.connection.rpc(msg, encoder=TypeEncoder)
+ return result
+
+
+class ActionFacade(Type):
+
+ class _FindTagsResults(Type):
+ _toSchema = {'matches': 'matches'}
+ _toPy = {'matches': 'matches'}
+
+ def __init__(self, matches=None, **unknown_fields):
+ '''
+ FindTagsResults wraps the mapping between the requested prefix and the
+ matching tags for each requested prefix.
+
+ Matches map[string][]Entity `json:"matches"`
+ '''
+ self.matches = {}
+ matches = matches or {}
+ for prefix, tags in matches.items():
+ self.matches[prefix] = [_definitions.Entity.from_json(r)
+ for r in tags]
+
+ @ReturnMapping(_FindTagsResults)
+ async def FindActionTagsByPrefix(self, prefixes):
+ '''
+ prefixes : typing.Sequence[str]
+ Returns -> typing.Sequence[~Entity]
+ '''
+ # map input types to rpc msg
+ _params = dict()
+ msg = dict(type='Action',
+ request='FindActionTagsByPrefix',
+ version=2,
+ params=_params)
+ _params['prefixes'] = prefixes
+ reply = await self.rpc(msg)
+ return reply
+
+
+class Number(_definitions.Number):
+ """
+ This type represents a semver string.
+
+ Because it is not standard JSON, the typical from_json parsing fails and
+ the parsing must be handled specially.
+
+ See https://github.com/juju/version for more info.
+ """
+ numberPat = re.compile(r'^(\d{1,9})\.(\d{1,9})(?:\.|-([a-z]+))(\d{1,9})(\.\d{1,9})?$') # noqa
+
+ def __init__(self, major=None, minor=None, patch=None, tag=None,
+ build=None, **unknown_fields):
+ '''
+ major : int
+ minor : int
+ patch : int
+ tag : str
+ build : int
+ '''
+ self.major = int(major or '0')
+ self.minor = int(minor or '0')
+ self.patch = int(patch or '0')
+ self.tag = tag or ''
+ self.build = int(build or '0')
+
+ def __repr__(self):
+ return '<Number major={} minor={} patch={} tag={} build={}>'.format(
+ self.major, self.minor, self.patch, self.tag, self.build)
+
+ def __str__(self):
+ return self.serialize()
+
+ @property
+ def _cmp(self):
+ return (self.major, self.minor, self.tag, self.patch, self.build)
+
+ def __eq__(self, other):
+ return isinstance(other, type(self)) and self._cmp == other._cmp
+
+ def __lt__(self, other):
+ return self._cmp < other._cmp
+
+ def __le__(self, other):
+ return self._cmp <= other._cmp
+
+ def __gt__(self, other):
+ return self._cmp > other._cmp
+
+ def __ge__(self, other):
+ return self._cmp >= other._cmp
+
+ @classmethod
+ def from_json(cls, data):
+ parsed = None
+ if isinstance(data, cls):
+ return data
+ elif data is None:
+ return cls()
+ elif isinstance(data, dict):
+ parsed = data
+ elif isinstance(data, str):
+ match = cls.numberPat.match(data)
+ if match:
+ parsed = {
+ 'major': match.group(1),
+ 'minor': match.group(2),
+ 'tag': match.group(3),
+ 'patch': match.group(4),
+ 'build': (match.group(5)[1:] if match.group(5)
+ else 0),
+ }
+ if not parsed:
+ raise TypeError('Unable to parse Number version string: '
+ '{}'.format(data))
+ d = {}
+ for k, v in parsed.items():
+ d[cls._toPy.get(k, k)] = v
+
+ return cls(**d)
+
+ def serialize(self):
+ s = ""
+ if not self.tag:
+ s = "{}.{}.{}".format(self.major, self.minor, self.patch)
+ else:
+ s = "{}.{}-{}{}".format(self.major, self.minor, self.tag,
+ self.patch)
+ if self.build:
+ s = "{}.{}".format(s, self.build)
+ return s
+
+ def to_json(self):
+ return self.serialize()
+
+
+class Binary(_definitions.Binary):
+ """
+ This type represents a semver string with additional series and arch info.
+
+ Because it is not standard JSON, the typical from_json parsing fails and
+ the parsing must be handled specially.
+
+ See https://github.com/juju/version for more info.
+ """
+ binaryPat = re.compile(r'^(\d{1,9})\.(\d{1,9})(?:\.|-([a-z]+))(\d{1,9})(\.\d{1,9})?-([^-]+)-([^-]+)$') # noqa
+
+ def __init__(self, number=None, series=None, arch=None, **unknown_fields):
+ '''
+ number : Number
+ series : str
+ arch : str
+ '''
+ self.number = Number.from_json(number)
+ self.series = series
+ self.arch = arch
+
+ def __repr__(self):
+ return '<Binary number={} series={} arch={}>'.format(
+ self.number, self.series, self.arch)
+
+ def __str__(self):
+ return self.serialize()
+
+ def __eq__(self, other):
+ return (
+ isinstance(other, type(self)) and
+ other.number == self.number and
+ other.series == self.series and
+ other.arch == self.arch)
+
+ @classmethod
+ def from_json(cls, data):
+ parsed = None
+ if isinstance(data, cls):
+ return data
+ elif data is None:
+ return cls()
+ elif isinstance(data, dict):
+ parsed = data
+ elif isinstance(data, str):
+ match = cls.binaryPat.match(data)
+ if match:
+ parsed = {
+ 'number': {
+ 'major': match.group(1),
+ 'minor': match.group(2),
+ 'tag': match.group(3),
+ 'patch': match.group(4),
+ 'build': (match.group(5)[1:] if match.group(5)
+ else 0),
+ },
+ 'series': match.group(6),
+ 'arch': match.group(7),
+ }
+ if parsed is None:
+ raise TypeError('Unable to parse Binary version string: '
+ '{}'.format(data))
+ d = {}
+ for k, v in parsed.items():
+ d[cls._toPy.get(k, k)] = v
+
+ return cls(**d)
+
+ def serialize(self):
+ return "{}-{}-{}".format(self.number.serialize(),
+ self.series, self.arch)
+
+ def to_json(self):
+ return self.serialize()
+
+
+class ConfigValue(_definitions.ConfigValue):
+ def __repr__(self):
+ return '<{} source={} value={}>'.format(type(self).__name__,
+ repr(self.source),
+ repr(self.value))
+
+
+class Resource(Type):
+ _toSchema = {'application': 'application',
+ 'charmresource': 'CharmResource',
+ 'id_': 'id',
+ 'pending_id': 'pending-id',
+ 'timestamp': 'timestamp',
+ 'username': 'username',
+ 'name': 'name',
+ 'origin': 'origin'}
+ _toPy = {'CharmResource': 'charmresource',
+ 'application': 'application',
+ 'id': 'id_',
+ 'pending-id': 'pending_id',
+ 'timestamp': 'timestamp',
+ 'username': 'username',
+ 'name': 'name',
+ 'origin': 'origin'}
+
+ def __init__(self, charmresource=None, application=None, id_=None,
+ pending_id=None, timestamp=None, username=None, name=None,
+ origin=None, **unknown_fields):
+ '''
+ charmresource : CharmResource
+ application : str
+ id_ : str
+ pending_id : str
+ timestamp : str
+ username : str
+ name: str
+ origin : str
+ '''
+ if charmresource:
+ self.charmresource = _client.CharmResource.from_json(charmresource)
+ else:
+ self.charmresource = None
+ self.application = application
+ self.id_ = id_
+ self.pending_id = pending_id
+ self.timestamp = timestamp
+ self.username = username
+ self.name = name
+ self.origin = origin
projects / osm / N2VC.git / blobdiff