跳转至

大型应用 - 多文件

如果你正在构建一个应用程序或 Web API,很少会将所有内容都放在单个文件中。

FastAPI 提供了一个便捷工具来构建你的应用程序,同时保持所有灵活性。

Info

如果你来自 Flask,这相当于 Flask 的 Blueprints。

示例文件结构

假设你有一个如下所示的文件结构:

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

Tip

有多个 __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" 是一个 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 应用程序/Web API 的一部分(它是同一个“Python 包”的一部分)。

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

导入 APIRouter

你导入它并创建一个“实例”,就像使用 FastAPI 类一样:

app/routers/users.py
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 类相同:

app/routers/users.py
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”类。

支持所有相同的选项。

所有相同的 parametersresponsesdependenciestags 等。

Tip

在这个例子中,变量名为 router,但你可以随意命名。

我们将把这个 APIRouter 包含在主 FastAPI 应用程序中,但首先,让我们检查依赖项和另一个 APIRouter

依赖项

我们看到我们将在应用程序的多个地方使用一些依赖项。

因此我们将它们放在它们自己的 dependencies 模块中(app/dependencies.py)。

我们现在将使用一个简单的依赖项来读取自定义的 X-Token 标头:

app/dependencies.py
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")
app/dependencies.py
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")

Tip

如果可能,请优先使用 Annotated 版本。

app/dependencies.py
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")

Tip

我们使用了一个虚构的标头来简化这个例子。

但在实际情况下,使用集成的 安全工具 会获得更好的结果。

另一个带有 APIRouter 的模块

假设你还有专门处理应用程序中“项目”的端点,位于 app/routers/items.py 模块中。

你有以下路径操作

  • /items/
  • /items/{item_id}

所有结构与 app/routers/users.py 相同。

但我们希望更智能一点,并简化代码。

我们知道此模块中的所有路径操作都具有相同的:

  • 路径 prefix/items
  • tags:(只有一个标签:items)。
  • 额外的 responses
  • dependencies:它们都需要我们创建的 X-Token 依赖项。

因此,我们可以将所有内容添加到 APIRouter 中,而不是添加到每个路径操作中。

app/routers/items.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"}

由于每个路径操作的路径必须以 / 开头,例如:

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

...前缀不能包含最后的 /

因此,在这种情况下,前缀是 /items

我们还可以添加一个 tags 列表和额外的 responses,这些将应用于此路由器中包含的所有路径操作

并且我们可以添加一个 dependencies 列表,这些将添加到路由器中的所有路径操作,并在对它们发出的每个请求中执行/解决。

Tip

请注意,与 路径操作装饰器中的依赖项 非常相似,不会将任何值传递给你的路径操作函数

最终结果是项目路径现在是:

  • /items/
  • /items/{item_id}

...正如我们预期的那样。

  • 它们将被标记为一个包含单个字符串 "items" 的标签列表。
    • 这些“标签”对于自动交互式文档系统(使用 OpenAPI)特别有用。
  • 它们都将包含预定义的 responses
  • 所有这些路径操作都将具有在它们之前评估/执行的 dependencies 列表。

Tip

APIRouter 中拥有 dependencies 可以用于,例如,要求对整个路径操作组进行身份验证。即使没有将依赖项单独添加到每个操作中。

Check

prefixtagsresponsesdependencies 参数(与许多其他情况一样)只是 FastAPI 的一个功能,帮助你避免代码重复。

导入依赖项

这段代码位于模块 app.routers.items 中,文件 app/routers/items.py

我们需要从模块 app.dependencies(文件 app/dependencies.py)中获取依赖函数。

因此我们使用带有 .. 的相对导入:

app/routers/items.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"}

相对导入的工作原理

Tip

如果你完全了解导入的工作原理,请继续阅读下一节。

单个点 .,例如:

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 等。但我们没有。因此,在我们的例子中会抛出错误。🚨

但现在你知道了它的工作原理,因此你可以在自己的应用程序中使用相对导入,无论它们多么复杂。🤓

添加一些自定义的 tagsresponsesdependencies

我们没有将前缀 /itemstags=["items"] 添加到每个路径操作中,因为我们已将它们添加到 APIRouter 中。

但我们仍然可以添加 更多 将应用于特定路径操作tags,以及一些特定于该路径操作的额外 responses

app/routers/items.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"}

Tip

最后一个路径操作将具有标签组合:["items", "custom"]

并且它还将在文档中同时包含两个响应,一个用于 404,一个用于 403

FastAPI

现在,让我们看看 app/main.py 模块。

这是你导入和使用 FastAPI 类的地方。

这将是你应用程序中连接所有内容的主文件。

由于你的大部分逻辑现在将存在于其自己的特定模块中,因此主文件将非常简单。

导入 FastAPI

你像通常一样导入并创建一个 FastAPI 类。

我们甚至可以声明 全局依赖项,这些将与每个 APIRouter 的依赖项组合:

app/main.py
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 的子模块:

app/main.py
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.pyapp/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 将有一个变量 routeritems.router)。这与我们在文件 app/routers/items.py 中创建的是同一个,它是一个 APIRouter 对象。

然后我们对模块 users 执行相同的操作。

我们也可以这样导入它们:

from app.routers import items, users

Info

第一个版本是“相对导入”:

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

usersrouter 将覆盖 itemsrouter,我们将无法同时使用它们。

因此,为了能够在同一文件中使用两者,我们直接导入子模块:

app/main.py
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!"}

包含 usersitemsAPIRouter

现在,让我们包含来自子模块 usersitemsrouter

app/main.py
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!"}

Info

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

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

使用 app.include_router(),我们可以将每个 APIRouter 添加到主 FastAPI 应用程序中。

它将包含该路由器中的所有路由作为其一部分。

技术细节

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

因此,在幕后,它实际上会像所有内容都是同一个应用程序一样工作。

Check

在包含路由器时,你无需担心性能问题。

这只需要微秒时间,并且只会在启动时发生。

因此不会影响性能。⚡

包含具有自定义 prefixtagsresponsesdependenciesAPIRouter

现在,假设你的组织给了你 app/internal/admin.py 文件。

它包含一个 APIRouter,其中包含一些你的组织在多个项目之间共享的管理员路径操作

对于这个例子,它将非常简单。但假设由于它与组织中的其他项目共享,我们无法直接修改它并添加 prefixdependenciestags 等到 APIRouter

app/internal/admin.py
from fastapi import APIRouter

router = APIRouter()


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

但我们仍然希望在包含 APIRouter 时设置自定义 prefix,以便其所有路径操作都以 /admin 开头,我们希望使用我们为此项目已有的 dependencies 来保护它,并且我们希望包含 tagsresponses

我们可以通过将那些参数传递给 app.include_router() 来声明所有这些,而无需修改原始的 APIRouter

app/main.py
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 应用程序中。

我们在这里这样做...只是为了表明我们可以 🤷:

app/main.py
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 模式和用户界面中。

由于我们不能只是隔离它们并“挂载”它们而独立于其余部分,因此路径操作是“克隆”(重新创建)的,而不是直接包含的。

检查自动 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 多次包含同一个路由器

你也可以使用 .include_router() 多次包含同一个路由器,但使用不同的前缀。

例如,这可能很有用,以便在不同的前缀下公开相同的 API,例如 /api/v1/api/latest

这是一个高级用法,你可能并不真正需要,但如果你需要的话,它就在那里。

在一个 APIRouter 中包含另一个

就像你可以将 APIRouter 包含在 FastAPI 应用程序中一样,你也可以使用以下方式将 APIRouter 包含在另一个 APIRouter 中:

router.include_router(other_router)

确保在将 router 包含在 FastAPI 应用程序中之前执行此操作,以便 other_router 中的路径操作也被包含在内。