跳至内容

大型应用 - 多个文件

如果您正在构建应用程序或 Web API,则很少会出现将所有内容都放在单个文件中的情况。

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

信息

如果您来自 Flask,这相当于 Flask 的蓝图。

文件结构示例

假设您有一个这样的文件结构

.
├── 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”类。

它支持所有相同的选项。

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

提示

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

提示

如果可能,建议使用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相同。

但我们希望更聪明一点,并简化代码。

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

  • 路径前缀/items
  • 标签:(只有一个标签:items)。
  • 额外的响应
  • 依赖项:它们都需要我们创建的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

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

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

提示

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

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

  • /items/
  • /items/{item_id}

…正如我们预期的那样。

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

提示

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

检查

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文件等。但我们没有这个。因此,这将在我们的示例中引发错误。🚨

但现在您知道它是如何工作的了,因此无论您的应用程序多么复杂,您都可以在自己的应用程序中使用相对导入。🤓

添加一些自定义的标签响应依赖项

我们没有将前缀/itemstags=["items"]添加到每个路径操作中,因为我们已将它们添加到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"}

提示

最后一个路径操作将具有标签组合:["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/目录)开始…
  • 查找子包routersapp/routers/目录)…
  • 并从中导入子模块itemsapp/routers/items.py中的文件)和usersapp/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中声明的每个路径操作创建一个路径操作

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

检查

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

这将花费几微秒,并且仅在启动时发生。

因此它不会影响性能。⚡

包含具有自定义前缀标签响应依赖项APIRouter

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

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

对于此示例,它将非常简单。但假设由于它与组织中的其他项目共享,我们无法修改它并直接将前缀依赖项标签等添加到APIRouter

app/internal/admin.py
from fastapi import APIRouter

router = APIRouter()


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

但我们仍然希望在包含APIRouter时设置自定义前缀,以便其所有路径操作都以/admin开头,我们希望使用我们已经为此项目设置的依赖项来保护它,并且我们希望包含标签响应

我们可以通过将这些参数传递给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

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

router.include_router(other_router)

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