Y
I used to do that. I think it's a good practice.
A good solution would be to create a basic class from which other models could be inherited and to add the necessary logic to the methods themselves.For his needs, working with the Flask Freamworth, wrote this thing:from ._base import db
from sqlalchemy.exc import IntegrityError, InterfaceError
from flask import flash
from sqlalchemy import event
from sqlalchemy.event import listen
from sqlalchemy.orm.interfaces import MapperExtension
from ..utils.redis import redis_store
from sqlalchemy import inspect
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from pickle import dumps, loads
@as_declarative()
class BaseExtension(MapperExtension):
def after_insert(self, mapper, connection, instance):
row = instance.query.filter_by(id = instance.id).first()
payload = {'model': instance.__class__.__name__, 'data': instance.id, 'type': 'INSERT', 'row': row}
redis_store.publish('realtime', dumps(payload))
def after_update(self, mapper, connection, instance):
row = instance.query.filter_by(id = instance.id).first()
payload = {'model': instance.__class__.__name__, 'data': instance.id, 'type': 'UPDATE', 'row': row}
redis_store.publish('realtime', dumps(payload))
def after_delete(self, mapper, connection, instance):
row = instance.query.filter_by(id = instance.id).first()
payload = {'model': instance.__class__.__name__, 'data': instance.id, 'type': 'DELETE', 'row': row}
redis_store.publish('realtime', dumps(payload))
class Base(db.Model):
__abstract__ = True
__mapper_args__ = { 'extension': BaseExtension() }
id = db.Column(db.Integer, primary_key=True)
created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
modified_at = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())
@classmethod
def create(cls,**kwargs):
c = cls(**kwargs)
db.session.add(c)
try:
db.session.commit()
flash((c.__tablename__).capitalize() + u' created successfully!', 'success')
except IntegrityError:
db.session.rollback()
flash((c.__tablename__).capitalize() + u' created failed!' + u' IntegrityError', 'error')
except InterfaceError:
db.session.rollback()
flash((c.__tablename__).capitalize() + u' created failed!' + u' InterfaceError', 'error')
return c
def __repr__(self):
mapper = inspect(self).mapper
ent = []
object_data = {col.key: getattr(self, col.key) if not col.key == 'created_at' and not col.key == 'password' and not col.key == 'modified_at' else None for col in mapper.column_attrs}
return "{0}".format(object_data)
Now I see how much unnecessary and wrong it is, but at least for self-development, it was a good experience. First, those fields that are repeated in all models (id, created_at, modified_al) are described only once. Second, the subject create(s) with some exceptions.Thirdly, http://docs.sqlalchemy.org/en/latest/orm/deprecated.html After_insert, after_update, after_delete is processed and redistributed. I did it for "real-time" data transmission (notification, data updates in dashboard, etc.)Four, reprwhich is also universal for all models.Also, we could add a so-called 'soft-delete', which would not remove the data from the OBD system, but also the methods to get out of the base "not removed," all and only deleted.There's a lot of things you can think of if you need to.Ultimately, once implemented base class will save time and space in models.