/usr/share/pyshared/juju/state/charm.py is in juju-0.7 0.7-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | from twisted.internet.defer import (
inlineCallbacks, returnValue, succeed)
from zookeeper import NoNodeException
from juju.charm.config import ConfigOptions
from juju.charm.metadata import MetaData
from juju.charm.url import CharmURL
from juju.lib import under, serializer
from juju.state.base import StateBase
from juju.state.errors import CharmStateNotFound
def _charm_path(charm_id):
return "/charms/%s" % under.quote(charm_id)
class CharmStateManager(StateBase):
"""Manages the state of charms in an environment."""
@inlineCallbacks
def add_charm_state(self, charm_id, charm, url):
"""Register metadata about the provided Charm.
:param str charm_id: The key under which to store the Charm.
:param charm: The Charm itself.
:param url: The provider storage url for the Charm.
"""
charm_data = {
"config": charm.config.get_serialization_data(),
"metadata": charm.metadata.get_serialization_data(),
"sha256": charm.get_sha256(),
"url": url
}
# XXX In the future we'll have to think about charm
# replacements here. For now this will do, and will
# explode reliably in case of conflicts.
yield self._client.create(
_charm_path(charm_id), serializer.dump(charm_data))
charm_state = CharmState(self._client, charm_id, charm_data)
returnValue(charm_state)
@inlineCallbacks
def get_charm_state(self, charm_id):
"""Retrieve a CharmState for the given charm id."""
try:
content, stat = yield self._client.get(_charm_path(charm_id))
except NoNodeException:
raise CharmStateNotFound(charm_id)
charm_data = serializer.load(content)
charm_state = CharmState(self._client, charm_id, charm_data)
returnValue(charm_state)
class CharmState(object):
"""State of a charm registered in an environment."""
def __init__(self, client, charm_id, charm_data):
self._client = client
self._charm_url = CharmURL.parse(charm_id)
self._charm_url.assert_revision()
self._metadata = MetaData()
self._metadata.parse_serialization_data(charm_data["metadata"])
self._config = ConfigOptions()
self._config.parse(charm_data["config"])
# Just a health check:
assert self._metadata.name == self.name
self._sha256 = charm_data["sha256"]
self._bundle_url = charm_data.get("url")
@property
def name(self):
"""The charm name."""
return self._charm_url.name
@property
def revision(self):
"""The monotonically increasing charm revision number.
"""
return self._charm_url.revision
@property
def bundle_url(self):
"""The url to the charm bundle in the provider storage."""
return self._bundle_url
@property
def id(self):
"""The charm id"""
return str(self._charm_url)
def get_metadata(self):
"""Return deferred MetaData."""
return succeed(self._metadata)
def get_config(self):
"""Return deferred ConfigOptions."""
return succeed(self._config)
def get_sha256(self):
"""Return deferred sha256 for the charm."""
return succeed(self._sha256)
def is_subordinate(self):
"""Is this a subordinate charm."""
return self._metadata.is_subordinate
|