大型应用 - 多文件¶
如果你正在构建一个应用程序或 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 类一样:
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 类相同:
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 等。
Tip
在这个例子中,变量名为 router,但你可以随意命名。
我们将把这个 APIRouter 包含在主 FastAPI 应用程序中,但首先,让我们检查依赖项和另一个 APIRouter。
依赖项¶
我们看到我们将在应用程序的多个地方使用一些依赖项。
因此我们将它们放在它们自己的 dependencies 模块中(app/dependencies.py)。
我们现在将使用一个简单的依赖项来读取自定义的 X-Token 标头:
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")
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 版本。
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 中,而不是添加到每个路径操作中。
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 列表,这些将添加到路由器中的所有路径操作,并在对它们发出的每个请求中执行/解决。
Tip
请注意,与 路径操作装饰器中的依赖项 非常相似,不会将任何值传递给你的路径操作函数。
最终结果是项目路径现在是:
/items//items/{item_id}
...正如我们预期的那样。
- 它们将被标记为一个包含单个字符串
"items"的标签列表。- 这些“标签”对于自动交互式文档系统(使用 OpenAPI)特别有用。
- 它们都将包含预定义的
responses。 - 所有这些路径操作都将具有在它们之前评估/执行的
dependencies列表。- 如果你还在特定的路径操作中声明了依赖项,它们也将被执行。
- 路由器依赖项首先执行,然后是 装饰器中的
dependencies,然后是普通参数依赖项。 - 你还可以添加 带有
scopes的Security依赖项。
Tip
在 APIRouter 中拥有 dependencies 可以用于,例如,要求对整个路径操作组进行身份验证。即使没有将依赖项单独添加到每个操作中。
Check
prefix、tags、responses 和 dependencies 参数(与许多其他情况一样)只是 FastAPI 的一个功能,帮助你避免代码重复。
导入依赖项¶
这段代码位于模块 app.routers.items 中,文件 app/routers/items.py。
我们需要从模块 app.dependencies(文件 app/dependencies.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是顶级 😱)... - 并在那里找到模块
dependencies(文件app/dependencies.py)... - 并从其中导入函数
get_token_header。
这将引用 app/ 上方的某个包,该包有自己的文件 __init__.py 等。但我们没有。因此,在我们的例子中会抛出错误。🚨
但现在你知道了它的工作原理,因此你可以在自己的应用程序中使用相对导入,无论它们多么复杂。🤓
添加一些自定义的 tags、responses 和 dependencies¶
我们没有将前缀 /items 和 tags=["items"] 添加到每个路径操作中,因为我们已将它们添加到 APIRouter 中。
但我们仍然可以添加 更多 将应用于特定路径操作的 tags,以及一些特定于该路径操作的额外 responses:
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"]。
并且它还将在文档中同时包含两个响应,一个用于 404,一个用于 403。
主 FastAPI¶
现在,让我们看看 app/main.py 模块。
这是你导入和使用 FastAPI 类的地方。
这将是你应用程序中连接所有内容的主文件。
由于你的大部分逻辑现在将存在于其自己的特定模块中,因此主文件将非常简单。
导入 FastAPI¶
你像通常一样导入并创建一个 FastAPI 类。
我们甚至可以声明 全局依赖项,这些将与每个 APIRouter 的依赖项组合:
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 的子模块:
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
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,我们将无法同时使用它们。
因此,为了能够在同一文件中使用两者,我们直接导入子模块:
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:
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
在包含路由器时,你无需担心性能问题。
这只需要微秒时间,并且只会在启动时发生。
因此不会影响性能。⚡
包含具有自定义 prefix、tags、responses 和 dependencies 的 APIRouter¶
现在,假设你的组织给了你 app/internal/admin.py 文件。
它包含一个 APIRouter,其中包含一些你的组织在多个项目之间共享的管理员路径操作。
对于这个例子,它将非常简单。但假设由于它与组织中的其他项目共享,我们无法直接修改它并添加 prefix、dependencies、tags 等到 APIRouter:
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:
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 应用程序中。
我们在这里这样做...只是为了表明我们可以 🤷:
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 中:
router.include_router(other_router)
确保在将 router 包含在 FastAPI 应用程序中之前执行此操作,以便 other_router 中的路径操作也被包含在内。