Last active
December 2, 2015 09:18
-
-
Save TheWaWaR/063eadc42ddecc0625e8 to your computer and use it in GitHub Desktop.
A way to write flask resultful view
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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