更大的应用 - 多个文件¶
如果你正在构建一个应用程序或 Web 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 文件：每个目录或子目录中都有一个。
这就是允许从一个文件导入代码到另一个文件的原因。
例如，在 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.users`。
- 还有一个子目录 `app/internal/`，其中包含另一个 `__init__.py` 文件，所以它是一个“Python 子包”：`app.internal`。
- 而 `app/internal/admin.py` 文件是另一个子模块：`app.internal.admin`。
带有注释的相同文件结构
.
├── app # "app" is a Python package
│ ├── __init__.py # this file makes "app" a "Python package"
│ ├── main.py # "main" module, e.g. import app.main
│ ├── dependencies.py # "dependencies" module, e.g. import app.dependencies
│ └── routers # "routers" is a "Python subpackage"
│ │ ├── __init__.py # makes "routers" a "Python subpackage"
│ │ ├── items.py # "items" submodule, e.g. import app.routers.items
│ │ └── users.py # "users" submodule, e.g. import app.routers.users
│ └── internal # "internal" is a "Python subpackage"
│ ├── __init__.py # makes "internal" a "Python subpackage"
│ └── admin.py # "admin" submodule, e.g. import app.internal.admin
APIRouter¶
假设专门处理用户的模块是 `app/routers/users.py`。
你希望将与用户相关的*路径操作*与代码的其余部分分开，以保持代码的组织性。
但它仍然是同一个 FastAPI 应用程序/Web API 的一部分（它是同一个“Python 包”的一部分）。
你可以使用 `APIRouter` 为该模块创建*路径操作*。
导入 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` 类相同
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`”类。
支持所有相同的选项。
所有相同的 `parameters`、`responses`、`dependencies`、`tags` 等。
提示
在这个例子中，变量名为 `router`，但你可以随意命名。
我们将把这个 `APIRouter` 包含在主 `FastAPI` 应用程序中，但首先，让我们检查依赖项和另一个 `APIRouter`。
依赖项¶
我们看到应用程序的多个地方将需要一些依赖项。
因此我们将它们放在自己的 `dependencies` 模块 (`app/dependencies.py`) 中。
我们现在将使用一个简单的依赖项来读取自定义的 `X-Token` 头
from typing import Annotated
from fastapi import Header, HTTPException
async def get_token_header(x_token: Annotated[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")
from fastapi import Header, HTTPException
from typing_extensions import Annotated
async def get_token_header(x_token: Annotated[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")
提示
如果可能，请优先使用 Annotated 版本。
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`，其中包含专门处理应用程序中“项目”的端点。
你有针对以下内容的*路径操作*：
/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):
...
...`prefix` 不得包含末尾的 `/`。
因此，在这种情况下，`prefix` 是 `/items`。
我们还可以添加一个 `tags` 列表和额外的 `responses`，它们将应用于此路由器中包含的所有*路径操作*。
并且我们可以添加一个 `dependencies` 列表，它将添加到路由器中的所有*路径操作*，并在对它们发出的每个请求中执行/解析。
提示
请注意，与*路径操作装饰器*中的依赖项类似，不会有任何值传递给你的*路径操作函数*。
最终结果是，项目路径现在是
/items//items/{item_id}
...正如我们所预期的那样。
- 它们将被标记为一个包含单个字符串 `“items”` 的标签列表。
- 这些“标签”对于自动交互式文档系统（使用 OpenAPI）特别有用。
- 所有这些都将包含预定义的 `responses`。
- 所有这些*路径操作*都将在它们之前评估/执行 `dependencies` 列表。
- 如果你还在特定的*路径操作*中声明了依赖项，它们也将被执行。
- 路由器依赖项首先执行，然后是装饰器中的 `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¶
我们没有将 `prefix` `/items` 或 `tags=["items"]` 添加到每个*路径操作*中，因为我们已将它们添加到了 `APIRouter`。
但我们仍然可以添加*更多*的 `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"]` 的标签组合。
并且它在文档中也会包含两个响应，一个用于 `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`s 的子模块
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`，我们将无法同时使用它们。
因此，为了能够在同一个文件中同时使用它们，我们直接导入子模块
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!"}
包含 APIRouters for users and items¶
现在，让我们包含子模块 `users` 和 `items` 中的 `router`s
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` 应用程序中。
它将把该路由器的所有路由作为其一部分包含进来。
技术细节
它实际上会在内部为在 `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` 应用程序中。
我们在这里这样做……只是为了表明我们可以🤷
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`s 没有被“挂载”，它们没有与应用程序的其余部分隔离。
这是因为我们希望将它们的*路径操作*包含在 OpenAPI 规范和用户界面中。
由于我们无法将它们隔离并独立于其余部分“挂载”，因此*路径操作*是“克隆”（重新创建）的，而不是直接包含的。
检查自动 API 文档¶
现在，运行你的应用程序
$ fastapi dev app/main.py
<span style="color: green;">INFO</span>: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
并在 http://127.0.0.1:8000/docs 打开文档。
你将看到自动 API 文档，其中包括来自所有子模块的路径，使用正确的路径（和前缀）和正确的标签
使用不同的 prefix 多次包含同一个路由器¶
你还可以使用 `app.include_router()` 多次，对*同一个*路由器使用不同的前缀。
这可能很有用，例如，在不同的前缀下暴露相同的 API，例如 `/api/v1` 和 `/api/latest`。
这是一种高级用法，你可能并不真正需要，但它在那里以防万一。
在另一个 APIRouter 中包含一个 APIRouter¶
就像你可以在 `FastAPI` 应用程序中包含一个 `APIRouter` 一样，你也可以在另一个 `APIRouter` 中使用以下方法包含一个 `APIRouter`
router.include_router(other_router)
确保在将 `router` 包含到 `FastAPI` 应用程序之前完成此操作，以便 `other_router` 中的*路径操作*也包含在内。