Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
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()))
@phniix

This comment has been minimized.

Copy link

@phniix phniix commented Jan 15, 2014

Very nice collection of mixins

@foxx

This comment has been minimized.

Copy link

@foxx foxx commented Jul 23, 2014

AuditMixin is not a proper way to do auditing, keeping only the last user ID that performed an update is wrong. You should also consider using utcnow() instead of now()

@kracekumar

This comment has been minimized.

Copy link

@kracekumar kracekumar commented May 22, 2015

@foxx: +1 for utcnow

@ghost

This comment has been minimized.

Copy link

@ghost ghost commented May 19, 2016

I will stick to now. instead of utcnow..

@atwalsh

This comment has been minimized.

Copy link

@atwalsh atwalsh commented Jul 31, 2018

On the discussion of now vs. utcnow, you may find this Stack Overflow answer useful. It discusses why you might want to use func.now() or func.current_timestamp() to have your database calculate the timestamp.

from sqlalchemy.sql import func

class AuditMixin(object):
    time_created = Column(DateTime(timezone=True), default=func.now())
    time_updated = Column(DateTime(timezone=True), onupdate=func.now())
@frfernandezdev

This comment has been minimized.

Copy link

@frfernandezdev frfernandezdev commented Feb 5, 2020

Great contribution, thanks very much

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment