跳到内容

大型应用 - 多文件

如果你正在构建一个应用程序或 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 类一样导入并创建一个“实例”

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

提示

在这个例子中,变量名为 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")

提示

如果可能,请优先使用 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")

提示

我们正在使用一个虚构的头部来简化这个例子。

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

另一个带有 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 列表,它将添加到路由器中的所有路径操作中,并为每个对它们发出的请求执行/解决。

提示

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

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

  • /items/
  • /items/{item_id}

...正如我们所愿。

  • 它们将被标记为包含单个字符串 "items" 的标签列表。
    • 这些“标签”对于自动交互式文档系统(使用 OpenAPI)特别有用。
  • 所有这些都将包含预定义的 responses
  • 所有这些路径操作都将在它们之前评估/执行 dependencies 列表。
    • 如果你还在特定的路径操作中声明了依赖项,它们也将被执行
    • 路由器依赖项首先执行,然后是装饰器中的 dependencies,然后是正常的参数依赖项。
    • 你还可以添加带 scopesSecurity 依赖项

提示

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

检查

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"}

相对导入的工作方式

提示

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

一个点 .,就像在

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"}

提示

这个最后的路径操作将具有标签组合:["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

信息

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

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!"}

信息

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

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

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

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

技术细节

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

所以,在幕后,它实际上会像所有东西都在同一个应用程序中一样工作。

检查

在包含路由器时,你不必担心性能。

这只会花费微秒,并且只会在启动时发生。

所以它不会影响性能。⚡

包含一个带有自定义 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() 添加的路径操作一起正常工作。

非常技术性的细节

注意:这是一个非常技术性的细节,你可能可以跳过


APIRouters 没有被“挂载”,它们没有与应用程序的其余部分隔离。

这是因为我们希望将它们的路径操作包含在 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 中包含另一个

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

router.include_router(other_router)

确保在将 router 包含到 FastAPI 应用程序之前完成此操作,以便也包含来自 other_router路径操作