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

Very nice collection of mixins

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

@foxx: +1 for utcnow

I will stick to now. instead of utcnow..

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

Great contribution, thanks very much

Check out https://github.com/BemiHQ/bemi-sqlalchemy if you want to automatically audit data changes in tables without changing the database structure (works with FastAPI)