Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def validate_qs(start=None, stop=None, granularity=None,
needed_overlap=None, fill=None):
if needed_overlap is not None:
try:
needed_overlap = float(needed_overlap)
except ValueError:
abort(400, {"cause": "Argument value error",
"detail": "needed_overlap",
"reason": "Must be a number"})
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, {"cause": "Argument value error",
"detail": "start",
"reason": "Must be a datetime or a timestamp"})
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, {"cause": "Argument value error",
"detail": "stop",
"reason": "Must be a datetime or a timestamp"})
if granularity is not None:
try:
granularity = [utils.to_timespan(granularity)]
except ValueError:
abort(400, {"cause": "Argument value error",
"detail": "needed_overlap",
"reason": "Must be a number"})
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, {"cause": "Argument value error",
"detail": "start",
"reason": "Must be a datetime or a timestamp"})
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, {"cause": "Argument value error",
"detail": "stop",
"reason": "Must be a datetime or a timestamp"})
if granularity is not None:
try:
granularity = [utils.to_timespan(granularity)]
except ValueError as e:
abort(400, {"cause": "Argument value error",
"detail": "granularity",
"reason": six.text_type(e)})
if fill is not None:
try:
fill = FillSchema(fill)
enforce("search metric", metric)
if not pecan.request.body:
abort(400, "No query specified in body")
query = deserialize_and_validate(self.MetricSearchSchema)
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, "Invalid value for start")
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, "Invalid value for stop")
try:
predicate = self.MeasureQuery(query)
except self.MeasureQuery.InvalidQuery as e:
abort(400, six.text_type(e))
if granularity is not None:
granularity = sorted(
map(utils.to_timespan, arg_to_list(granularity)),
reverse=True)
metrics_and_aggregations = collections.defaultdict(list)
for metric in metrics:
reaggregation=None,
granularity=None,
needed_overlap=100.0, fill=None,
refresh=False, resample=None,
transform=None):
try:
needed_overlap = float(needed_overlap)
except ValueError:
abort(400, 'needed_overlap must be a number')
if needed_overlap != 100.0 and start is None and stop is None:
abort(400, 'start and/or stop must be provided if specifying '
'needed_overlap')
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, "Invalid value for start")
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, "Invalid value for stop")
if (aggregation
not in archive_policy.ArchivePolicy.VALID_AGGREGATION_METHODS):
abort(
400,
'Invalid aggregation value %s, must be one of %s'
% (aggregation,
archive_policy.ArchivePolicy.VALID_AGGREGATION_METHODS))
"detail": list(missing_metric_ids)})
for metric in metrics:
enforce("get metric", metric)
start = kwargs.get('start')
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, "Invalid value for start")
stop = kwargs.get('stop')
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, "Invalid value for stop")
aggregation = kwargs.get('aggregation', 'mean')
if (aggregation
not in archive_policy.ArchivePolicy.VALID_AGGREGATION_METHODS):
abort(
400,
'Invalid aggregation value %s, must be one of %s'
% (aggregation,
archive_policy.ArchivePolicy.VALID_AGGREGATION_METHODS))
granularity = kwargs.get('granularity')
if granularity is not None:
try:
granularity = utils.to_timespan(granularity)
granularity = [utils.to_timespan(g)
for g in arg_to_list(granularity or [])]
metrics = pecan.request.indexer.list_metrics(
ids=arg_to_list(metric_id))
for metric in metrics:
enforce("search metric", metric)
if not pecan.request.body:
abort(400, "No query specified in body")
query = deserialize_and_validate(self.MetricSearchSchema)
if start is not None:
try:
start = utils.to_timestamp(start)
except Exception:
abort(400, "Invalid value for start")
if stop is not None:
try:
stop = utils.to_timestamp(stop)
except Exception:
abort(400, "Invalid value for stop")
try:
return {
str(metric.id): values
for metric, values in six.iteritems(
pecan.request.storage.search_value(
metrics, query, start, stop, aggregation,
granularity