Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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 _get_measures_by_name(resources, metric_wildcards, operations,
start, stop, granularity, needed_overlap, fill,
details):
references = []
for r in resources:
references.extend([
processor.MetricReference(m, agg, r, wildcard)
for wildcard, agg in metric_wildcards
for m in r.metrics if fnmatch.fnmatch(m.name, wildcard)
])
if not references:
raise indexer.NoSuchMetric(set((m for (m, a) in metric_wildcards)))
response = {
"measures": get_measures_or_abort(
references, operations, start, stop, granularity,
needed_overlap, fill)
}
if details:
response["references"] = set((r.resource for r in references))
return response
def post(self):
resource = pecan.request.indexer.get_resource(
self.resource_type, self.resource_id)
if not resource:
abort(404, six.text_type(indexer.NoSuchResource(self.resource_id)))
enforce("update resource", resource)
metrics = deserialize_and_validate(MetricsSchema)
try:
r = pecan.request.indexer.update_resource(
self.resource_type,
self.resource_id,
metrics=metrics,
append_metrics=True,
create_revision=False)
except (indexer.NoSuchMetric,
indexer.NoSuchArchivePolicy,
ValueError) as e:
abort(400, six.text_type(e))
except indexer.NamedMetricAlreadyExists as e:
abort(409, six.text_type(e))
except indexer.NoSuchResource as e:
abort(404, six.text_type(e))
return r.metrics
def _lookup(self, id, *remainder):
try:
metric_id = uuid.UUID(id)
except ValueError:
abort(404, indexer.NoSuchMetric(id))
metrics = pecan.request.indexer.list_metrics(
id=metric_id, details=True)
if not metrics:
abort(404, indexer.NoSuchMetric(id))
return MetricController(metrics[0]), remainder
def _lookup(self, id, *remainder):
try:
metric_id = uuid.UUID(id)
except ValueError:
abort(404, six.text_type(indexer.NoSuchMetric(id)))
# Load details for ACL
metrics = pecan.request.indexer.list_metrics(
attribute_filter={"=": {"id": metric_id}}, details=True)
if not metrics:
abort(404, six.text_type(indexer.NoSuchMetric(id)))
return MetricController(metrics[0]), remainder
def delete_metric(self, id):
with self.facade.writer() as session:
if session.query(Metric).filter(
Metric.id == id, Metric.status == 'active').update(
{"status": "delete", "resource_id": None}) == 0:
raise indexer.NoSuchMetric(id)
def _set_metrics_for_resource(session, r, metrics):
for name, value in six.iteritems(metrics):
if isinstance(value, uuid.UUID):
try:
update = session.query(Metric).filter(
Metric.id == value,
Metric.status == 'active',
Metric.creator == r.creator,
).update({"resource_id": r.id, "name": name})
except exception.DBDuplicateEntry:
raise indexer.NamedMetricAlreadyExists(name)
if update == 0:
raise indexer.NoSuchMetric(value)
else:
unit = value.get('unit')
ap_name = value['archive_policy_name']
m = Metric(id=uuid.uuid4(),
creator=r.creator,
archive_policy_name=ap_name,
name=name,
unit=unit,
resource_id=r.id)
session.add(m)
try:
session.flush()
except exception.DBDuplicateEntry:
raise indexer.NamedMetricAlreadyExists(name)
except exception.DBReferenceError as e:
if (e.constraint ==
results = []
for key, resources in itertools.groupby(resources, groupper):
try:
results.append({
"group": dict(key),
"measures": self._get_measures_by_name(
resources, references, body["operations"],
start, stop, granularity, needed_overlap, fill,
details=details)
})
except indexer.NoSuchMetric:
pass
if not results:
api.abort(
400,
indexer.NoSuchMetric(set((m for (m, a) in references))))
return results
else:
try:
metric_ids = set(six.text_type(utils.UUID(m))
for (m, a) in references)
except ValueError as e:
api.abort(400, {"cause": "Invalid metric references",
"reason": six.text_type(e),
"detail": references})
metrics = pecan.request.indexer.list_metrics(
attribute_filter={"in": {"id": metric_ids}})
missing_metric_ids = (set(metric_ids)
- set(six.text_type(m.id) for m in metrics))
if missing_metric_ids: