Skip to content

Instantly share code, notes, and snippets.

@Marhc
Forked from techniq/audit_mixin.py
Created November 21, 2019 22:40
Show Gist options
  • Save Marhc/e2808909ce9ba3874ff0064bcd3c1ec6 to your computer and use it in GitHub Desktop.
Save Marhc/e2808909ce9ba3874ff0064bcd3c1ec6 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