Source code for panko.storage.base

#
# Copyright 2012 New Dream Network, LLC (DreamHost)
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Base classes for storage engines
"""

import six

import panko


[docs]class Model(object): """Base class for storage API models.""" def __init__(self, **kwds): self.fields = list(kwds) for k, v in six.iteritems(kwds): setattr(self, k, v)
[docs] def as_dict(self): d = {} for f in self.fields: v = getattr(self, f) if isinstance(v, Model): v = v.as_dict() elif isinstance(v, list) and v and isinstance(v[0], Model): v = [sub.as_dict() for sub in v] d[f] = v return d
def __eq__(self, other): return self.as_dict() == other.as_dict() def __ne__(self, other): return not self.__eq__(other)
[docs]class Connection(object): """Base class for event storage system connections.""" # A dictionary representing the capabilities of this driver. CAPABILITIES = { 'events': {'query': {'simple': False}}, } STORAGE_CAPABILITIES = { 'storage': {'production_ready': False}, } @staticmethod def __init__(url, conf): pass @staticmethod
[docs] def upgrade(): """Migrate the database to `version` or the most recent version."""
@staticmethod
[docs] def clear(): """Clear database."""
@staticmethod
[docs] def record_events(events): """Write the events to the backend storage system. :param events: a list of model.Event objects. """ raise panko.NotImplementedError('Events not implemented.')
@staticmethod
[docs] def get_events(event_filter, pagination=None): """Return an iterable of model.Event objects."""
@staticmethod
[docs] def get_event_types(): """Return all event types as an iterable of strings.""" raise panko.NotImplementedError('Events not implemented.')
@staticmethod
[docs] def get_trait_types(event_type): """Return a dictionary containing the name and data type of the trait. Only trait types for the provided event_type are returned. :param event_type: the type of the Event """ raise panko.NotImplementedError('Events not implemented.')
@staticmethod
[docs] def get_traits(event_type, trait_type=None): """Return all trait instances associated with an event_type. If trait_type is specified, only return instances of that trait type. :param event_type: the type of the Event to filter by :param trait_type: the name of the Trait to filter by """ raise panko.NotImplementedError('Events not implemented.')
@classmethod
[docs] def get_capabilities(cls): """Return an dictionary with the capabilities of each driver.""" return cls.CAPABILITIES
@classmethod
[docs] def get_storage_capabilities(cls): """Return a dictionary representing the performance capabilities. This is needed to evaluate the performance of each driver. """ return cls.STORAGE_CAPABILITIES
@staticmethod
[docs] def clear_expired_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 panko.NotImplementedError('Clearing events not implemented')