从 Flask 迁移到 FastAPI_flask转fastapi

---

Python 是最流行的编程语言之一。从脚本到 API 开发再到机器学习，Python 都有着它自己的足迹。因为 Python 注重开发者的体验和其所能提供的大量工具而大受欢迎。网络框架 Flask 就是这样一个工具，它在机器学习社区中很受欢迎。它也被广泛用于 API开发。但是有一个新的框架正在崛起： FastAPI。与 Flask 不同，FastAPI 是一个 ASGI（Asynchronous Server Gateway Interface 异步服务器网关接口）框架。与 Go 和 NodeJS 一样，FastAPI 是最快的基于 Python 的 Web 框架之一。

本文针对那些有兴趣从 Flask 转移到 FastAPI 的人，比较和对比了 Flask 和 FastAPI 的常见模式。

FastAPI vs Flask

FastAPI 的构建考虑了以下三个主要问题：

• 速度
• 开发者经验
• 开放标准

你可以把 FastAPI 看作是把 Starlette、Pydantic、OpenAPI 和 JSON Schema 粘合在一起的胶水。

• 本质上说，FastAPI 使用 Pydantic 进行数据验证，并使用 Starlette 作为工具，使其与 Flask 相比快得惊人，具有与 Node 或 Go 中的高速 Web APIs 相同的性能。

• Starlette + Uvicorn 提供异步请求能力，这是 Flask 所缺乏的。

• 有了 Pydantic 以及类型提示，你就可以得到一个具有自动完成功能的良好的编辑体验。你还可以得到数据验证、序列化和反序列化（用于构建一个 API），以及自动化文档（通过 JSON Schema 和 OpenAPI ）。

也就是说，Flask 的使用更为广泛，所以它经过了实战检验，并且有更大的社区支持它。由于这两个框架都是用来扩展的，Flask 显然是赢家，因为它有庞大的插件生态系统。

建议：

• 如果你对上述三个问题有共鸣，厌倦了 Flask 扩展时的大量选择，希望利用异步请求，或者只是想建立一个 RESTful API，请使用 FastAPI。
• 如果你对 FastAPI 的成熟度不满意，需要用服务器端模板构建一个全栈应用，或者离不开一些社区维护的 Flask 扩展，就可以使用 Flask。

开始

安装

与任何其他 Python 包一样，安装非常简单。

Flask

pip install flask

# or
poetry add flask
pipenv install flask
conda install flask
1
2
3
4
5
6

FastAPI

pip install fastapi uvicorn

# or
poetry add fastapi uvicorn
pipenv install fastapi uvicorn
conda install fastapi uvicorn -c conda-forge
1
2
3
4
5
6

与 Flask 不同，FastAPI 没有内置的开发服务器，因此需要像 Uvicorn 或 Daphne 这样的 ASGI 服务器。

“Hello World” 应用

Flask

# flask_code.py

from flask import Flask

app = Flask(__name__)

@app.route("/")
def home():
    return {"Hello": "World"}

if __name__ == "__main__":
    app.run()
1
2
3
4
5
6
7
8
9
10
11
12

FastAPI

# fastapi_code.py

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def home():
    return {"Hello": "World"}

if __name__ == "__main__":
    uvicorn.run("fastapi_code:app")
1
2
3
4
5
6
7
8
9
10
11
12
13

像 reload=True 这样的参数可以被传递到 uvicorn.run() 中，以实现开发时的热重载。

或者，您可以直接从终端启动服务器：

uvicorn run fastapi_code:app
1

热加载模式：

uvicorn run fastapi_code:app --reload
1

配置

Flask 和 FastAPI 都提供了许多选项来处理不同环境的不同配置。两者都支持以下模式：

1. 环境变量
2. 配置文件
3. 实例文件夹
4. 类和继承

有关更多信息，请参阅其各自的文档：

• Flask: https://flask.palletsprojects.com/en/2.0.x/config/

• FastAPI:https://fastapi.tiangolo.com/advanced/settings/

Flask

import os
from flask import Flask

class Config(object):
    MESSAGE = os.environ.get("MESSAGE")

app = Flask(__name__)
app.config.from_object(Config)

@app.route("/settings")
def get_settings():
    return { "message": app.config["MESSAGE"] }

if __name__ == "__main__":
    app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

现在，在你运行服务器之前，设置适当的环境变量：

export MESSAGE="hello, world"
1

FastAPI

import uvicorn
from fastapi import FastAPI
from pydantic import BaseSettings

class Settings(BaseSettings):
    message: str

settings = Settings()
app = FastAPI()

@app.get("/settings")
def get_settings():
    return { "message": settings.message }

if __name__ == "__main__":
    uvicorn.run("fastapi_code:app")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

同样，在运行服务器之前，设置适当的环境变量：

export MESSAGE="hello, world"
1

路由, 模板和视图

HTTP 方法

Flask

from flask import request

@app.route("/", methods=["GET", "POST"])
def home():
    # handle POST
    if request.method == "POST":
        return {"Hello": "POST"}
    # handle GET
    return {"Hello": "GET"}
1
2
3
4
5
6
7
8
9

FastAPI

@app.get("/")
def home():
    return {"Hello": "GET"}

@app.post("/")
def home_post():
    return {"Hello": "POST"}
1
2
3
4
5
6
7

FastAPI 为每个方法提供单独的装饰器：

@app.get("/")
@app.post("/")
@app.delete("/")
@app.patch("/")
1
2
3
4

URL 参数

通过 URL（如 /employee/1 ）传递信息以管理状态：

Flask

@app.route("/employee/<int:id>")
def home():
    return {"id": id}
1
2
3

FastAPI

@app.get("/employee/{id}")
def home(id: int):
    return {"id": id}
1
2
3

URL参数的指定类似于一个 f-string 表达式。此外，你还可以利用类型提示。这里，我们在运行时告诉 Pydantic， id 是 int 类型的。在开发中，这也可以帮助完成更好的代码完成度。

查询参数

与 URL 参数一样，查询参数（如 /employee?department=sales ）也可用于管理状态（通常用于过滤或排序）：

Flask

from flask import request

@app.route("/employee")
def home():
    department = request.args.get("department")
    return {"department": department}
1
2
3
4
5
6

FastAPI

@app.get("/employee")
def home(department: str):
    return {"department": department}
1
2
3

模板

Flask

from flask import render_template

@app.route("/")
def home():
    return render_template("index.html")
1
2
3
4
5

默认情况下，Flask会在 "templates "文件夹中寻找模板。

FastAPI

你需要安装 Jinja：

pip install jinja2
1

实现：

from fastapi import Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse

app = FastAPI()

templates = Jinja2Templates(directory="templates")

@app.get("/", response_class=HTMLResponse)
def home(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})
1
2
3
4
5
6
7
8
9
10
11

对于 FastAPI，你需要明确地定义 "模板 "文件夹。然后对于每个响应，需要提供请求上下文。

静态文件

Flask

默认情况下，Flask 从“static”文件夹中提供静态文件。

FastAPI

在 FastAPI 中，需要为静态文件挂载一个文件夹：

from fastapi.staticfiles import StaticFiles

app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")
1
2
3
4
5

异步任务

Flask

从 Flask 2.0 开始，您可以使用 async/await 创建异步路由处理程序：

@app.route("/")
async def home():
    result = await some_async_task()
    return result
1
2
3
4

有关 Flask 中异步视图的更多信息，请查看 Flask 2.0 中的异步一文。

Flask 中的异步也可以通过使用线程（并发）或多处理（并行）或 Celery 或 RQ 等工具来实现：

1. Asynchronous Tasks with Flask and Celery：https://testdriven.io/blog/flask-and-celery/
2. Asynchronous Tasks with Flask and Redis Queue：https://testdriven.io/blog/asynchronous-tasks-with-flask-and-redis-queue/

FastAPI

由于 FastAPI 对 asyncio 的原生支持，它极大地简化了异步任务。要使用的话，只需在视图函数中添加 async 关键字：

@app.get("/")
async def home():
    result = await some_async_task()
    return result
1
2
3
4

FastAPI 还具有后台任务功能，您可以使用它来定义返回响应后要运行的后台任务。这对于不需要在发送回响应之前完成的操作很有用。

from fastapi import BackgroundTasks

def process_file(filename: str):
    # process file :: takes minimum 3 secs (just an example)
    pass

@app.post("/upload/{filename}")
async def upload_and_process(filename: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_file, filename)
    return {"message": "processing file"}
1
2
3
4
5
6
7
8
9
10

在这里，响应将被即时发送，而不会让用户等待文件处理完成。

当你需要进行繁重的后台计算时，或者你需要一个任务队列来管理任务(tasks)和工作者(workers)时，你可能想使用Celery 而不是 BackgroundTasks。更多内容请参考 FastAPI 和 Celery 的异步任务：https://testdriven.io/blog/fastapi-and-celery/

依赖注入

Flask

虽然你可以实现自己的依赖注入解决方案，但 Flask 默认没有真正的一流支持。相反，你需要使用一个外部包，如 flask-injector。

FastAPI

另一方面，FastAPI 具有处理依赖注入的强大解决方案。

例如：

from databases import Database
from fastapi import Depends
from starlette.requests import Request

from db_helpers import get_all_data
def get_db(request: Request):
    return request.app.state._db

@app.get("/data")
def get_data(db: Database = Depends(get_db)):
    return get_all_data(db)
1
2
3
4
5
6
7
8
9
10
11

因此，get_db 将获取对在应用程序的启动事件处理程序中创建的数据库连接的引用。 Depends 然后用于向 FastAPI 指示路由“依赖于” get_db。因此，它应该在路由处理程序中的代码之前执行，并且结果应该“注入”到路由本身。

数据校验

Flask

Flask 没有任何内部数据验证支持。您可以使用功能强大的 Pydantic 包通过 Flask-Pydantic 进行数据验证。

FastAPI

FastAPI 如此强大的原因之一是它支持 Pydantic。

from pydantic import BaseModel

app = FastAPI()

class Request(BaseModel):
    username: str
    password: str

@app.post("/login")
async def login(req: Request):
    if req.username == "testdriven.io" and req.password == "testdriven.io":
        return {"message": "success"}
    return {"message": "Authentication Failed"}
1
2
3
4
5
6
7
8
9
10
11
12
13

在这里，我们接受一个模型 Request 的输入。该 payload 必须包含一个用户名和密码。

# correct payload format
✗ curl -X POST 'localhost:8000/login' \
    --header 'Content-Type: application/json' \
    --data-raw '{"username": "testdriven.io","password":"testdriven.io"}'

{"message":"success"}

# incorrect payload format
✗ curl -X POST 'localhost:8000/login' \
    --header 'Content-Type: application/json' \
    --data-raw '{"username": "testdriven.io","passwords":"testdriven.io"}'

{"detail":[{"loc":["body","password"],"msg":"field required","type":"value_error.missing"}]}
1
2
3
4
5
6
7
8
9
10
11
12
13

注意到这个请求。我们把密码 passwords 作为一个键而不是 password 传递进去。Pydantic 模型会自动告诉用户，password 字段是缺失的。

序列化和反序列化

Flask

最简单的序列化方法是使用 jsonify：

from flask import jsonify
from data import get_data_as_dict

@app.route("/")
def send_data():
    return jsonify(get_data_as_dict)
1
2
3
4
5
6

对于复杂的对象，Flask 开发者经常使用 Flask-Marshmallow

FastAPI

FastAPI 自动序列化任何返回的字典 dict 。对于更复杂和结构化的数据，使用 Pydantic：

from pydantic import BaseModel

app = FastAPI()

class Request(BaseModel):
    username: str
    email: str
    password: str

class Response(BaseModel):
    username: str
    email: str

@app.post("/login", response_model=Response)
async def login(req: Request):
    if req.username == "testdriven.io" and req.password == "testdriven.io":
        return req
    return {"message": "Authentication Failed"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

在这里，我们添加了一个包含三个输入的 Request 模型：用户名、电子邮件和密码。我们还定义了一个仅包含用户名和电子邮件的 Response 模型。输入 Request 模型处理反序列化，而输出 Response 模型处理对象序列化。然后通过 response_model 参数将响应模型传递给装饰器。

现在，如果我们将请求本身作为响应返回，Pydantic 将省略 password ，因为我们定义的响应模型不包含密码字段。

例如：

# output
✗ curl -X POST 'localhost:8000/login' \
    --header 'Content-Type: application/json' \
    --data-raw '{"username":"testdriven.io","email":"admin@testdriven.io","password":"testdriven.io"}'

{"username":"testdriven.io","email":"admin@testdriven.io"}
1
2
3
4
5
6

中间件

中间件被用来在每个请求被视图功能处理之前应用逻辑。

Flask

class middleware:
    def __init__(self, app) -> None:
        self.app = app

    def __call__(self, environ, start_response):
        start = time.time()
        response = self.app(environ, start_response)
        end = time.time() - start
        print(f"request processed in {end} s")
        return response

app = Flask(__name__)
app.wsgi_app = middleware(app.wsgi_app)
1
2
3
4
5
6
7
8
9
10
11
12
13

FastAPI

from fastapi import Request

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    print(f"request processed in {process_time} s")
    return response
1
2
3
4
5
6
7
8
9

@app.middleware(“http”) 装饰器是在 FastAPI 中创建中间件的必备工具。上述中间件计算处理请求所花费的时间。视图函数处理请求后，计算总处理时间并将其作为响应头返回。

# flask output(logs)
request processed in 0.0010077953338623047 s
127.0.0.1 - - [22/Sep/2020 18:56:21] "GET / HTTP/1.1" 200 -

# fastapi output(logs)
request processed in 0.0009925365447998047 s
INFO:     127.0.0.1:51123 - "GET / HTTP/1.1" 200 OK
1
2
3
4
5
6
7

模块化

随着应用程序的发展，在某些时候你会想把类似的视图、模板、静态文件和模型组合在一起，以帮助把应用程序分解成更小的组件。

Flask

在 Flask 中，蓝图被用来实现模块化：

# blueprints/product/views.py
from flask import Blueprint

product = Blueprint("product", __name__)

@product.route("/product1")
    ...
# main.py

from blueprints.product.views import product

app.register_blueprint(product)
1
2
3
4
5
6
7
8
9
10
11
12

FastAPI

同时，在 FastAPI 中，模块化是通过 APIRouter 实现的：

# routers/product/views.py
from fastapi import APIRouter

product = APIRouter()

@product.get("/product1")
    ...
# main.py

from routers.product.views import product

app.include_router(product)
1
2
3
4
5
6
7
8
9
10
11
12

其他特点

自动文档

Flask

Flask 不会自动创建开箱即用的 API 文档。然而，有几个扩展可以处理这个问题，比如 flask-swagger 和 Flask RESTX，但它们需要额外的设置。

FastAPI

默认情况下，FastAPI 支持 OpenAPI 以及 Swagger UI 和 ReDoc。这意味着每个端点都自动从与端点关联的元数据中记录下来。

此处列出了所有已注册的端点

替代文档

管理应用

Flask

Flask 有一个广泛使用的第三方管理包，称为 Flask-Admin，用于快速对您的模型执行 CRUD 操作。

FastAPI

截至目前，有两个流行的 FastAPI 扩展用于此：

1. FastAPI Admin - 功能性管理面板，提供用于对数据执行 CRUD 操作的用户界面。
2. SQLAlchemy Admin -FastAPI/Starlette 的管理面板，可与 SQLAlchemy 模型一起使用。

身份认证

Flask

虽然 Flask 没有原生解决方案，但可以使用多个第三方扩展。

FastAPI

FastAPI 通过 fastapi.security 包原生支持许多安全和身份验证工具。通过几行代码，您可以将基本的 HTTP 身份验证添加到您的应用程序中：

import secrets

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

app = FastAPI()

security = HTTPBasic()


def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, "stanleyjobson")
    correct_password = secrets.compare_digest(credentials.password, "swordfish")
    if not (correct_username and correct_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return credentials.username


@app.get("/whoami")
def who_ami_i(username: str = Depends(get_current_username)):
    return {"username": username}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

FastAPI 通过 OpenAPI 标准实现 OAuth2 和 OpenID Connect。

查看官方文档中的以下资源以获取更多信息：

1. Security Intro：https://fastapi.tiangolo.com/tutorial/security/
2. Advanced Security:https://fastapi.tiangolo.com/advanced/security/

其他资源

1. Web Authentication Methods Compared：https://testdriven.io/blog/web-authentication-methods/
2. Adding Social Authentication to Flask:https://testdriven.io/blog/flask-social-auth/
3. Session-based Auth with Flask for Single Page Apps:https://testdriven.io/blog/flask-spa-auth/
4. Securing FastAPI with JWT Token-based Authentication：https://testdriven.io/blog/fastapi-jwt-auth/

CORS

CORS（跨源资源共享）中间件检查请求是否来自允许的来源。如果是，则将请求传递给下一个中间件或视图函数。如果不是，它会拒绝请求，并将错误响应发送回调用者。

Flask

Flask 需要一个名为 Flask-CORS 的外部包来支持 CORS：

pip install flask-cors
1

基本实现：

from flask_cors import CORS

app = Flask(__name__)

CORS(app)
1
2
3
4
5

FastAPI

FastAPI 原生支持 CORS：

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = ["*"]

app.add_middleware(CORSMiddleware, allow_origins=origins)
1
2
3
4
5
6
7

测试

Flask

import pytest
from flask import Flask

app = Flask(__name__)

@app.route("/")
def home():
    return {"message": "OK"}

def test_hello():
    res = app.test_client().get("/")

    assert res.status_code == 200
    assert res.data == b'{"message":"OK"}\n'
1
2
3
4
5
6
7
8
9
10
11
12
13
14

FastAPI

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
async def home():
    return {"message": "OK"}

client = TestClient(app)

def test_home():
    res = client.get("/")

    assert res.status_code == 200
    assert res.json() == {"message": "OK"}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

FastAPI 提供了一个 TestClient。有了它，你可以直接用 FastAPI 运行 pytest。有关更多信息，请查看官方文档中的测试指南。

部署

生产服务器

Flask

Flask 默认运行开发 WSGI（Web 服务器网关接口）应用程序服务器。对于生产环境，您需要使用生产级 WSGI 应用服务器，例如 Gunicorn、uWSGI 或 mod_wsgi

安装 Gunicorn:

pip install gunicorn
1

启动服务：

# main.py
# app = Flask(__name__)

gunicorn main:app
1
2
3
4

FastAPI

由于 FastAPI 没有开发服务器，您将使用 Uvicorn（或 Daphne）进行开发和生产。

安装 Uvicorn：

pip install uvicorn
1

启动服务：

# main.py
# app = FastAPI()

uvicorn main:app
1
2
3
4

您可能希望使用 Gunicorn 来管理 Uvicorn，以便同时利用并发性（通过 Uvicorn）和并行性（通过 Gunicorn worker）：

# main.py
# app = FastAPI()

gunicorn -w 3 -k uvicorn.workers.UvicornWorker main:app
1
2
3
4

Docker

Flask

FROM python3.10-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install -r requirements.txt

COPY . .

EXPOSE 5000

CMD ["gunicorn", "main:app"]
1
2
3
4
5
6
7
8
9
10
11
12
13

这是 Flask 最简单的 Dockerfile 之一。要了解如何针对生产对其进行全面配置，请查看使用 Postgres、Gunicorn 和 Nginx 教程对 Flask 进行 Docker 化。

FastAPI

FROM python3.10-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app"]
1
2
3
4
5
6
7
8
9
10
11
12
13

同样，这是一个非常简单的配置。 FastAPI 作者提供了几个生产就绪的 Dockerfile。有关更多信息，请查看官方 FastAPI 文档以及 Dockerizing FastAPI with Postgres、Uvicorn 和 Traefik 教程。

总结

退一步讲，Django 和 Flask 是两个最流行的基于 Python 的网络框架（FastAPI 是第三大流行框架）。不过它们（Django 和 Flask）的理念非常不同。Flask 比 Django 的优势在于 Flask 是一个微框架。程序结构由程序员自己决定，不强制执行。开发者可以在他们认为合适的时候添加第三方扩展来改进他们的代码。也就是说，通常情况下，随着代码库的增长，需要一些几乎所有网络应用都需要的通用功能。这些功能与框架的紧密结合，使得终端开发者需要自己创建和维护的代码大大减少。

本文中的代码实例也表达了同样的意思。换句话说，FastAPI 包括许多必要的功能。它还遵循严格的标准，使你的代码可以生产并更容易维护。FastAPI 的文档也非常完善。

虽然 FastAPI 可能不像 Flask 那样久经考验，但越来越多的开发人员正在转向它来提供机器学习模型或开发 RESTful API。切换到 FastAPI 是一个不错的选择。

官方文档

1. FastAPI：https://fastapi.tiangolo.com/
2. Flask：https://flask.palletsprojects.com/en/2.0.x/

其他资源

• Porting Flask to FastAPI for ML Model Serving：https://www.pluralsight.com/tech-blog/porting-flask-to-fastapi-for-ml-model-serving/

• Why we switched from Flask to FastAPI for production machine learning：https://towardsdatascience.com/why-we-switched-from-flask-to-fastapi-for-production-machine-learning-765aab9b3679

• Awesome Flask:https://github.com/mjhea0/awesome-flask

• Awesome FastAPI:https://github.com/mjhea0/awesome-fastapi