FastAPI 大型应用:使用多个文件组织项目

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

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

一个文件结构示例

假设你的文件结构如下:

.
├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   └── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── internal
│       ├── __init__.py
│       └── admin.py
  • 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。

Python 包与子模块目录关系图。

带有注释的同一文件结构:

.
├── 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 类相同的方式创建一个「实例」:

Python 3.8+

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 类相同:

Python 3.8+

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

我们将在主 FastAPI 应用中包含该 APIRouter,但首先,让我们来看看依赖项和另一个 APIRouter。

依赖项

我们了解到我们将需要一些在应用程序的好几个地方所使用的依赖项。

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

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

Python 3.10+

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

其他使用 APIRouter 的模块

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

你具有以下路径操作:

  • /items/
  • /items/{item_id}

这和 app/routers/users.py 的结构完全相同。

但是我们想变得更聪明并简化一些代码。

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

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

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

Python 3.8+

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 列表。

导入依赖项

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

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

因此,我们通过 .. 对依赖项使用了相对导入:

Python 3.8+

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 文件中。

请记住我们的程序/文件结构是怎样的:

Python 包与子模块目录关系图。


两个点 ..,例如:

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

我们不打算在每个路径操作中添加前缀 /items 或 tags =["items"],因为我们将它们添加到了 APIRouter 中。

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

Python 3.8+

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

FastAPI 主体

现在,让我们来看看位于 app/main.py 的模块。

在这里你导入并使用 FastAPI 类。

这将是你的应用程序中将所有内容联结在一起的主文件。

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

导入 FastAPI

你可以像平常一样导入并创建一个 FastAPI 类。

我们甚至可以声明全局依赖项,它会和每个 APIRouter 的依赖项组合在一起:

Python 3.10+

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 的其他子模块:

Python 3.10+

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.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

避免名称冲突

我们将直接导入 items 子模块,而不是仅导入其 router 变量。

这是因为我们在 users 子模块中也有另一个名为 router 的变量。

如果我们一个接一个地导入,例如:

from .routers.items import router
from .routers.users import router

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

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

Python 3.10+

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 和 items 的 APIRouter

现在,让我们来包含来自 users 和 items 子模块的 router。

Python 3.10+

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 添加到主 FastAPI 应用程序中。

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

包含一个有自定义 prefix、tags、responses 和 dependencies 的 APIRouter

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

它包含一个带有一些由你的组织在多个项目之间共享的管理员路径操作的 APIRouter。

对于此示例,它将非常简单。但是假设由于它是与组织中的其他项目所共享的,因此我们无法对其进行修改,以及直接在 APIRouter 中添加 prefix、dependencies、tags 等:

Python 3.8+

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 保护它,并且我们希望它包含自定义的 tags 和 responses。

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

Python 3.10+

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 应用中。

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

Python 3.10+

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() 添加的所有其他路径操作一起正常运行。

在 pyproject.toml 中配置 entrypoint

因为你的 FastAPI app 对象位于 app/main.py 中,你可以在 pyproject.toml 中这样配置 entrypoint:

[tool.fastapi]
entrypoint = "app.main:app"

等价于像这样导入:

from app.main import app

这样 fastapi 命令就知道到哪里去找到你的应用了。

查看自动化的 API 文档

现在,运行你的应用:

$ uv run fastapi dev

<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 文档,包括了来自所有子模块的路径:

FastAPI 自动交互式 API 文档,展示各模块路由、前缀与标签。

多次使用不同的 prefix 包含同一个路由器

你也可以在同一路由器上使用不同的前缀来多次使用 .include_router()。

在有些场景这可能有用,例如以不同的前缀公开同一个的 API,比方说 /api/v1 和 /api/latest。

这是一个你可能并不真正需要的高级用法,但万一你有需要了就能够用上。

在另一个 APIRouter 中包含一个 APIRouter

与在 FastAPI 应用程序中包含 APIRouter 的方式相同,你也可以在另一个 APIRouter 中包含 APIRouter,通过:

router.include_router(other_router)

你可以在将 router 包含到 FastAPI 应用之前或之后执行此操作。FastAPI 仍然会在路由和 OpenAPI 中包含 other_router 中的路径操作。

同样适用于之后添加到这些路由器的路径操作。它们也会通过先前的包含可见。

来源:FastAPI 官方教程:Bigger Applications – Multiple Files,中文版本以官方中文文档为基础,逐节对照英文正文;代码保留英文原版。采用 MIT 许可证:

The MIT License (MIT)

Copyright (c) 2018 Sebastián Ramírez

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the “Software”), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.

© 版权声明
THE END
喜欢就支持一下吧
点赞0 分享
评论 抢沙发

请登录后发表评论

    暂无评论内容