Sqlalchemy mixins / и прослушиватель событий
Я сразу же пытаюсь сделать две новые вещи, поэтому приветствуется помощь в упрощении и уточнении.
from sqlalchemy.ext.declarative import declared_attr from sqlalchemy import Column, Float, event class TimeStampMixin(object): @declared_attr def __tablename__(cls): return cls.__name__.lower() created = Column(Float) modified = Column(Float) def __init__(self, created = None, modified = None): self.created = created self.modified = modified def create_time(mapper, connection, target): target.created = time() #def modified_time(mapper, connection, target): # target.modified = time() event.listen(TimeStampMixin, 'before_insert', create_time) #event.listen(TimeStampMixin, 'before_update', modified_time) Поэтому я хочу создать mixin, который я могу применить в любом классе:
- Выполнение сопоставлений DateTime в фильтрах SQLAlchemy
- SQLAlchemy: создание или повторное использование сеанса
- SQLAlchemy Отношение «многие ко многим» на одной таблице
- Настройка Flask-SQLAlchemy для использования нескольких баз данных с Flask-Restless
- добавить столбец в таблицу SQLAlchemy
- Как закрыть сеанс SQLAlchemy?
class MyClass(TimeStampMixin, Base): etc, etc, etc
Этот класс наследует функциональность, которая создает временную метку при создании и создает / изменяет временную метку при обновлении.
при импорте я получаю эту ошибку:
raise exc.UnmappedClassError(class_) sqlalchemy.orm.exc.UnmappedClassError: Class 'db.database.TimeStampMixin' is not mapped
aaaand Я в тупике.
4 Solutions collect form web for “Sqlalchemy mixins / и прослушиватель событий”
Вот что я сделал бы, чтобы прослушать события before_insert : добавьте classmethod к вашему TimeStampMixin который регистрирует текущий класс и обрабатывает время создания установки.
Например
class TimeStampMixin(object): # other class methods @staticmethod def create_time(mapper, connection, target): target.created = time() @classmethod def register(cls): sqlalchemy.event.listen(cls, 'before_insert', cls.create_time)
Таким образом, вы можете:
- Легко расширять и изменять то, что вы слушаете, и то, что вы регистрируете.
- Переопределить метод create_time для определенных классов
- Будьте ясны в отношении того, какие методы должны иметь установленные временные метки.
Вы можете использовать его просто:
class MyMappedClass(TimeStampMixin, Base): pass MyMappedClass.register()
Простая, очень ясная, без магии, но все же инкапсулирует, как вы хотите.
Прикрепите слушателя внутри метода класса и приложите это событие к дочернему классу.
class TimeStampMixin(object): @staticmethod def create_time(mapper, connection, target): target.created = time() @classmethod def __declare_last__(cls): # get called after mappings are completed # http://docs.sqlalchemy.org/en/rel_0_7/orm/extensions/declarative.html#declare-last event.listen(cls, 'before_insert', cls.create_time)
вы также можете сделать это следующим образом:
from sqlalchemy.orm.interfaces import MapperExtension class BaseExtension(MapperExtension): """Base entension class for all entities """ def before_insert(self, mapper, connection, instance): """ set the created_at """ instance.created = datetime.datetime.now() def before_update(self, mapper, connection, instance): """ set the updated_at """ instance.modified = datetime.datetime.now() class TimeStampMixin(object): id = Column(Integer, primary_key=True, autoincrement=True) created = Column(DateTime()) modified = Column(DateTime()) __table_args__ = { 'mysql_engine': 'InnoDB', 'mysql_charset': 'utf8' } __mapper_args__ = { 'extension': BaseExtension() }
и определите свои классы следующим образом:
class User(TimeStampMixin, Base):
Лучший способ в современной SqlAlchemy – использовать декоратор @listens_for с propagate=True .
from datetime import datetime from sqlalchemy import Column, Float from sqlalchemy.ext.declarative import declared_attr from sqlalchemy.event import listens_for class TimestampMixin(): @declared_attr def created(cls): return Column(DateTime(timezone=True)) @listens_for(TimeStampMixin, "init", propagate=True) def timestamp_init(target, args, kwargs): kwargs["created"] = datetime.utcnow()