Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get(self):
resource = pecan.request.indexer.get_resource(
self._resource_type, self.id, with_metrics=True)
if resource:
enforce("get resource", resource)
etag_precondition_check(resource)
etag_set_headers(resource)
return resource
abort(404, indexer.NoSuchResource(self.id))
def _lookup(self, name, *remainder):
details = True if pecan.request.method == 'GET' else False
m = pecan.request.indexer.list_metrics(details=details,
name=name,
resource_id=self.resource_id)
if m:
return MetricController(m[0]), remainder
resource = pecan.request.indexer.get_resource(self.resource_type,
self.resource_id)
if resource:
abort(404, indexer.NoSuchMetric(name))
else:
abort(404, indexer.NoSuchResource(self.resource_id))
def delete(self):
resource = pecan.request.indexer.get_resource(
self._resource_type, self.id)
if not resource:
abort(404, six.text_type(indexer.NoSuchResource(self.id)))
enforce("delete resource", resource)
etag_precondition_check(resource)
try:
pecan.request.indexer.delete_resource(self.id)
except indexer.NoSuchResource as e:
abort(404, six.text_type(e))
def _lookup(self, name, *remainder):
m = pecan.request.indexer.list_metrics(
details=True,
attribute_filter={"and": [
{"=": {"name": name}},
{"=": {"resource_id": self.resource_id}},
]})
if m:
return MetricController(m[0]), remainder
resource = pecan.request.indexer.get_resource(self.resource_type,
self.resource_id)
if resource:
abort(404, six.text_type(indexer.NoSuchMetric(name)))
else:
abort(404, six.text_type(indexer.NoSuchResource(self.resource_id)))
def get(self):
resource = pecan.request.indexer.get_resource(
self._resource_type, self.id, with_metrics=True)
if resource:
enforce("get resource", resource)
etag_precondition_check(resource)
etag_set_headers(resource)
return resource
abort(404, six.text_type(indexer.NoSuchResource(self.id)))
def __init__(self, resource_type, id):
self._resource_type = resource_type
creator = pecan.request.auth_helper.get_current_user(
pecan.request)
try:
self.id = utils.ResourceUUID(id, creator)
except ValueError:
abort(404, indexer.NoSuchResource(id))
self.metric = NamedMetricController(str(self.id), self._resource_type)
self.history = ResourceHistoryController(str(self.id),
self._resource_type)
def get(self, **kwargs):
details = get_details(kwargs)
pagination_opts = get_pagination_options(
kwargs, RESOURCE_DEFAULT_PAGINATION)
resource = pecan.request.indexer.get_resource(
self.resource_type, self.resource_id)
if not resource:
abort(404, indexer.NoSuchResource(self.resource_id))
enforce("get resource", resource)
try:
# FIXME(sileht): next API version should returns
# {'resources': [...], 'links': [ ... pagination rel ...]}
return pecan.request.indexer.list_resources(
self.resource_type,
attribute_filter={"=": {"id": self.resource_id}},
details=details,
history=True,
**pagination_opts
)
except indexer.IndexerException as e:
abort(400, e)
def patch(self):
resource = pecan.request.indexer.get_resource(
self._resource_type, self.id, with_metrics=True)
if not resource:
abort(404, indexer.NoSuchResource(self.id))
enforce("update resource", resource)
etag_precondition_check(resource)
body = deserialize_and_validate(
schema_for(self._resource_type),
required=False)
if len(body) == 0:
etag_set_headers(resource)
return resource
for k, v in six.iteritems(body):
if k != 'metrics' and getattr(resource, k) != v:
create_revision = True
break
else:
with self.facade.writer() as session:
mappers = self._resource_type_to_mappers(session, resource_type)
resource_cls = mappers["resource"]
resource_history_cls = mappers["history"]
try:
# NOTE(sileht): We use FOR UPDATE that is not galera friendly,
# but they are no other way to cleanly patch a resource and
# store the history that safe when two concurrent calls are
# done.
q = session.query(resource_cls).filter(
resource_cls.id == resource_id).with_for_update()
r = q.first()
if r is None:
raise indexer.NoSuchResource(resource_id)
if create_revision:
# Build history
rh = resource_history_cls()
for col in sqlalchemy.inspect(resource_cls).columns:
setattr(rh, col.name, getattr(r, col.name))
now = utils.utcnow()
rh.revision_end = now
session.add(rh)
r.revision_start = now
# Update the resource
if ended_at is not _marker:
# NOTE(jd) MySQL does not honor checks. I hate it.
engine = session.connection()
if engine.dialect.name == "mysql":