flask 项目搭建及配置分享
Posted Python中文社区
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了flask 项目搭建及配置分享相关的知识，希望对你有一定的参考价值。
概要
好久没有碰flask框架了，近期写点东西，回忆一下，分享小伙伴入门flask，并分享源代码，见文章底部
拓展flask支持banner， 支持config.properties配置文件导入
模块化设计，支持数据库迁移
封装sqlalchemy数据库操作
自动转json
配置拦截器，异常自动解析(web请求返回错误页面，curl请求返回错误json)
拓展flask内置函数，支持环境变量
集成celery框架异步处理
支持docker构建
flask jinja2模板示例
swagger api文档配置
等等
模块结构图
.
.
├── banner.txt
├── bootstrap_app.py
├── bootstrap_celery.py
├── config.properties
├── config.py
├── Dockerfile
├── examples
│ ├── extensions_flask_form.py
│ ├── extensions_flask_SQLAlchemy.py
│ ├── hello_world.py
│ ├── index.py
│ ├── __init__.py
│ └── rest_api.py
├── flaskapp
│ ├── common
│ │ ├── error_view.py
│ │ ├── exceptions.py
│ │ ├── __init__.py
│ │ ├── logger.py
│ │ ├── response.py
│ │ ├── tools.py
│ │ └── utils.py
│ ├── core
│ │ ├── database.py
│ │ ├── http_handler.py
│ │ ├── http_interceptor.py
│ │ └── __init__.py
│ ├── extends
│ │ ├── banner.py
│ │ ├── functions.py
│ │ └── __init__.py
│ ├── __init__.py
│ ├── models
│ │ ├── base.py
│ │ ├── clazz.py
│ │ ├── __init__.py
│ │ ├── school.py
│ │ └── user.py
│ ├── plugins
│ │ ├── flask_celery.py
│ │ └── __init__.py
│ ├── services
│ │ ├── base.py
│ │ ├── __init__.py
│ │ └── statement.py
│ └── views
│ ├── async_handler.py
│ ├── error_handler.py
│ ├── index_hander.py
│ ├── __init__.py
│ ├── rest_clazz_handler.py
│ ├── rest_login_handler.py
│ ├── rest_school_handler.py
│ └── rest_user_handler.py
├── git-user-config.sh
├── README.md
├── requirements.txt
├── static
│ ├── css
│ │ └── layout.css
│ ├── favicon.ico
│ ├── images
│ │ └── 0.jpg
│ └── js
│ └── app.js
├── stop-app.sh
├── templates
│ ├── 404.html
│ ├── examples
│ │ ├── extensions_flask_form.html
│ │ └── extensions_flask_sqlAlchemy.html
│ ├── index.html
│ └── layout.html
└── test
├── config.properties
├── __init__.py
├── plugins
│ ├── __init__.py
│ └── test_celery_task.py
├── test_banner.py
├── test_celery.py
├── test_db.py
├── test_extend_func.py
├── test_lru.py
├── test_platform.py
└── views
└── test_school.py
数据库封装
class Database(object):
"""
database interface
"""
class Transactional(Database):
def __init__(self, **kwargs):
"""
事务层
:param auto_commit: 是否自动提交
"""
self._auto_commit = kwargs.get('auto_commit', True)
self.model = kwargs.get('model_class')
if not self.model:
raise AssertionError('<class {}>: Required parameter model_class is not present.'
.format(self.__class__.__name__))
self.session = db.session
# logger.info('init Transactional')
def auto_commit(self):
"""
是否自动提交事务
:return:
"""
if self._auto_commit:
self.session.commit()
def _check_type(self, obj):
if not isinstance(obj, self.model):
raise AssertionError('obj must be <class {}> type.'
.format(self.model.__class__.__name__))
class Persistence(Transactional):
def __init__(self, **kwargs):
super(Persistence, self).__init__(**kwargs)
# logger.info('init Persistence')
class Modify(Transactional):
def __init__(self, **kwargs):
super(Modify, self).__init__(**kwargs)
# logger.info('init Modify')
class Remove(Transactional):
def __init__(self, **kwargs):
super(Remove, self).__init__(**kwargs)
# logger.info('init Remove')
class Query(Database):
def __init__(self, **kwargs):
# logger.info('init Query')
self.model = kwargs.get('model_class', None)
if not self.model:
raise AssertionError('<class {}>: model_class is not found.'
.format(self.__class__.__name__))
class Modify2(Database):
@classmethod
def _auto_commit(cls):
db.session.commit()
class Query2(Database):
def __init__(self):
"""需要传入实体类型来使用该类"""
# logger.info('init Query2')
banner 配置
def _banner():
banner_path = os.path.join(
os.path.dirname(os.path.dirname(
os.path.dirname(__file__))), 'banner.txt')
if os.path.exists(banner_path) and os.path.isfile(banner_path):
with open(banner_path) as f:
for line in f:
print line.rstrip(' ')
else:
print banner_text
接口浏览
$ curl localhost:5000/paths
错误处理
页面请求：
curl请求：
$ curl localhost:5000/api/vi/students/err
级联查询转json
def json(self, lazy=False, ignore=None, deep=1):
"""
转化json
:param lazy: 是否懒加载
:param ignore: 过滤属性
:param deep： 当前深度
:return:
"""
ignore_filed = ['query', 'metadata'] + ignore if isinstance(ignore, list) else ['query', 'metadata', ignore]
def _filter_filed(obj):
return filter(lambda y: all(
(
y not in ignore_filed,
not y.endswith('_'),
not y.startswith('_'),
not callable(getattr(obj, y))
)), dir(obj))
def _get_ignore_filed(base_obj, obj, _filed_list):
_ignore_filed = []
for _filed in _filed_list:
_filed_obj = getattr(obj, _filed)
if isinstance(_filed_obj, BaseQuery):
_primary_entity = '%s' % _filed_obj.attr.target_mapper
if _primary_entity.split('|')[1] == base_obj.__class__.__name__:
_ignore_filed.append(_filed)
if isinstance(_filed_obj, type(base_obj)):
_ignore_filed.append(_filed)
return _ignore_filed
__relationship__, res, filed_list = None, {}, _filter_filed(self)
for filed in filed_list:
filed_type = getattr(self, filed)
if filed == __relationship__:
continue
if isinstance(filed_type, DictModel):
_ignore = _get_ignore_filed(self, filed_type, _filter_filed(filed_type))
relationship_model = filed_type.json(ignore=_ignore, deep=deep + 1)
if not lazy:
res[filed] = relationship_model
elif isinstance(filed_type, (int, list)):
res[filed] = filed_type
elif isinstance(filed_type, BaseQuery):
res[filed] = []
if not lazy:
for f in filed_type.all():
_ignore = _get_ignore_filed(self, f, _filter_filed(f))
res[filed].append(f.json(lazy=lazy, ignore=_ignore, deep=deep + 1))
else:
try:
if isinstance(filed_type, unicode):
filed_type = filed_type.encode('UTF-8')
res[filed] = '{}'.format(filed_type)
except (UnicodeEncodeError, Exception), e:
logger.error('{class_name}.{filed}: {e}'.format(
class_name=self.__class__.__name__, filed=filed, e=e))
res[filed] = None
return res
拓展flask启动方法start
from flaskapp import app
if __name__ == "__main__":
app.start()
# app.start(port=5258, debug=False)
数据库更新迁移
$ python manager.py db init
$ python manager.py db migrate
Dockerfile 构建
$ ./docker-build.sh
celery异步处理
见项目test目录test_celery.py
@celery.task()
def async_compute(a, b):
from time import sleep
sleep(10)
return a + b
@cly.route('/compute')
@json_encoder
def task():
result = async_compute.delay(1, 2)
print result.wait()
return 'task id: {}'.format(result)
swagger配置
见项目examples目录swaggerforapi.py
Python中文社区作为一个去中心化的全球技术社区，以成为全球20万Python中文开发者的精神部落为愿景，目前覆盖各大主流媒体和协作平台，与阿里、腾讯、百度、微软、亚马逊、开源中国、CSDN等业界知名公司和技术社区建立了广泛的联系，拥有来自十多个国家和地区数万名登记会员，会员来自以公安部、工信部、清华大学、北京大学、北京邮电大学、中国人民银行、中科院、中金、华为、BAT、谷歌、微软等为代表的政府机关、科研单位、金融机构以及海内外知名公司，全平台近20万开发者关注。
▼点击阅读原文加入
Python Web开发学习小分队！
以上是关于flask 项目搭建及配置分享的主要内容，如果未能解决你的问题，请参考以下文章