Skip to content

Instantly share code, notes, and snippets.

@TheWaWaR
Last active December 2, 2015 09:18
Show Gist options
  • Save TheWaWaR/063eadc42ddecc0625e8 to your computer and use it in GitHub Desktop.
Save TheWaWaR/063eadc42ddecc0625e8 to your computer and use it in GitHub Desktop.
A way to write flask resultful view
class SessionMixin(object):
""" Common methods for model classes """
def to_dict(self):
data = {}
for c in self.__class__.__table__.columns:
value = getattr(self, c.name)
if isinstance(value, datetime):
data['{}_str'.format(c.name)] = value.strftime("%Y-%m-%d %H:%M:%S")
value = datetime_to_utcts(value)
elif isinstance(value, date):
data['{}_str'.format(c.name)] = value.strftime("%Y-%m-%d")
value = datetime_to_utcts(value)
data[c.name] = value
return data
class BaseModel(db.Model, SessionMixin):
__abstract__ = True
class BaseMethodView(MethodView):
# The blueprint of this view
blueprint = None
# The endpoint name
endpoint = ''
# Url rules:
# API: http://flask.pocoo.org/docs/0.10/api/#flask.Flask.url_map
# Example: http://flask.pocoo.org/docs/0.10/views/
url_rules = []
# Default data model (subclass of Flask-SQLAlchemy.Model)
Modal = None
@classmethod
def register_urls(cls, bp=None):
""" Call this classmethod before register blueprints to flask app:
def register_blueprints(app, bps):
for view_cls in BaseMethodView.__subclasses__():
view_cls.register_urls()
for bp in bps:
app.register_blueprint(bp)
"""
view = cls.as_view(cls.endpoint)
if bp is None:
bp = cls.blueprint
for args, kwargs in cls.url_rules:
kwargs.setdefault('view_func', view)
bp.add_url_rule(*args, **kwargs)
def get_list(self):
""" Default method for get a page of objects """
return QueryProcessor.build(request, self.Modal)
def get_one(self, oid):
""" Get object by id (primary key)"""
obj = self.Modal.query.get_or_404(oid)
return obj.to_dict()
def get(self, oid):
if oid is None:
return self.get_list()
else:
return self.get_one(oid)
class QueryProcessor():
FILTER_DICT = {
# f ==> field; v ==> value;
'contains' : lambda f, v: f.contains(v),
'~contains' : lambda f, v: ~f.contains(v),
'ilike' : lambda f, v: f.ilike(u'%{}%'.format(v)),
'~ilike' : lambda f, v: ~f.ilike(u'%{}%'.format(v)),
'like' : lambda f, v: f.like(u'%{}%'.format(v)),
'~like' : lambda f, v: ~f.like(u'%{}%'.format(v)),
'in' : lambda f, v: f.in_(v),
'~in' : lambda f, v: ~f.in_(v),
'==' : lambda f, v: f == v,
'!=' : lambda f, v: f != v,
'>' : lambda f, v: f > v,
'>=' : lambda f, v: f >= v,
'<' : lambda f, v: f < v,
'<=' : lambda f, v: f <= v,
}
def __init__(self, filters=tuple(), sort=tuple(), page=1, perpage=20, model=None):
self.filters = filters
self.sort = sort
self.page = page
self.perpage = perpage
self.model = model
self.query = model.query
def resolve(self):
"""
Steps:
=====
1. filter
1.1 Check total records.
2. sort
3. offset
4. limit
"""
Model = self.model
query = self.query
if query is None:
raise ValueError('DB Model query not given: {}'.format(str(self)))
filter_dict = QueryProcessor.FILTER_DICT
def gen_filter_cond(tModel, name, op, value):
if op not in filter_dict:
raise InvalidQueryOperator(op)
field = getattr(tModel, name)
filter_func = filter_dict[op]
return filter_func(field, value)
filters = self.filters
orderBy = self.sort
offset = self.perpage * (self.page - 1)
limit = self.perpage
# 1. Filter
filter_conds = [gen_filter_cond(Model, name, op, value)
for name, op, value in filters if value is not None]
query = query.filter(db.and_(*filter_conds))
total = query.count()
if 0 < total <= offset:
raise PageOverflow(str(self), offset, total)
# 2. Sort
orderBy_conds = [getattr(getattr(Model, name), order)()
for name, order in orderBy]
query = query.order_by(*orderBy_conds)
# 3. Offset
query = query.offset(offset)
# 4. Limit: if limit <=0, then get all records
if limit > 0:
query = query.limit(limit)
return total, query
def get_rv(self):
total, query = self.resolve()
return {
'total': total,
'objects': [obj.to_dict() for obj in query.all()]
}
@staticmethod
def build(request, model):
args = request.args.get('q', '{}')
args = json.loads(args)
page = args.get('page', current_app.config['DEFAULT_PAGE'])
perpage = args.get('perpage', current_app.config['DEFAULT_PERPAGE'])
filters = args.get('filters', [])
sort = args.get('sort', [])
Meta = getattr(model, 'Meta', object())
filters = filters or getattr(Meta, 'default_filters', [])
sort = sort or getattr(Meta, 'default_sort', []) or [["id", "desc"]]
return QueryProcessor(filters, sort, page, perpage, model)
def update_filters(self, callback):
self.filters = callback(self.filters)
def update_sort(self, callback):
self.sort = callback(self.sort)
def __str__(self): return unicode(self).encode('utf-8')
def __unicode__(self):
return u'<QueryProcessor(page={}, perpage={}, filters={}, sort={})>'.format(
self.page, self.perpage, self.filters, self.sort)
class UserView(BaseMethodView):
blueprint = bp
endpoint = 'users'
url_rules = [
(['/users/'],
{'defaults': {'oid': None}, 'methods': ['GET']}),
(['/users/'],
{'methods': ['POST']}),
(['/users/<int:oid>'],
{'methods': ['GET', 'PUT', 'DELETE']}),
]
Modal = User
def post(self):
return ('ok', 201)
def put(self, oid):
return 'PUT:{}'.format(oid)
def delete(self, oid):
return 'DELETE:{}'.format(oid)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment