我这几天上手体验 FastAPI,感受到这个框架易用和方便。之前也使用过 Python 中的 Django 和 Flask 作为项目的框架。Django 说实话上手也方便,但是学习起来有点重量级框架的感觉,FastAPI 带给我的直观体验还是很轻便的,本文就会着重介绍 FastAPI 和 Flask 的区别。
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
- 开始
- 安装
- "Hello World" 应用
- 配置
- 路由, 模板和视图
- HTTP 方法
- URL 参数
- 查询参数
- 模板
- 静态文件
- 异步任务
- 依赖注入
- 数据校验
- 序列化和反序列化
- 中间件
- 模块化
- 其他特点
- 自动文档
- 管理应用
- 身份认证
- CORS
- 测试
- 部署
- 生产服务器
- Docker
- 总结
- 官方文档
- 其他资源
实战项目系列
- 实战:基于Python+Flask+MySQL的豆瓣电影可视化系统
- 实战:搭建基于 Python+Flask+MySQL 的学生培养计划管理系统(附源码)
- 实战:一款基于 Python+flask 的态势感知系统(附完整源码)
- 实战:基于 Python 的 Flask 框架开发的在线电影网站系统(附完整源码)
- 实战:基于 Echarts + Python Flask 动态实时大屏轻松可以实现
- 实战:基于 Python+Django 构建智能互动拍照系统
- 实战:基于 Python+Flask+SQLite 的网易云音乐评论情感分析系统
- 实战:基于 Python 和Surprise库,新手轻松搭建推荐系统
- 实战:基于 Python+Django+MySQL 数据库的租房数据可视化系统
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
FastAPI
pip install fastapi uvicorn
# or
poetry add fastapi uvicorn
pipenv install fastapi uvicorn
conda install fastapi uvicorn -c conda-forge
与 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()
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")
像 reload=True
这样的参数可以被传递到 uvicorn.run()
中,以实现开发时的热重载。
或者,您可以直接从终端启动服务器:
uvicorn run fastapi_code:app
热加载模式:
uvicorn run fastapi_code:app --reload
配置
Flask 和 FastAPI 都提供了许多选项来处理不同环境的不同配置。两者都支持以下模式:
-
环境变量
-
配置文件
-
实例文件夹
-
类和继承
有关更多信息,请参阅其各自的文档:
-
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()
现在,在你运行服务器之前,设置适当的环境变量:
export MESSAGE="hello, world"
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")
同样,在运行服务器之前,设置适当的环境变量:
export MESSAGE="hello, world"
路由, 模板和视图
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"}
FastAPI
@app.get("/")
def home():
return {"Hello": "GET"}
@app.post("/")
def home_post():
return {"Hello": "POST"}
FastAPI 为每个方法提供单独的装饰器:
@app.get("/")
@app.post("/")
@app.delete("/")
@app.patch("/")
URL 参数
通过 URL(如 /employee/1
)传递信息以管理状态:
Flask
@app.route("/employee/<int:id>")
def home():
return {"id": id}
FastAPI
@app.get("/employee/{id}")
def home(id: int):
return {"id": id}
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}
FastAPI
@app.get("/employee")
def home(department: str):
return {"department": department}
模板
Flask
from flask import render_template
@app.route("/")
def home():
return render_template("index.html")
默认情况下,Flask会在 "templates "文件夹中寻找模板。
FastAPI
你需要安装 Jinja:
pip install jinja2
实现:
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})
对于 FastAPI,你需要明确地定义 "模板 "文件夹。然后对于每个响应,需要提供请求上下文。
静态文件
Flask
默认情况下,Flask 从“static”文件夹中提供静态文件。
FastAPI
在 FastAPI 中,需要为静态文件挂载一个文件夹:
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
异步任务
Flask
从 Flask 2.0 开始,您可以使用 async/await
创建异步路由处理程序:
@app.route("/")
async def home():
result = await some_async_task()
return result
有关 Flask 中异步视图的更多信息,请查看 Flask 2.0 中的异步一文。
Flask 中的异步也可以通过使用线程(并发)或多处理(并行)或 Celery 或 RQ 等工具来实现:
-
Asynchronous Tasks with Flask and Celery:https://testdriven.io/blog/flask-and-celery/
-
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
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"}
在这里,响应将被即时发送,而不会让用户等待文件处理完成。
当你需要进行繁重的后台计算时,或者你需要一个任务队列来管理任务(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)
因此,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"}
在这里,我们接受一个模型 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"}]}
注意到这个请求。我们把密码 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)
对于复杂的对象,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"}
在这里,我们添加了一个包含三个输入的 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"}
中间件
中间件被用来在每个请求被视图功能处理之前应用逻辑。
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)
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
@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
模块化
随着应用程序的发展,在某些时候你会想把类似的视图、模板、静态文件和模型组合在一起,以帮助把应用程序分解成更小的组件。
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)
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)
其他特点
自动文档
Flask
Flask 不会自动创建开箱即用的 API 文档。然而,有几个扩展可以处理这个问题,比如 flask-swagger 和 Flask RESTX,但它们需要额外的设置。
FastAPI
默认情况下,FastAPI 支持 OpenAPI 以及 Swagger UI 和 ReDoc。这意味着每个端点都自动从与端点关联的元数据中记录下来。
所有注册的端点都列在这里
此处列出了所有已注册的端点
替代文档
管理应用
Flask
Flask 有一个广泛使用的第三方管理包,称为 Flask-Admin,用于快速对您的模型执行 CRUD 操作。
FastAPI
截至目前,有两个流行的 FastAPI 扩展用于此:
-
FastAPI Admin - 功能性管理面板,提供用于对数据执行 CRUD 操作的用户界面。
-
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}
FastAPI 通过 OpenAPI 标准实现 OAuth2 和 OpenID Connect。
查看官方文档中的以下资源以获取更多信息:
-
Security Intro:https://fastapi.tiangolo.com/tutorial/security/
-
Advanced Security:https://fastapi.tiangolo.com/advanced/security/
其他资源
-
Web Authentication Methods Compared:https://testdriven.io/blog/web-authentication-methods/
-
Adding Social Authentication to Flask:https://testdriven.io/blog/flask-social-auth/
-
Session-based Auth with Flask for Single Page Apps:https://testdriven.io/blog/flask-spa-auth/
-
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
基本实现:
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
FastAPI
FastAPI 原生支持 CORS:
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = ["*"]
app.add_middleware(CORSMiddleware, allow_origins=origins)
测试
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'
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"}
FastAPI 提供了一个 TestClient。有了它,你可以直接用 FastAPI 运行 pytest。有关更多信息,请查看官方文档中的测试指南。
部署
生产服务器
Flask
Flask 默认运行开发 WSGI(Web 服务器网关接口)应用程序服务器。对于生产环境,您需要使用生产级 WSGI 应用服务器,例如 Gunicorn、uWSGI 或 mod_wsgi
安装 Gunicorn:
pip install gunicorn
启动服务:
# main.py
# app = Flask(__name__)
gunicorn main:app
FastAPI
由于 FastAPI 没有开发服务器,您将使用 Uvicorn(或 Daphne)进行开发和生产。
安装 Uvicorn:
pip install uvicorn
启动服务:
python
# main.py
# app = FastAPI()
uvicorn main:app
您可能希望使用 Gunicorn 来管理 Uvicorn,以便同时利用并发性(通过 Uvicorn)和并行性(通过 Gunicorn worker):
# main.py
# app = FastAPI()
gunicorn -w 3 -k uvicorn.workers.UvicornWorker main:app
Docker
Flask
FROM python3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["gunicorn", "main:app"]
这是 Flask 最简单的 Dockerfile 之一。要了解如何针对生产对其进行全面配置,请查看使用 Postgres、Gunicorn 和 Nginx 教程对 Flask 进行 Docker 化。
FastAPI
sql
FROM python3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app"]
同样,这是一个非常简单的配置。 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 是一个不错的选择。
官方文档
-
FastAPI:https://fastapi.tiangolo.com/
-
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