Useful SQLAlchemy Mixins

audit_mixin.py

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

base_mixin.py

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__()

child.py

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

dict_serializable.py

# 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()))

Hi Michael!
I occasionally found your BaseMixin and just wanted to please you that I implemented the same ideas in my package https://github.com/absent1706/sqlalchemy-mixins#active-record

Also I saw your DictSerializableMixin and realized I should implement it in my package too. Thanks!

Hi Michael,

Please, could you write down an example about using function get_by method?

Am stuck trying it from CLI non results.

Thanks 😊