从Flask到FastAPI的平滑迁移

vlambda
2020-12-30

从Flask到FastAPI的平滑迁移

作者 | Ng Wai Foong

译者 | 马可薇

策划 | 田晓旭

迁移 Flask 服务器以获得更好的性能和可维护性。

本文最初发布于 BetterProgramming，经原作者授权由 InfoQ 中文站翻译并分享。

本文将介绍 FastAPI 背后的基本概念，以及将 Flask 服务器过渡到 FastAPI 服务器所涉及的步骤和代码对比以供参考。根据官方文档，FastAPI 的框架是：

...... 一个现代、快速（高性能）的 Web 框架，基于标准 Python 类型提示，使用 Python 3.6+ 构建 API。

众所周知，Flask 是百分百 WSGI（Web Server Gateway Interface，Web 服务器网关接口）的微型 web 框架。随着发展，ASGI（Asynchronous Server Gateway Interface，异步服务器网关接口）作为 WSGI 精神的继承者，实现了在 I/O 绑定的语境下的高吞吐量，支持 HTTP/2 以及 WebSockets，这些都是 WSGI 所不能及的。

随着科技的发展，快如闪电的 ASGI 服务器 Uvicorn 诞生了。然而，Uvicorn 也仅仅只是一个不具备任何路由功能的 web 服务器。Starlette 的出现则是在 ASGI 的服务器（Uvicorn、Daphne，以及 Hypercorn）的基础上提供了一套完整的 ASGI 工具箱。如果要说这二者有什么直接的区别，Starlette 是 ASGI 的 web 框架，而 Flask 则是 WSGI 的 web 框架。

FastAPI 框架充分利用 Starlette 的功能和 Flask 的编程风格，打造出了一款类 Flask 的 ASGIweb 框架。除此之外，作为创建 RESTful API 的理想 web 框架，FastAPI 还包含以下功能：

数据校验。使用 Pydantic 确保运行时强制执行类型提示，数据无效时会有用户友好的错误提示。

文档生成。支持 JSON Scheme 自动生成的数据模型文档，自带 Swagger UI 和 ReDoc 两个交互式 API 文档。

尽管 Flask 和 FastAPI 这两种框架在代码编写上所需时间基本一致，但 FastAPI 自带的 Web 服务器数据校验和数据模型文档生成可以让团队的开发流程更加顺利。

下面让我们动起手来，开始安装必要的模块。

第一步：配置

强烈建议在开始安装前先搭建一个虚拟环境。如果你只想试水 FastAPI，那么安装 FastAPI 和 Uvicorn 即可，其余的安装包都是可选项，本篇教程中也会有所讲解。

FastAPI

安装过程非常简单，运行 pip install 即可。

pip install fastapi

Uvicorn

FastAPI 的 ASGI 服务器建议使用 Uvicorn。同样使用 pip install 安装。

pip install uvicorn

Jinja2（可选）

任何模板引擎都可以用于 FastAPI 的网页渲染，简单起见，这里使用 Flask 里通用的模板引擎 Jinja2。

pip install jinja2

Aiofiles（可选）

如果想要渲染静态文件，那么你还需要安装 aiofiles。

pip install aiofiles

Python-multipart（可选）

默认下，FastAPI 会将输入请求标准化为 JSON，如果需要接收 form 字段，那么你还需要安装 python-multipart。

pip install python-multipart

Flask（可选）

本篇教程主要为 Flask 服务器与 FastAPI 在功能性上的对比演示，如果你想一步一步跟着做，那么请安装 Flask 以供对比。否则，请直接无视本段，直接运行 FastAPI 服务器即可。

pip install flask

下面，让我们进入正题，开始应用阶段。

第二步：对比

这一部分将展示 Flask 服务器与 FastAPI 服务器在相同 API 和功能中的代码对比。

导入（Flask）

Flask 的所有东西都捆绑在它的安装包中，所以导入声明非常直接。random 模块是用于在后续 API 中生成随机数的。

from flask import Flask, request, jsonify, render_template, send_from_directory
import random

导入（FastAPI）

FastAPI 的导入声明则被归类到不同的包中，所以看起来很复杂。下面的代码块导入了对表单字段输入的支持、对返回不同类型返回的支持，以及通过模板引擎渲染静态文件和 HTML 文件的支持。

from fastapi import FastAPI, Form, Request
from fastapi.responses import PlainTextResponse, HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
import random
import uvicorn

最最基础的导入应该长这个样子：

from fastapi import FastAPI
from pydantic import BaseModel
import random # needed for generating a random number for an API
import uvicorn # optional if you run it directly from terminal

初始化（Flask）

Flask 可以处理静态文件及模板引擎，初始化代码如下：

app = Flask(__name__)

初始化（FastAPI）

除了标准初始化流程，还需“挂载”静态文件路径。同样，模板引擎渲染也需要声明一个变量。大部分的初始化代码都是在用基于 Pydantic 的语法创建数据模型类。

app = FastAPI()
# 可选，用于渲染静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")
# 可选，用于模板引擎渲染网页
templates = Jinja2Templates(directory="templates")
# 视使用情况
class Item(BaseModel):
  language = 'english'

Hello World（Flask）

创建一个可以返回字符串的路由。

@app.route('/')
def hello():
    return "Hello World!"

Hello World（FastAPI）

FastAPI 版本的“Hello World”如下。因为默认返回类型为 JSON，所以需要修改 response_class 到 PlainTextResponse 来返回字符串。

async 字段会让异步代码更简单，虽然不是必需，但除非你的代码不支持异步，否则我建议你最好加上。

@app.get("/", response_class=PlainTextResponse)
async def hello():
    return "Hello World!"

随机数（Flask）

在 Flask 服务器上返回随机生成数字 API 的代码如下。

@app.route('/random-number')
def random_number():
    return str(random.randrange(100))

随机数（FastAPI）

FastAPI 的代码只需简单修改：

@app.get('/random-number', response_class=PlainTextResponse)
async def random_number():
    return str(random.randrange(100))

检查 isAlpha（Flask）

下面我们将测试一个接收名为 text 的查询参数，并返回 JSON 结果的 API。在 Flask 中，这一步是通过路由装饰器设置完成的。在这里我们将其设置为仅接受 GET 请求。

@app.route('/alpha', methods=['GET'])
def alpha():
    text = request.args.get('text', '')
    result = {'text': text, 'is_alpha' : text.isalpha()}
    return jsonify(result)

检查 isAlpha（FastAPI）

首先定义 HTTP 请求方法和装饰器，这在 FastAPI 中被称作操作（operation）。GET 操作需要调用 app.get 来完成；而对于多个功能相同的 HTTP 请求方法则需要将其逻辑打包到一个函数中，然后在其各自的操作中独立调用。

查询参数需要和类型提示一起指定。text: str 表示一个需要字符串查询参数 text。也可以通过指定默认值 text = 'text'，使其成为一个可选参数。

app.get('/alpha')
async def alpha(text: str):
    result = {'text': text, 'is_alpha' : text.isalpha()}
return result

创建新 User（Flask）

添加新数据到数据库，通常需要使用 POST 请求。下面的例子接收两个表单字段，返回一个 JSON 结果。

@app.route('/create-user', methods=['POST'])
def create_user():
  id = request.form.get('id', '0001')
  name = request.form.get('name', 'Anonymous')
  # 用于认证、校验、更新数据库
  data = {'id': id, 'name': name}
  result = {'status_code': '0', 'status_message' : 'Success', 'data': data}
  return jsonify(result)

创建新 User（FastAPI）

POST 请求由 app.post 装饰器处理。默认情况下的实现是基于 JSON 或查询参数的，如果要声明输入参数，则需指定 Form(...)。

@app.post('/create-user')
async def create_user(id: str = Form(...), name: str = Form(...)):
  # 用于认证、校验、更新数据库
  data = {'id': id, 'name': name}
  result = {'status_code': '0', 'status_message' : 'Success', 'data': data}
  return result

更新 Language（Flask）

目前为止，本文已经介绍了查询参数以及表单字段的代码。下面这个例子则将根据 JSON 输入更新一个名为 language 的变量，这类对现有数据更新的操作，建议使用 PUT 方法。展示的代码中没有直接指定 PUT 方法，而是通过条件语句来实现。

@app.route('/update-language', methods=['POST', 'PUT', 'GET', 'DELETE'])
def update_language():
  language = 'english'
  if request.method == 'PUT':
    json_data = request.get_json()
    language = json_data['language']
  return "Successfully updated language to %s" % (language)

更新 Language（FastAPI）

同理，PUT 操作是由 app.put 装饰器处理。在初始化的过程中，我们曾定义过以下的 class：

class Item(BaseModel):
    language = 'english'

然后我们需要将这个 class 作为 item 输入参数的类型提示。直接使用 variable_name.attribute_name 语法调用即可，解析会在后台正确完成。在本例中，我们用 item.language。

@app.put('/update-language', response_class=PlainTextResponse)
async def update_language(item: Item):
    language = item.language
return "Successfully updated language to %s" % (language)

HTML 网页（Flask）

在 Flask 中服务网页相对比较简单，使用 Jinja2 模板引擎即可完成。我们只需要在 templates 的文件夹中声明 HTML 文件，如果你需要提供静态文件，则需要将其放入名为 static 的文件夹中。

下面的代码示例使用 index.html 渲染网页，变量可以作为输入参数传入：

@app.route('/get-webpage', methods=['GET'])
def get_webpage():
    return render_template('index.html', message="Contact Us")

HTML 网页（FastAPI）

在初始化过程中，我们“挂载”了一个 static 文件夹，用于提供静态文件：

app.mount("/static", StaticFiles(directory="static"), name="static")

基于 Jinja2Templates 创建了一个变量：

templates = Jinja2Templates(directory="templates")

这些都是用于渲染 HTML 模板的。为提供 HTML 网页，我们需要将 response_class 改为 HTMLResponse。如果你用的不是模板引擎，那么可以直接将结果返回为字符串。

Request 参数需要返回模板以及自定义参数。

@app.get('/get-webpage', response_class=HTMLResponse)
async def get_webpage(request: Request):
    return templates.TemplateResponse("index.html", {"request": request, "message": "Contact Us"})

文件响应（Flask）

若要将文件返回给用户，最佳的处理方式是通过内置函数 send_from_directory，如果路径或文件是通过用户输入获得，那么则更应如此。该内置函数接受两个主要输入：

文件路径

文件名

另外，你也可以额外声明其他参数，诸如：as_attachment，通过修改 Content-Disposition 头来指定其为附件。

路径参数可以通过 <type:variable_name>语法来指定。在本例中，我们用 <string:language>。

@app.route('/get-language-file/<string:language>', methods=['GET'])
def get_language_file(language):
    return send_from_directory('./static/language', language + '.json', as_attachment=True)

文件响应（FastAPI）

FastAPI 根据要求和需要，提供了相当多的响应类型。如果需要返回文件，可以用 FileResponse 或 StreamingResponse。在本文中我们将展示 FileResponse 的使用案例，它接受以下输入：

path：需要流式传输的文件路径

headers：任何自定义头，以字典形式输入

media_type：给定媒体类型的字符串。默认通过文件名或路径推断媒体类型。

filename：设置后，会被响应的 Content-Disposition 引用。

代码展示：

@app.get('/get-language-file/{language}')
async def get_language_file(language: str):
  file_name = "%s.json" % (language)
  file_path = "./static/language/" + file_name
return FileResponse(path=file_path, headers={"Content-Disposition": "attachment; filename=" + file_name})

主函数（Flask）

Flask 中主函数应如下：

if __name__ == '__main__':
    app.run('0.0.0.0',port=8000)

然后在终端中通过这条命令运行文件：

python myapp.py

主函数（FastAPI）

FastAPI 则需要导入 uvicorn。

import uvicorn

并且用如下方法指定主函数（myapp 为文件名，app 为 FastAPI 实例所声明的变量名）：

if __name__ == '__main__':
    uvicorn.run('myapp:app', host='0.0.0.0', port=8000)

然后在终端中正常运行即可：

python myapp.py

更好的方法则是不调用主函数，在终端中通过 uvicorn 直接运行。

uvicorn myapp:app

还可以再额外指定一些参数，诸如：

reload：启用自动加载功能，修改文件后会刷新服务器。对本地开发非常有用。

port：服务器端口，默认为 8000。

这条代码可以将端口号改为 5000：

uvicorn myapp:app --reload --port 5000

Flask 服务器

以下为 Flask 服务器中的完整代码。

# 导入声明
from flask import Flask, request, jsonify, render_template, send_from_directory
import random
# 初始化
app = Flask(__name__)

# hello world，GET 方法，返回字符串
@app.route('/')
def hello():
  return "Hello World!"
# 随机数，GET 方法，返回字符串
@app.route('/random-number')
def random_number():
  return str(random.randrange(100))

# 检查 isAlpha，GET 方法，查询参数，返回 JSON
@app.route('/alpha', methods=['GET'])
def alpha():
  text = request.args.get('text', '')
  result = {'text': text, 'is_alpha' : text.isalpha()}
  return jsonify(result)

# 创建新 user，POST 方法，表单字段，返回 JSON
@app.route('/create-user', methods=['POST'])
def create_user():
  id = request.form.get('id', '0001')
  name = request.form.get('name', 'Anonymous')

  # 用于认证、校验、更新数据库 
  data = {'id': id, 'name': name}
  result = {'status_code': '0', 'status_message' : 'Success', 'data': data}
  return jsonify(result)

# 更新 language，PUT 方法，JSON 输入，返回字符串
@app.route('/update-language', methods=['POST', 'PUT', 'GET', 'DELETE'])
def update_language():
  language = 'english'

  if request.method == 'PUT':
    json_data = request.get_json()
    language = json_data['language']

  return "Successfully updated language to %s" % (language)

# 服务网页，GET 方法，返回 HTML
@app.route('/get-webpage', methods=['GET'])
def get_webpage():
  return render_template('index.html', message="Contact Us")

# 文件响应，GET 方法，返回文件为附件
@app.route('/get-language-file/<string:language>', methods=['GET'])
def get_language_file(language):
  return send_from_directory('./static/language', language + '.json', as_attachment=True)

# main
if __name__ == '__main__':
  app.run('0.0.0.0',port=8000)

FastAPI 服务器

以下代码是使用 FastAPI 实现的相同功能。

# 导入声明
from fastapi import FastAPI, Form, Request
from fastapi.responses import PlainTextResponse, HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates

from pydantic import BaseModel
import random
import uvicorn

# 初始化
app = FastAPI()

# “挂载”静态文件夹，用于渲染静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")

# Jinja2 模板，用于利用模板引擎返回网页
templates = Jinja2Templates(directory="templates")

# Pydantic 数据模型 class
class Item(BaseModel):
  #language: str
  language = 'english'

# hello world，GET 方法，返回字符串
@app.get("/", response_class=PlainTextResponse)
async def hello():
  return "Hello World!"

# 随机数，GET 方法，返回字符串
@app.get('/random-number', response_class=PlainTextResponse)
async def random_number():
  return str(random.randrange(100))

# 检查 isAlpha，GET 方法，查询参数，返回 JSON 
@app.get('/alpha')
async def alpha(text: str):
  result = {'text': text, 'is_alpha' : text.isalpha()} 
  return result

# 创建新 user，POST 方法，表单字段，返回 JSON
@app.post('/create-user')
async def create_user(id: str = Form(...), name: str = Form(...)):
  # 用于认证、校验、更新数据库 
  data = {'id': id, 'name': name}
  result = {'status_code': '0', 'status_message' : 'Success', 'data': data}
  return result
# 更新 language，PUT 方法，JSON 输入，返回字符串 
@app.put('/update-language', response_class=PlainTextResponse)
async def update_language(item: Item):
  language = item.language
  return "Successfully updated language to %s" % (language)

# 服务网页，GET 方法，返回 HTML
@app.get('/get-webpage', response_class=HTMLResponse)
async def get_webpage(request: Request):
  return templates.TemplateResponse("index.html", {"request": request, "message": "Contact Us"})

# 文件响应，GET 方法，返回文件为附件 
@app.get('/get-language-file/{language}')
async def get_language_file(language: str):
  file_name = "%s.json" % (language)
  file_path = "./static/language/" + file_name

  return FileResponse(path=file_path, headers={"Content-Disposition": "attachment; filename=" + file_name})

# main
if __name__ == '__main__':
  uvicorn.run('myapp:app', host='0.0.0.0', port=8000)