Source code for sgcache.entity
import collections
import datetime
import sqlalchemy as sa
import migrate # this monkey-patches sqlalchemy
from .fields import sg_field_types
[docs]class EntityType(collections.Mapping):
"""Largely a mapping of fields, but also responsible for entity-level
caching mechanisms, including columns (on ``table.c``):
- ``_active``: Is the entity not "retired"?
- ``_cache_created_at``: When was this entity first cached?
- ``_cache_updated_at``: When was this entity last updated in the cache?
- ``_last_log_event_id``: What was the last ``LogEventEntry`` id to affect this entity?
"""
def __init__(self, cache, name, schema):
self.cache = cache
#: The name of this entity type, e.g.: ``Task``.
self.type_name = name
#: The name of the SQLAlchemy table, usually the lower-cased ``type_name``;
#: see :ref:`db_schema_rules`.
self.table_name = name.lower()
#: The SQLAlchemy table itself.
self.table = None # just a stub for the docs, really.
#: The :class:`EntitySchema` for this entity type.
self.schema = schema
self.fields = {}
for name, field_schema in schema.iteritems():
cls = sg_field_types[field_schema.data_type]
field = self.fields[name] = cls(self, name, field_schema)
def __repr__(self):
return '<%s %s>' % (self.__class__.__name__, self.type_name)
def __getitem__(self, key):
return self.fields[key]
def __iter__(self):
return iter(self.fields)
def __len__(self):
return len(self.fields)
def _construct_schema(self):
self.table = self.cache.metadata.tables.get(self.table_name)
if self.table is None:
self.table = sa.Table(self.table_name, self.cache.metadata,
sa.Column('id', sa.Integer, primary_key=True),
# We call this "active" to match the RPC values, even though
# the Python API tends to call the negative of this "retired".
sa.Column('_active', sa.Boolean, nullable=False, default=True),
)
self.table.create()
if '_cache_created_at' not in self.table.c:
# TODO: make these not-nullable
sa.Column('_cache_created_at', sa.DateTime, default=datetime.datetime.utcnow).create(self.table)
sa.Column('_cache_updated_at', sa.DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow).create(self.table)
if '_last_log_event_id' not in self.table.c:
sa.Column('_last_log_event_id', sa.Integer).create(self.table)
for field in self.fields.itervalues():
field._construct_schema(self.table)
def _clear(self, con):
for field in self.fields.itervalues():
field._clear(con)
con.execute(self.table.delete())