Skip to content

Instantly share code, notes, and snippets.

@victorusachev
Forked from amancioandre/audit_mixin.py
Created April 18, 2020 01:08
Show Gist options
  • Save victorusachev/7f76289fc160de3cbd7476d52cbc9017 to your computer and use it in GitHub Desktop.
Save victorusachev/7f76289fc160de3cbd7476d52cbc9017 to your computer and use it in GitHub Desktop.
Useful SQLAlchemy Mixins
from datetime import datetime
from sqlalchemy import Column, Integer, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declared_attr
from flask_security import current_user
class AuditMixin(object):
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
@declared_attr
def created_by_id(cls):
return Column(Integer,
ForeignKey('user.id', name='fk_%s_created_by_id' % cls.__name__, use_alter=True),
# nullable=False,
default=_current_user_id_or_none
)
@declared_attr
def created_by(cls):
return relationship(
'User',
primaryjoin='User.id == %s.created_by_id' % cls.__name__,
remote_side='User.id'
)
@declared_attr
def updated_by_id(cls):
return Column(Integer,
ForeignKey('user.id', name='fk_%s_updated_by_id' % cls.__name__, use_alter=True),
# nullable=False,
default=_current_user_id_or_none,
onupdate=_current_user_id_or_none
)
@declared_attr
def updated_by(cls):
return relationship(
'User',
primaryjoin='User.id == %s.updated_by_id' % cls.__name__,
remote_side='User.id'
)
def _current_user_id_or_none():
try:
return current_user.id
except:
return None
from flask import abort
from application.database import db
class BaseMixin(object):
_repr_hide = ['created_at', 'updated_at']
@classmethod
def query(cls):
return db.session.query(cls)
@classmethod
def get(cls, id):
return cls.query.get(id)
@classmethod
def get_by(cls, **kw):
return cls.query.filter_by(**kw).first()
@classmethod
def get_or_404(cls, id):
rv = cls.get(id)
if rv is None:
abort(404)
return rv
@classmethod
def get_or_create(cls, **kw):
r = cls.get_by(**kw)
if not r:
r = cls(**kw)
db.session.add(r)
return r
@classmethod
def create(cls, **kw):
r = cls(**kw)
db.session.add(r)
return r
def save(self):
db.session.add(self)
def delete(self):
db.session.delete(self)
def __repr__(self):
values = ', '.join("%s=%r" % (n, getattr(self, n)) for n in self.__table__.c.keys() if n not in self._repr_hide)
return "%s(%s)" % (self.__class__.__name__, values)
def filter_string(self):
return self.__str__()
class ChildMixin(object):
@property
def parents(self):
"""
Returns an entity's parent and all parents of parents as a list,
ordered by oldest parent first
"""
parents = []
if hasattr(self, "parent"):
if hasattr(self.parent, "parent"):
parents.extend(self.parent.parents)
parents.append(self.parent)
return parents
# Helpful to easily json serialize SQLAlchemy query results
# http://piotr.banaszkiewicz.org/blog/2012/06/30/serialize-sqlalchemy-results-into-json/
from collections import OrderedDict
class DictSerializableMixin(object):
def _asdict(self):
result = OrderedDict()
for key in self.__mapper__.c.keys():
result[key] = getattr(self, key)
return result
# Example
from flask import jsonify
from application.models import BlogPost
class BlogPost(db.Model, DictSerializableMixin):
pass
@app.route("/posts")
def blog_posts():
return jsonify(posts=list(BlogPost.query.all()))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment