Skip to content

Instantly share code, notes, and snippets.

@ksindi
Forked from techniq/audit_mixin.py
Created July 9, 2014 13:37
Show Gist options
  • Save ksindi/530359c1bf7b9c7bc011 to your computer and use it in GitHub Desktop.
Save ksindi/530359c1bf7b9c7bc011 to your computer and use it in GitHub Desktop.
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