Skip to content

Instantly share code, notes, and snippets.

@lu911
Created December 13, 2014 17:52
Show Gist options
  • Save lu911/851967b04d1f7cb6ba56 to your computer and use it in GitHub Desktop.
Save lu911/851967b04d1f7cb6ba56 to your computer and use it in GitHub Desktop.
SQLAlchemy Mixins
# -*- coding:utf-8 -*-
from flask import abort
from datetime import datetime
from project.ext import db
class IdMixin(object):
"""
Provides the :attr:`id` primary key column
"""
#: Database identity for this model, used for foreign key
#: references from other models
id = db.Column(db.Integer, primary_key=True)
class TimestampMixin(object):
"""
Provides the :attr:`created_at` and :attr:`updated_at` audit timestamps
"""
#: Timestamp for when this instance was created, in UTC
created_at = db.Column(db.DateTime, default=datetime.now, nullable=False)
#: Timestamp for when this instance was last updated (via the app), in UTC
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)
class CRUDMixin(object):
__table_args__ = {'extend_existing': True}
@classmethod
def query(cls):
return db.session.query(cls)
@classmethod
def get(cls, _id):
if any((isinstance(_id, basestring) and _id.isdigit(),
isinstance(_id, (int, float))),):
return cls.query.get(int(_id))
return None
@classmethod
def get_by(cls, **kwargs):
return cls.query.filter_by(**kwargs).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, **kwargs):
r = cls.get_by(**kwargs)
if not r:
r = cls(**kwargs)
db.session.add(r)
return r
@classmethod
def create(cls, **kwargs):
instance = cls(**kwargs)
return instance.save()
def update(self, commit=True, **kwargs):
for attr, value in kwargs.iteritems():
setattr(self, attr, value)
return commit and self.save() or self
def save(self, commit=True):
db.session.add(self)
if commit:
try:
db.session.commit()
except Exception:
db.session.rollback()
raise
return self
def delete(self, commit=True):
db.session.delete(self)
return commit and db.session.commit()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment