Flask 与 FastAPI 的差异

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.get("/get")
def get_args():
    return jsonify({"args": request.args.to_dict()})


@app.post("/post")
def post_json():
    data = request.get_json(silent=True)
    if not isinstance(data, dict):
        return jsonify({"error": "请求体必须是 JSON 对象"}), 400
    return jsonify({"json": data})

1
2

flask --app flask_app run --port 5000
# 输出：Running on http://127.0.0.1:5000

1
2
3
4

curl "http://127.0.0.1:5000/get?page=1&size=2"
curl -X POST http://127.0.0.1:5000/post \
  -H "Content-Type: application/json" \
  -d '{"text": "hello"}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

from fastapi import FastAPI, Request
from pydantic import BaseModel, Field

app = FastAPI()


class TextBody(BaseModel):
    text: str = Field(min_length=1)  # 声明字段类型和最小长度


@app.get("/get")
def get_args(request: Request):
    return {"args": dict(request.query_params)}


@app.post("/post")
def post_json(body: TextBody):
    return {"json": body.model_dump()}  # 把校验后的模型转成字典

1
2

uvicorn fastapi_app:app --port 8000
# 输出：Uvicorn running on http://127.0.0.1:8000

1
2
3
4

curl "http://127.0.0.1:8000/get?page=1&size=2"
curl -X POST http://127.0.0.1:8000/post \
  -H "Content-Type: application/json" \
  -d '{"text": "hello"}'

1
2
3

curl -X POST http://127.0.0.1:8000/post \
  -H "Content-Type: application/json" \
  -d '{"text": ""}'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, Query
from pydantic import BaseModel, Field

app = FastAPI()
records = {1: {"record_id": 1, "text": "hello"}}  # 示例数据放在内存中


class Record(BaseModel):
    text: str = Field(min_length=1)


# Annotated[类型, 附加声明]：保留 int、str 等类型，同时标明参数来源和约束。
# Path(ge=1) 要求路径编号至少为 1；Query 限制 q 最长 20 个字符，缺省值为 None。
@app.get("/records/{record_id}")
def read_record(record_id: Annotated[int, Path(ge=1)], q: Annotated[str | None, Query(max_length=20)] = None):
    if record_id not in records:
        # HTTPException 立即结束请求，客户端收到 404 和 detail 中的错误说明。
        raise HTTPException(status_code=404, detail="Record not found")
    result = dict(records[record_id])
    if q:
        result["q"] = q
    return result


@app.put("/records/{record_id}")
def put_record(record_id: Annotated[int, Path(ge=1)], record: Record):
    # Record 告诉 FastAPI 从 JSON 请求体读取数据，校验后再传入函数。
    records[record_id] = {"record_id": record_id, **record.model_dump()}
    return records[record_id]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from fastapi import FastAPI, status
from pydantic import BaseModel, Field

app = FastAPI()
records = {1: {"record_id": 1, "text": "hello"}}


class Record(BaseModel):
    text: str = Field(min_length=1)


class RecordOut(BaseModel):
    record_id: int
    text: str


@app.post("/records", response_model=RecordOut, status_code=status.HTTP_201_CREATED)
def create_record(record: Record):
    record_id = max(records) + 1
    records[record_id] = {"record_id": record_id, "text": record.text}
    return {**records[record_id], "internal": "hidden"}  # internal 不在 RecordOut 中，会被过滤

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from flask import Flask, jsonify, request
from pydantic import BaseModel, Field, ValidationError

app = Flask(__name__)
records = {1: {"record_id": 1, "text": "hello"}}


class Record(BaseModel):
    text: str = Field(min_length=1)


@app.post("/records")
def create_record():
    try:
        record = Record.model_validate(request.get_json())  # 显式调用模型校验
    except ValidationError:
        return jsonify({"error": "参数不合法"}), 400
    record_id = max(records) + 1
    records[record_id] = {"record_id": record_id, "text": record.text}
    saved = {**records[record_id], "internal": "hidden"}
    return jsonify({"record_id": saved["record_id"], "text": saved["text"]}), 201

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from typing import Annotated

from fastapi import FastAPI, HTTPException, Path, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()
records = {1: "hello"}


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    # 只接管参数校验失败，不改变视图主动抛出的 HTTPException。
    return JSONResponse(status_code=400, content={"error": "参数不合法"})


@app.get("/records/{record_id}")
def read_record(record_id: Annotated[int, Path(ge=1)]):
    if record_id not in records:
        raise HTTPException(status_code=404, detail="Record not found")
    return {"record_id": record_id, "text": records[record_id]}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def get_message():
    # 普通 def 依赖由 FastAPI 在线程池中执行。
    return "hello"


@app.get("/message")
async def read_message(message: Annotated[str, Depends(get_message)]):
    # async def 路由在事件循环中执行，message 接收依赖的返回值。
    return {"message": message}

1
2

uvicorn depends_fastapi:app --port 8000
# 输出：Uvicorn running on http://127.0.0.1:8000

1

curl http://127.0.0.1:8000/message

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import time

from flask import Flask

app = Flask(__name__)


@app.get("/slow")
def slow():
    time.sleep(1)  # 当前线程在这一秒内不能处理其他请求。
    return {"ok": True}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import asyncio

from fastapi import FastAPI

app = FastAPI()


@app.get("/slow")
async def slow():
    await asyncio.sleep(1)  # 等待期间，事件循环可以运行其他请求。
    return {"ok": True}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

Flask，单 worker 单线程：
  req1 |---- 1s ----|
  req2              |---- 1s ----|
  req3                           |---- 1s ----|
  req4                                        |---- 1s ----|

FastAPI，单 worker：
  req1 |---- 1s ----|
  req2 |---- 1s ----|
  req3 |---- 1s ----|
  req4 |---- 1s ----|

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


class DB:
    def get(self, record_id: int):
        return {"record_id": record_id, "text": "example"}

    def close(self):
        print("close")


def get_db():
    db = DB()
    try:
        yield db  # 把对象交给视图，函数暂时停在这里。
    finally:
        db.close()  # 默认在响应发完后执行清理。


# Annotated 保留 DB 类型，同时用 Depends 指定对象的提供函数。
DBDep = Annotated[DB, Depends(get_db)]


@app.get("/db/{record_id}")
def read_record(record_id: int, db: DBDep):
    return db.get(record_id)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from flask import Flask, g, jsonify

app = Flask(__name__)


class DB:
    def get(self, record_id: int):
        return {"record_id": record_id, "text": "example"}

    def close(self):
        print("close")


def get_db():
    if "db" not in g:
        g.db = DB()  # 当前请求第一次使用时创建，之后复用。
    return g.db


@app.teardown_appcontext
def close_db(exc):
    # 没有用到数据库的请求，也会经过这个清理函数。
    db = g.pop("db", None)
    if db is not None:
        db.close()


@app.get("/db/<int:record_id>")
def read_record(record_id):
    return jsonify(get_db().get(record_id))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

from contextlib import asynccontextmanager

from fastapi import FastAPI


class Pool:
    def close(self):
        print("close")


@asynccontextmanager
async def manage_resources(app: FastAPI):
    # 启动阶段：把共享对象放在应用实例上；这里的 Pool 只是演示类。
    app.state.pool = Pool()
    yield  # 启动准备完成，应用开始处理请求。
    app.state.pool.close()  # 应用正常关闭时释放资源。


app = FastAPI(lifespan=manage_resources)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from fastapi import FastAPI

app = FastAPI()


@app.get("/records", tags=["records"])
def list_records():
    return [{"record_id": 1, "text": "hello"}]


@app.get("/records/{record_id}", tags=["records"])
def read_record(record_id: int):
    return {"record_id": record_id, "text": "hello"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from fastapi import APIRouter, FastAPI

app = FastAPI()
router = APIRouter(prefix="/records", tags=["records"])


@router.get("")  # 与前缀拼接后是 /records。
def list_records():
    return [{"record_id": 1, "text": "hello"}]


@router.get("/{record_id}")
def read_record(record_id: int):
    return {"record_id": record_id, "text": "hello"}


app.include_router(router)  # 集中注册；只有这一步之后，app 才会接收这组路径的请求。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from flask import Blueprint, Flask

app = Flask(__name__)
records = Blueprint("records", __name__, url_prefix="/records")


@records.get("")
def list_records():
    return [{"record_id": 1, "text": "hello"}]


@records.get("/<int:record_id>")
def read_record(record_id):
    return {"record_id": record_id, "text": "hello"}


app.register_blueprint(records)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(text: str):
    print("notify", text)


@app.post("/notify")
async def notify(background_tasks: BackgroundTasks):
    # 注册函数及参数；此处不会立即调用 write_notification。
    background_tasks.add_task(write_notification, "ready")
    return {"ok": True}

1
2
3

# 终端 1：停止此前的 8000 端口应用后启动。
uvicorn background_fastapi:app --port 8000
# 输出：Uvicorn running on http://127.0.0.1:8000

1
2

# 终端 2：发出通知请求。
curl -X POST http://127.0.0.1:8000/notify

1

Authorization: Bearer secret

1
2
3
4
5

HTTPBearer 读取请求头
    ↓
get_current_user 验证 token
    ↓
/me 接收用户信息

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

app = FastAPI()
security = HTTPBearer()


# Depends(security) 先调用 HTTPBearer，再把解析结果传入 credentials。
def get_current_user(credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)]):
    token = credentials.credentials  # 请求头中 Bearer 后面的内容。
    if token != "secret":
        raise HTTPException(status_code=401, detail="invalid token")
    return {"id": 7, "name": "Lin"}


@app.get("/me")
def me(user: Annotated[dict, Depends(get_current_user)]):
    # user 是 get_current_user 的返回值；验证失败时不会执行这个视图。
    return user

1
2
3

# 终端 1：停止此前占用 8000 端口的应用后启动。
uvicorn auth_fastapi:app --port 8000
# 输出：Uvicorn running on http://127.0.0.1:8000

1
2

# 终端 2：-H 指定请求头，把 token 发送给服务器。
curl -H "Authorization: Bearer secret" http://127.0.0.1:8000/me

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.get("/me")
def me():
    credentials = request.authorization
    # type 是认证方式，token 是 Bearer 后面的内容。
    if credentials is None or credentials.type != "bearer" or credentials.token != "secret":
        return jsonify({"detail": "invalid token"}), 401
    return jsonify({"id": 7, "name": "Lin"})

1
2
3

# 终端 1：停止此前占用 5000 端口的应用后启动。
flask --app auth_flask run --port 5000
# 输出：Running on http://127.0.0.1:5000

1
2

# 终端 2：发送相同的凭据。
curl -H "Authorization: Bearer secret" http://127.0.0.1:5000/me

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()  # 等待下一条消息，不占住事件循环。
            await websocket.send_text(f"echo: {text}")
    except WebSocketDisconnect:
        pass  # 客户端断开后，结束这条连接的处理。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import time

from flask import Flask, Response

app = Flask(__name__)


@app.get("/events")
def events():
    def generate():
        for index in range(3):
            time.sleep(1)
            yield f"data: {index}\n\n"  # 空行标记一条 SSE 事件结束。

    return Response(generate(), mimetype="text/event-stream")

1
2
3

# 终端 1：停止此前的 5000 端口应用后启动。
flask --app sse_flask run --port 5000
# 输出：Running on http://127.0.0.1:5000

1
2

# 终端 2：-N 关闭 curl 的输出缓冲，逐条显示到达的消息。
curl -N http://127.0.0.1:5000/events

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import asyncio

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def events():
    for index in range(3):
        await asyncio.sleep(1)
        yield f"data: {index}\n\n"


@app.get("/events")
async def sse():
    # StreamingResponse 逐次读取生成器，不等全部内容生成完再返回。
    return StreamingResponse(events(), media_type="text/event-stream")

1
2
3

# 终端 1：停止此前的 8000 端口应用后启动。
uvicorn sse_fastapi:app --port 8000
# 输出：Uvicorn running on http://127.0.0.1:8000

1
2

# 终端 2：请求流式响应。
curl -N http://127.0.0.1:8000/events

1
2

gunicorn -w 4 flask_app:app
# 输出：Listening at: http://127.0.0.1:8000

1
2

uvicorn fastapi_app:app --workers 4
# 输出：Uvicorn running on http://127.0.0.1:8000