快速创建Flask Restful API项目

Python必学的两大web框架之一Flask，俗称微框架。它只需要一个文件，几行代码就可以完成一个简单的http请求服务。

但是我们需要用flask来提供中型甚至大型web restful api服务的时候就需要我们来规划项目结构。

Django 框架有自身的生成项目结构的命令和生成app的命令，但是Flask没有。

这篇文章主要介绍使用工具帮助我们进行一些Flask rest api项目初始化的工作

cookiecutter-flask-restful 介绍

cookiecutter-flask-restful是一个基于cookiecutter的Flask rest api项目生成器，你可以用它快速生成Flask restful api项目。

目前支持的python版本：3.6 | 3.7 | 3.8

项目特性

Flask应用工厂函数，使用蓝图

集成Flask命令行接口

数据迁移

JWT登录认证

分页工具类

单元测试

使用环境变量配置

集成swagger UI

主要使用的包

Flask

Flask-RESTful

Flask-Migrate

Flask-SQLAlchemy

Flask-Marshmallow

Flask-JWT-Extended

marshmallow-sqlalchemy

passlib

tox

pytest

factoryboy

dotenv

apispec

快速开始

安装cookiecutter

$ pip3 install cookiecutter

创建第一个项目

$ cookiecutter https://github.com/karec/cookiecutter-flask-restful

一路按回车键，就会生成一个新项目。

项目名字暂且叫myproject, app名字暂且叫myapi。

运行项目

如果项目使用了pipenv管理虚拟环境：

安装依赖：

$ cd myproject $ pipenv install -r requirements.txt # 进入到虚拟环境 $ pipenv shell $ pip install -e . $ myapi db upgrade $ myapi init $ myapi --help

启动项目：

$ pipenv run flask run

测试项目：

登录请求：

$ curl -X POST -H "Content-Type: application/json" -d '{"username": "admin", "password": "admin"}' :5000/auth/login

登录返回：

{ "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1ODYyMjUyNzIsIm5iZiI6MTU4NjIyNTI3MiwianRpIjoiZDM3NmViODktYWFkOC00ODZkLWE1NmUtZDdiNmQ3NjNhYmI2IiwiZXhwIjoxNTg2MjI2MTcyLCJpZGVudGl0eSI6MSwiZnJlc2giOmZhbHNlLCJ0eXBlIjoiYWNjZXNzIn0.-xCHCJHS3lKeiar-Z12HSOwd2DxAb9EeKFg9zMKWHfU", "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1ODYyMjUyNzIsIm5iZiI6MTU4NjIyNTI3MiwianRpIjoiMTcyNmI0ODEtMDJjYi00NWFhLTgyOWUtOGY0ODJlYmFlZTEyIiwiZXhwIjoxNTg4ODE3MjcyLCJpZGVudGl0eSI6MSwidHlwZSI6InJlZnJlc2gifQ.fpFH_vw8Lp-_xjDVoFLjF6HnjOKXR7PCLKIp8v9gSSU" }

获取所有用户请求：

curl -X GET -H "Content-Type: application/json" -H "Authorization: Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE1ODYyMjUyNzIsIm5iZiI6MTU4NjIyNTI3MiwianRpIjoiZDM3NmViODktYWFkOC00ODZkLWE1NmUtZDdiNmQ3NjNhYmI2IiwiZXhwIjoxNTg2MjI2MTcyLCJpZGVudGl0eSI6MSwiZnJlc2giOmZhbHNlLCJ0eXBlIjoiYWNjZXNzIn0.-xCHCJHS3lKeiar-Z12HSOwd2DxAb9EeKFg9zMKWHfU" :5000/api/v1/users

获取所有用户返回：

{ "total": 3, "pages": 1, "next": "/api/v1/users?page=1&per_page=50", "prev": "/api/v1/users?page=1&per_page=50", "results": [ { "active": true, "id": 1, "email": "admin@mail.com", "username": "admin" }, { "active": true, "id": 2, "email": "", "username": "" }, { "active": true, "id": 3, "email": "string", "username": "string" } ] }

刷新token请求：

curl -X POST -H "Content-Type: application/json" -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZGVudGl0eSI6MSwiaWF0IjoxNTEwMDAwNDQxLCJ0eXBlIjoicmVmcmVzaCIsImp0aSI6IjRmMjgxOTQxLTlmMWYtNGNiNi05YmI1LWI1ZjZhMjRjMmU0ZSIsIm5iZiI6MTUxMDAwMDQ0MSwiZXhwIjoxNTEyNTkyNDQxfQ.SJPsFPgWpZqZpHTc4L5lG_4aEKXVVpLLSW1LO7g4iU0" :5000/auth/refresh

刷新token返回：

{ "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0eXBlIjoiYWNjZXNzIiwiaWRlbnRpdHkiOjEsImlhdCI6MTUxMDAwMDYxOCwiZnJlc2giOmZhbHNlLCJqdGkiOiIzODcxMzg4Ni0zNGJjLTRhOWQtYmFlYS04MmZiNmQwZjEyNjAiLCJuYmYiOjE1MTAwMDA2MTgsImV4cCI6MTUxMDAwMTUxOH0.cHuNf-GxVFJnUZ_k9ycoMMb-zvZ10Y4qbrW8WkXdlpw" }

如果没有使用pipenv管理虚拟环境，直接使用的系统python环境:

$ cd myproject $ pip install -r requirements.txt $ pip install -e . # 初始化项目 $ myapi db upgrade $ myapi init $ myapi --help

测试项目同上

项目环境变量配置

环境变量在.flaskenv文件中

FLASK_ENV=development FLASK_APP="myapp.app:create_app" SECRET_KEY=changeme DATABASE_URI="sqlite:////tmp/myapp.db" CELERY_BROKER_URL=amqp://guest:guest@localhost/ # only present when celery is enabled CELERY_RESULT_BACKEND_URL=amqp://guest:guest@localhost/ # only present when celery is enabled

请根据你的实际情况进行配置。

测试

直接用tox

$ tox

使用pytest

$ pip install pytest pytest-runner pytest-flask pytest-factoryboy factory_boy $ pytest

使用docker-compose 和 Makefile

make tests

如果出现连接失败，你需要更改redis | rabbitmq地址

使用wsgi服务启动

gunicorn示例:

$ pip install gunicorn $ gunicorn myapi.wsgi:app

uwsgi示例：

$ pip install uwsgi $ uwsgi --http 127.0.0.1:5000 --module myapi.wsgi:app 使用celery

启动celery

$ celery worker -A myapi.celery_app:app --loglevel=info

如果你的配置没有问题，在终端你会看到：

[tasks] . myapi.tasks.example.dummy_task

调用task：

>>> from myapi.tasks.example import dummy_task >>> result = dummy_task.delay() >>> result.get() 'OK'

使用celery extension:

>>> from myapi.extensions import celery >>> celery.send_task('myapi.tasks.example.dummy_task').get() 'OK' 使用Docker

注意：目前Docker的配置只有开发环境

构建镜像：

$ docker build -t myapp .