大型应用 - 多文件¶
如果你正在构建一个应用程序或一个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
" 类。
所有相同的选项都支持。
所有相同的 参数
、响应
、依赖项
、标签
等。
Tip
在这个例子中,变量被称为 router
,但你可以随意命名它。
我们将在主 FastAPI
应用程序中包含这个 APIRouter
,但首先,让我们检查一下依赖项和另一个 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")
Tip
我们使用了一个虚构的头来简化这个例子。 但在实际情况下,使用集成的安全工具会获得更好的结果。
另一个带有 APIRouter
的模块¶
假设你还有一个模块,位于 app/routers/items.py
,专门处理应用程序中的 "items" 端点。
你为以下路径操作定义了 path operations:
/items/
/items/{item_id}
它的结构与 app/routers/users.py
完全相同。
但我们希望更聪明一些,稍微简化代码。
我们知道这个模块中的所有 path operations 都有相同的:
- 路径
prefix
:/items
。 tags
:(只有一个标签:items
)。- 额外的
responses
。 dependencies
:它们都需要我们创建的X-Token
依赖项。
因此,我们不必将所有这些添加到每个 path operation 中,而是可以将它们添加到 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"}
由于每个 path operation 的路径必须以 /
开头,例如:
@router.get("/{item_id}")
async def read_item(item_id: str):
...
...前缀不能包含最后的 /
。
因此,在这种情况下,前缀是 /items
。
我们还可以添加一个 tags
列表和额外的 responses
,这些将应用于该路由器中包含的所有 path operations。
我们还可以添加一个 dependencies
列表,这些依赖项将添加到路由器中的所有 path operations 中,并在对它们进行请求时执行/解析。
Tip
请注意,与 path operation decorators 中的依赖项 类似,不会将任何值传递给你的 path operation 函数。
最终结果是,item 路径现在是:
/items/
/items/{item_id}
...正如我们所预期的。
- 它们将被标记为一个包含单个字符串
"items"
的标签列表。- 这些 "tags" 对于自动交互式文档系统(使用 OpenAPI)特别有用。
- 它们都将包含预定义的
responses
。 - 所有这些 path operations 都将在它们之前评估/执行
dependencies
列表。- 如果你还在特定的 path operation 中声明了依赖项,它们也会被执行。
- 路由器依赖项首先执行,然后是
decorator
中的dependencies
,最后是普通参数依赖项。 - 你还可以添加带有
scopes
的Security
依赖项。
Tip
在 APIRouter
中拥有 dependencies
可以用于,例如,要求对整个 path operations 组进行身份验证。即使没有将依赖项单独添加到每一个中。
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
是顶级目录 😱)...
这会导致错误,因为 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
文件是 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
等。
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
中包含一个 APIRouter
,使用:
router.include_router(other_router)
确保在将 router
包含在 FastAPI
应用中之前这样做,以便 other_router
的 路径操作 也被包含。