flask 项目搭建及配置分享

作者:Tom .Lee,GitHub ID:tomoncle ,Web and cloud computing developer, Java, Golang, Python enthusiast.

概要

好久没有碰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('\n')
    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
  • 源码地址:https://github.com/tomoncle/flaskapp

原文发布于微信公众号 - Python中文社区(python-china)

原文发表时间:2018-07-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏自动化测试实战

Flask第六篇——项目配置文件

39670
来自专栏深度学习自然语言处理

有关vi(vim)的常用命令

导读 vi(vim)是上Linux非常常用的编辑器,很多Linux发行版都默认安装了vi(vim)。vi(vim)命令繁多但是如果使用灵活之后将会大大提高效率。...

33460
来自专栏积累沉淀

Java 命令行运行参数大全

javac 用法:javac <选项> <源文件> 其中,可能的选项包括:   -g                                   ...

1.3K100
来自专栏互联网高可用架构

SSM源码解密-第2章 Spring MVC整体架构第2章 Spring MVC整体架构

11720
来自专栏乐沙弥的世界

RAC 环境下的重要参数

    Oracle 数据库启动时会根据参数文件中提供的相关参数启动Oracle实例。这些参数包括数据库名字、sga,pga的分配,控制文件的位置,undo,p...

7710
来自专栏马洪彪

Java设计模式(三)单例模式

一、场景描述 在采集到仪器数据后,需要将数据发送到lims系统中,通过调用lims系统服务实现数据的上传。 在仪器数据采集组件中实现lims系统服务代理,该代理...

36160
来自专栏Jerry的SAP技术分享

一个简单的例子理解Kubernetes的三种IP地址类型

很多Kubernetes的初学者对Kubernetes里面三种不同的IP地址和工作机制理解得不是很清楚。

15300
来自专栏日常分享

JSP/Servlet Web 学习笔记 DayThree

  使用JSP语法可以存取这些内置对象来执行JSP网页的Servlet环境相互作用。内置对象其实是由特定的Java类所产生的。每一种内置对象都映射到一个特定的J...

11320
来自专栏运维小白

4.3/4.4 磁盘分区

添加虚拟磁盘 第一步,选择虚拟机中的“设置” ? 第二步,选择“添加硬盘” ? 第三步,选择_SCSI (推荐) # 保持默认 ? 第四...

26350
来自专栏技巅

Glusterfs之rpc模块源码分析(中)之Glusterfs的rpc模块实现(3)

22740

扫码关注云+社区

领取腾讯云代金券