大型应用 - 多个文件¶

开发应用或网络 API时，我们很少会把全部代码都放在一个文件里。

FastAPI 提供了方便、灵活的应用构建工具。

说明

如果您之前使用过 Flask，这种方式类似于 Flask 的 Blueprints。

文件架构示例¶

假设文件架构如下：

.
├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   └── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── internal
│       ├── __init__.py
│       └── admin.py

提示

__init__.py：每个文件夹或子文件夹都包含 __init__.py。

因此，可以把代码从一个文件导入到另一个文件。

例如，app/main.py 的这行代码：

from app.routers import items

app 文件夹包含了全部文件。其中有一个空文件 app/__init__.py，因此，它是 Python 包（Python 模块集合）：app
app/main.py 是 Python 包（包含 __init__.py 的文件夹）的模块：app.main
app/dependencies.py 和 app/main.py 一样也是模块：app.dependencies
app/routers/ 是包含 __init__.py 的 Python 子包：app.routers
app/routers/items.py 是 app/routers/ 的子模块：app.routers.items
app/routers/users.py 也是 app/routers/ 的子模块：app.routers.users
app/internal/ 是包含 __init__.py 的 Python 子包：app.internal
app/internal/admin.py 是 app/internal/ 的子模块：app.internal.admin

带注释的文件架构：

.
├── app                  # `app` 是 Python 包
│   ├── __init__.py      # 把 `app` 识别为 Python 包
│   ├── main.py          # `main` 模块，例如 import app.main
│   ├── dependencies.py  # `dependencies` 模块，例如 import app.dependencies
│   └── routers          # `routers` 是 Python 子包
│   │   ├── __init__.py  # 把 `routers` 识别为 Python 子包
│   │   ├── items.py     # `items` 子模块，例如 import app.routers.items
│   │   └── users.py     # `users` 子模块，例如 import app.routers.users
│   └── internal         # `internal`是 Python 子包
│       ├── __init__.py  # 把 `internal` 识别为 Python 子包
│       └── admin.py     # `admin` 子模块，例如 import app.internal.admin

`APIRouter`¶

假设专门处理用户的是 /app/routers/users.py 子模块。

该模块把用户相关的路径操作和其他代码分开，使项目文件井井有条。

但它仍属于 FastAPI 应用/网络 API（也是 Python 包的一部分）。

您可以使用 APIRouter 为该模块创建路径操作。

导入 `APIRouter`¶

与创建 FastAPI 类实例相同，导入 APIRouter 并创建实例：

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

使用 `APIRouter` 的路径操作¶

然后，用它声明路径操作。

使用方式与 FastAPI 类相同：

from fastapi import APIRouter

router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

APIRouter 就像是迷你 FastAPI 类。

它支持与 FastAPI 相同的选项。

包括 parameters、responses、dependencies、tags 等。

提示

本例中，路由变量命名为 router，但也可以随意命名。

接下来要向 FastAPI 主应用中添加 APIRouter，但我们首先看下依赖项和另一个 APIRouter。

依赖项¶

依赖项在 FastAPI 应用的多个地方使用。

因此，要把依赖项放在专属的 dependencies 模块（app/dependencies.py）里。

接下来，使用依赖项读取自定义 X-Token 请求头：

from fastapi import Header, HTTPException


async def get_token_header(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

提示

本例使用虚构的请求头进行了简化。

但实际开发时最好使用内置的安全工具。

其他使用 `APIRouter` 的模块¶

假设 app/routers/items.py 模块中还有一个专门处理 item 的端点。

路径操作如下：

/items/
/items/{item_id}

与 app/routers/users.py 的架构完全相同。

但此处还能进一步简化代码。

该模块的所有路径操作都有相同的：

路径 prefix：/items
tags：仅有一个 items 标签
附加的 responses
dependencies：共用的 X-Token 依赖项

不用在每个路径操作中添加这些内容，只在 APIRouter 里添加即可。

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)


fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

路径操作的路径必须以 / 开头，例如：

@router.get("/{item_id}")
async def read_item(item_id: str):
    ...

……前缀不能以 / 结尾。

本例中的前缀是 /items。

这个 router 里为所有路径操作添加了 tags 列表和附加的 responses。

还添加了用于处理接收请求的 dependencies 列表。

提示

注意，和路径操作装饰器中的依赖项类似，这里的依赖项也不会向路径操作函数传递任何值。

最终的 item 路径如下：

/items/
/items/{item_id}

……这就是我们想要的。

这些路径由仅含单字符串 "items" 的标签列表标记
- 这些标签用于（使用 OpenAPI 的） API 文档
所有路径操作都包括预定义的 responses
所有路径操作执行前都要先执行 dependencies 列表
- 在指定的路径操作中声明的依赖项也会被执行
- 首先执行的是 router 的依赖项，然后是装饰器的 dependencies，最后是普通的参数依赖项
- 还可以添加含 scopes 的 Security 依赖项

提示

APIRouter中的 dependencies 用于为一组路径操作进行身份验证，即便没有为每个路径操作单独添加依赖项。

检查

和其他很多功能一样，prefix、tags、responses、dependencies 等参数只是 FastAPI 用于减少代码重复的特性。

导入依赖项¶

这些代码在 app.routers.items 模块里，即 app/routers/items.py。

此时，需要从 app.dependencies 模块（ app/dependencies.py）里提取依赖函数。

通过 .. 相对导入依赖项：

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)


fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

相对导入如何工作¶

提示

如果您已经掌握了导入的工作原理，请跳过此段。

单点 .，例如：

from .dependencies import get_token_header

表示：

从模块（app/routers/items.py）所在的包（app/routers/）开始……
查找 dependencies 模块（不存在的 app/routers/dependencies.py 文件）……
然后，导入 get_token_header 函数

但该文件不存在，依赖项在 app/dependencies.py 里。

请记住本项目的文件架构是：

两个点 ..，例如：

from ..dependencies import get_token_header

表示：

从模块（app/routers/items.py）所在的包（app/routers/）开始……
跳转到父包（app/）……
在父包中查找 dependencies 模块（app/dependencies.py）……
然后，导入 get_token_header 函数

成功了！🎉

使用三个点 ...，例如：

from ...dependencies import get_token_header

表示：

从模块（app/routers/items.py）所在的包（app/routers/）开始……
跳转到父包（app/）……
再跳转到父包的父包（该父包不存在，app 已经是最顶层的包了 😱）……
在父包的父包中查找 dependencies 模块（app/ 上级文件夹中的 dependencies.py）……
然后，导入 get_token_header 函数

这时指向的是 app/ 之上的包，且要包含 __init __.py 。但其实并没有这个包，因此示例会报错。🚨

现在您已经了解了相对导入的工作原理，不管项目架构多复杂，都能在应用中使用相对导入。🤓

添加自定义 `tags`、`responses` 和 `dependencies`¶

因为我们已经为 APIRouter 添加了前缀 /items 和 tags =["items"]，因此，不必再在每个路径操作中单独添加。

但仍可以为指定的路径操作添加更多 tags 和 responses：

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import get_token_header

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)


fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}


@router.get("/")
async def read_items():
    return fake_items_db


@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "item_id": item_id}


@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
    if item_id != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

提示

最后，路径操作的标签组合是：["items","custom"]。

API 文档中也有两个响应，一个是 404，别一个是 403。

`FastAPI` 主模块¶

接下来是 app/main.py 模块。

在此，导入并使用 FastAPI 类。

这是把所有应用的内容联结在一起的主文件。

因为绝大多数逻辑都在专属的模块里，主文件就显得非常简单。

导入 `FastAPI`¶

导入 FastAPI 并创建类实例。

声明与 APIRouter 依赖项组合在一起使用的全局依赖项：

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

导入 `APIRouter`¶

导入包含 APIRouter 的子模块：

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

app/routers/users.py 和 app/routers/items.py 都是 Python 包（ app）的子模块，可使用单点 . 相对导入。

导入是怎么运作的¶

这行代码：

from .routers import items, users

表示：

从模块（app/main.py）所在的包（app/ ）开始……
查找 routers 子包（app/routers/）……
从子包导入子模块 items（app/routers/items.py）与 users（app/routers/users.py）……

items 模块包含 router 变量（items.router），这个变量是在 app/routers/items.py 中创建的，是 APIRouter 对象。

然后为 users 模块执行相同操作。

以如下方式导入：

from app.routers import items, users

说明

第一个版本是相对导入：

from .routers import items, users

第二个版本是绝对导入：

from app.routers import items, users

Python 包和模块详见 Python 官档 - 模块。

避免名称冲突¶

要直接导入 items 子模块，不能只导入 router 变量。

因为 users 子模块也有 router 变量。

如果逐个导入，例如：

from .routers.items import router
from .routers.users import router

users 的 router 会覆盖 items 的 router，就无法同时使用了。

为了在同一个文件中使用两个 router，需要直接导入子模块：

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

添加 `users` 和 `items` 的 `APIRouter`¶

接下来，添加 users 和 items 子模块的 router。

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

说明

users.router 包含 app/routers/users.py 中的 APIRouter。

items.router 包含 app/routers/items.py 中的 APIRouter。

app.include_router() 把 APIRouter 添加到 FastAPI 主应用。

它把 router 中的所有路由都作为主路由的组成部分。

技术细节

实际上，它在内部为 APIRouter 里声明的每个路径操作创建一个路径操作。

所以，在后台，所有部件就像是在同一个应用里运行。

检查

不用担心包含路由器操作的性能，

这项操作只需要几微秒，而且只在应用启动时运行。

不会影响性能。⚡

添加自定义 `prefix`、`tags`、`responses` 和 `dependencies` 的 `APIRouter`¶

假设公司提供了 app/internal/admin.py。

这个文件包含了公司里多个项目共享的管理员路径操作的 APIRouter。

这个例子非常简单，但假设它要与其他项目共享，不能修改，也不能直接在它的 APIRouter 中添加 prefix、dependencies、tags 等内容：

from fastapi import APIRouter

router = APIRouter()


@router.post("/")
async def update_admin():
    return {"message": "Admin getting schwifty"}

但我们依然希望在添加 APIRouter 时设置自定义 prefix，让管理员项下的所有路径操作都以 /admin 开头，同时还要使用已有的 dependencies 保护路径操作，并添加自定义 tags 和 responses。

此时，只需把参数传递给 app.include_router()，不用修改原始 APIRouter：

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

这样，原始 APIRouter 保持不变，但仍能与其他项目共享相同的 app/internal/admin.py。

最后，admin 模块的每个路径操作都包含：

/admin 前缀
admin 标签
get_token_header 依赖项
418 响应 🍵

但这只影响本应用中的 APIRouter，不影响其他应用。

也就是说，其他项目可以为这个 APIRouter 使用其他身份验证的方法。

添加路径操作¶

直接把路径操作添加到 FastAPI 应用。

以下代码只是为了证明 FastAPI 能做到这一点🤷：

from fastapi import Depends, FastAPI

from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users

app = FastAPI(dependencies=[Depends(get_query_token)])


app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
)


@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

这个路径操作与 app.include_router() 添加的路径操作能够一起正常运行。

特别的技术细节

注意：这是非常技术性的细节，可以直接跳过。

APIRouter 没有被挂载，也没有与应用的其他部分隔离。

这是因为我们想在 OpenAPI 概图和用户界面里包含它们的路径操作。

因为不能隔离，也不能把它们与其余部分独立开来，并挂载，因此这里是克隆（重新创建）了路径操作，而不是直接包含。

查看文档¶

现在，使用 app.main 模块和 app 变量运行 uvicorn：

$ uvicorn app.main:app --reload

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

打开 API 文档： http://127.0.0.1:8000/docs。

就能看到 API 文档包含了所有子模块的路径，并使用了正确的路径（和前缀）及标签：

使用不同 `prefix` 多次包含同一个路由器¶

多次使用 .include_router()，并为同一个 router 使用不同前缀。

有些场景可能用得上这个功能，例如，以不同前缀发布同一个 API，比如 /api/v1 和 /api/latest。

这种高级用法一般用不上，但万一有需要时就可以使用。

`APIRouter` 包含 `APIRouter`¶

与在 FastAPI 应用中添加 APIRouter 的方式一样，可在 APIRouter 中包含 APIRouter，代码如下：

router.include_router(other_router)

注意，一定要在把 router 添加到 FastAPI 应用前执行此操作，这样才能添加 other_router 中的路径操作。

大型应用 - 多个文件¶

文件架构示例¶

APIRouter¶

导入 APIRouter¶

使用 APIRouter 的路径操作¶

依赖项¶

其他使用 APIRouter 的模块¶

导入依赖项¶

相对导入如何工作¶

添加自定义 tags、responses 和 dependencies¶

FastAPI 主模块¶

导入 FastAPI¶

导入 APIRouter¶

导入是怎么运作的¶

避免名称冲突¶

添加 users 和 items 的 APIRouter¶

添加自定义 prefix、tags、responses 和 dependencies 的 APIRouter¶

添加路径操作¶

查看文档¶

使用不同 prefix 多次包含同一个路由器¶

APIRouter 包含 APIRouter¶

`APIRouter`¶

导入 `APIRouter`¶

使用 `APIRouter` 的路径操作¶

其他使用 `APIRouter` 的模块¶

添加自定义 `tags`、`responses` 和 `dependencies`¶

`FastAPI` 主模块¶

导入 `FastAPI`¶

导入 `APIRouter`¶

添加 `users` 和 `items` 的 `APIRouter`¶

添加自定义 `prefix`、`tags`、`responses` 和 `dependencies` 的 `APIRouter`¶

使用不同 `prefix` 多次包含同一个路由器¶

`APIRouter` 包含 `APIRouter`¶