大型应用 - 多个文件¶
如果您正在构建应用程序或 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
类一样导入它并创建一个“实例”。
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
”类。
它支持所有相同的选项。
所有相同的参数
、响应
、依赖项
、标签
等。
提示
在这个例子中,变量名为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")
提示
如果可能,建议使用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
相同。
但我们希望更聪明一点,并简化代码。
我们知道此模块中的所有路径操作都具有相同的
- 路径
前缀
:/items
。 标签
:(只有一个标签:items
)。- 额外的
响应
。 依赖项
:它们都需要我们创建的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
。
我们还可以添加标签
和额外响应
的列表,这些列表将应用于此路由器中包含的所有路径操作。
我们还可以添加依赖项
列表,这些依赖项将添加到路由器中的所有路径操作中,并在对它们发出的每个请求中执行/解决。
提示
请注意,与路径操作装饰器中的依赖项非常相似,不会将任何值传递给您的路径操作函数。
最终结果是项目路径现在为
/items/
/items/{item_id}
…正如我们预期的那样。
- 它们将被标记为包含单个字符串
"items"
的标签列表。- 这些“标签”对于自动交互式文档系统(使用OpenAPI)特别有用。
- 所有这些都将包含预定义的
响应
。 - 所有这些路径操作都将在其之前评估/执行
依赖项
列表。- 如果您还在特定的路径操作中声明依赖项,它们也将被执行。
- 路由器依赖项首先执行,然后是装饰器中的
依赖项
,然后是正常的参数依赖项。 - 您还可以添加带有
作用域
的安全
依赖项。
提示
例如,可以在APIRouter
中使用依赖项
来要求对整个路径操作组进行身份验证。即使没有将依赖项单独添加到每个依赖项中。
检查
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"}
相对导入的工作原理¶
提示
如果您完全了解导入的工作原理,请继续下一节。
单个点.
,例如在
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
文件等。但我们没有这个。因此,这将在我们的示例中引发错误。🚨
但现在您知道它是如何工作的了,因此无论您的应用程序多么复杂,您都可以在自己的应用程序中使用相对导入。🤓
添加一些自定义的标签
、响应
和依赖项
¶
我们没有将前缀/items
或tags=["items"]
添加到每个路径操作中,因为我们已将它们添加到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"}
提示
最后一个路径操作将具有标签组合:["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
信息
第一个版本是“相对导入”
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!"}
信息
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
中
from fastapi import APIRouter
router = APIRouter()
@router.post("/")
async def update_admin():
return {"message": "Admin getting schwifty"}
但我们仍然希望在包含APIRouter
时设置自定义前缀
,以便其所有路径操作都以/admin
开头,我们希望使用我们已经为此项目设置的依赖项
来保护它,并且我们希望包含标签
和响应
。
我们可以通过将这些参数传递给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
¶
就像你可以在FastAPI
应用程序中包含一个APIRouter
一样,你也可以使用以下方法在另一个APIRouter
中包含一个APIRouter
router.include_router(other_router)
确保在将router
包含在FastAPI
应用程序中之前执行此操作,以便也包含other_router
中的路径操作。