Skip to content

大型应用 - 多文件

如果你正在构建一个应用程序或一个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" 类。

所有相同的选项都支持。

所有相同的 参数响应依赖项标签 等。

Tip

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

我们将在主 FastAPI 应用程序中包含这个 APIRouter,但首先,让我们检查一下依赖项和另一个 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" 端点。

你为以下路径操作定义了 path operations

  • /items/
  • /items/{item_id}

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

但我们希望更聪明一些,稍微简化代码。

我们知道这个模块中的所有 path operations 都有相同的:

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

因此,我们不必将所有这些添加到每个 path operation 中,而是可以将它们添加到 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"}

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

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

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

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

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

我们还可以添加一个 dependencies 列表,这些依赖项将添加到路由器中的所有 path operations 中,并在对它们进行请求时执行/解析。

Tip

请注意,与 path operation decorators 中的依赖项 类似,不会将任何值传递给你的 path operation 函数

最终结果是,item 路径现在是:

  • /items/
  • /items/{item_id}

...正如我们所预期的。

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

Tip

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

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 是顶级目录 😱)...

这会导致错误,因为 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"]

并且它还将在文档中同时显示 404403 的响应。

主要的 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 文件是 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

users 中的 router 会覆盖 items 中的 router,我们将无法同时使用它们。

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

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 等。

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 中包含一个 APIRouter,使用:

router.include_router(other_router)

确保在将 router 包含在 FastAPI 应用中之前这样做,以便 other_router路径操作 也被包含。