Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _handle_metadata(self, op, field_name, value):
if op == self.operators["in"]:
raise ceilometer.NotImplementedError('Metadata query with in '
'operator is not implemented')
field_name = field_name[len('resource_metadata.'):]
meta_table = META_TYPE_MAP[type(value)]
meta_alias = aliased(meta_table)
on_clause = and_(self.table.internal_id == meta_alias.id,
meta_alias.meta_key == field_name)
# outer join is needed to support metaquery
# with or operator on non existent metadata field
# see: test_query_non_existing_metadata_with_result
# test case.
self.query = self.query.outerjoin(meta_alias, on_clause)
return op(meta_alias.value, value)
def clear_expired_event_data(ttl):
"""Clear expired data from the backend storage system.
Clearing occurs according to the time-to-live.
:param ttl: Number of seconds to keep records for.
"""
raise ceilometer.NotImplementedError('Clearing events not implemented')
def get_events(event_filter, limit=None):
"""Return an iterable of model.Event objects."""
raise ceilometer.NotImplementedError('Events not implemented.')
def get_samples(sample_filter, limit=None):
"""Return an iterable of model.Sample instances.
:param sample_filter: Filter.
:param limit: Maximum number of results to return.
"""
raise ceilometer.NotImplementedError('Samples not implemented')
def get_event_types():
"""Return all event types as an iterable of strings."""
raise ceilometer.NotImplementedError('Events not implemented.')
def query_samples(filter_expr=None, orderby=None, limit=None):
"""Return an iterable of model.Sample objects.
:param filter_expr: Filter expression for query.
:param orderby: List of field name and direction pairs for order by.
:param limit: Maximum number of results to return.
"""
raise ceilometer.NotImplementedError('Complex query for samples '
'is not implemented.')
is_regex = True
column = 'f:' + value
r = {}
for row in rows:
data = rows[row]
r_data = {}
for key in data:
if ((op == '=' and key.startswith(column)) or
(op == '>=' and key >= column) or
(op == '<=' and key <= column) or
(op == '>' and key > column) or
(op == '<' and key < column) or
(is_regex and re.search(value, key))):
r_data[key] = data[key]
else:
raise ceilometer.NotImplementedError(
"In-memory QualifierFilter "
"doesn't support the %s "
"operation yet" % op)
if r_data:
r[row] = r_data
return r
def update_alarm(alarm):
"""Update alarm."""
raise ceilometer.NotImplementedError('Alarms not implemented')
def query_alarm_history(filter_expr=None, orderby=None, limit=None):
"""Return an iterable of model.AlarmChange objects.
:param filter_expr: Filter expression for query.
:param orderby: List of field name and direction pairs for order by.
:param limit: Maximum number of results to return.
"""
raise ceilometer.NotImplementedError('Complex query for alarms '
'history is not implemented.')