跳到内容

大型应用 - 多个文件

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

所有相同的选项都受支持。

所有相同的 parameters, responses, dependencies, tags 等等。

提示

在此示例中,变量名为 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")
🤓 其他版本和变体

提示

如果可能,请优先使用 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 的模块

假设您还拥有应用程序中处理“items”的端点,位于 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,它们将应用于包含在此 router 中的所有路径操作

我们还可以添加一组 dependencies,它们将被添加到 router 中的所有路径操作,并且将在每次发出请求时进行评估/解决。

提示

请注意,与路径操作装饰器中的依赖项中一样,您的路径操作函数不会收到任何值。

最终结果是 item 路径现在是

  • /items/
  • /items/{item_id}

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

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

提示

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

检查

prefix, tags, responsesdependencies 参数(与其他许多情况一样)是 FastAPI 的一项功能,旨在帮助您避免代码重复。

导入依赖项

此代码位于 app.dependencies 模块,文件为 app/dependencies.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 文件中。

请记住我们的 app/文件结构是怎样的


两个点 ..,例如

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, responsesdependencies

我们没有为每个路径操作添加 /items 前缀或 tags=["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!"}
🤓 其他版本和变体

提示

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

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!"}
🤓 其他版本和变体

提示

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

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 将覆盖来自 items 的,我们将无法同时使用它们。

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

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!"}
🤓 其他版本和变体

提示

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

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

现在,让我们包含 usersitems 子模块中的 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!"}
🤓 其他版本和变体

提示

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

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 应用程序。

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

技术细节

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

因此,在后台,它的工作方式实际上与所有内容都位于同一个应用程序中一样。

检查

您无需担心包含 router 的性能。

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

因此不会影响性能。⚡

包含一个带有自定义 prefix, tags, responsesdependenciesAPIRouter

现在,让我们想象您的组织给了您 app/internal/admin.py 文件。

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

在此示例中,它将非常简单。但让我们假设,因为它与组织中的其他项目共享,所以我们不能修改它并直接向 APIRouter 添加 prefix, dependencies, tags 等。

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!"}
🤓 其他版本和变体

提示

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

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!"}
🤓 其他版本和变体

提示

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

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 多次包含同一个 router

您也可以使用 .include_router() 多次包含同一个 router,并使用不同的前缀。

例如,这可能很有用,可以将同一个 API 暴露在不同的前缀下,例如 /api/v1/api/latest

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

在一个 APIRouter 中包含另一个 APIRouter

就像您可以在 FastAPI 应用程序中包含 APIRouter 一样,您可以使用

router.include_router(other_router)

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