FastAPI（40）- 大型應用程序的項目拆分

背景

• 假設要搭建一個測試平台，那么整個項目的 API 數量肯定很多個，他們不可能放在同一個文件中
• FastAPI 提供了一個方便的工具來構建應用程序，同時保持所有的靈活性

 

項目架構

假設結構如下

.
├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   └── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── internal
│       ├── __init__.py
│       └── admin.py

• main：應用程序的主入口，會添加所有子路由
• dependencies：存放應用程序要用到的依賴項
• routers：子路由，根據模塊划分，比如 users 存放用戶信息相關的路由，items 存放其他內容的路由
• internal：一些公共路由

 

APIRouter

有點像 Flask 里面的藍圖，為某個模塊創建路徑操作

 

users.py 代碼

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/28 7:26 下午
# file: users.py
"""

from fastapi import APIRouter, Depends, HTTPException
from dependencies import get_token_header

# 屬於該模塊的路由
user_router = APIRouter(
    # 這里配置的 tags、dependencies、responses 對這個模塊的內的所有路徑操作都生效
    # 路徑前綴，該模塊下所有路徑操作的前綴
    prefix="/users",
    # 標簽
    tags=["users"],
    # 依賴項
    dependencies=[Depends(get_token_header)],
    # 響應
    responses={404: {"description": "users Not found"}}
)


@user_router.get('/account/login')
async def login():
    return {}


@user_router.get('/account/logout')
async def logout():
    return {}


# 單獨給某個路徑操作聲明 tags、responses
@user_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.py 代碼

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/28 7:26 下午
# file: items.py
"""

from fastapi import APIRouter, Depends
from dependencies import get_token_header

item_router = APIRouter(
    # 這里配置的 tags、dependencies、responses 對這個模塊的內的所有路徑操作都生效
    # 路徑前綴，該模塊下所有路徑操作的前綴
    prefix="/items",
    # 標簽
    tags=["items"],
    # 依賴項
    dependencies=[Depends(get_token_header)],
    # 響應
    responses={404: {"description": "items Not found"}}
)


@item_router.get('/item')
async def index():
    return {}


@item_router.get('/item/list')
async def list():
    return {}

 

routers/__init__.py

from .items import item_router
from .users import user_router

 

admin.py

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/28 7:26 下午
# file: admin.py
"""

from fastapi import APIRouter

admin_router = APIRouter()


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

 

common/__init__.py

from .admin import admin_router

 

main.py

#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠蘿測試筆記
# blog:  https://www.cnblogs.com/poloyy/
# time: 2021/9/28 7:25 下午
# file: main.py
"""
import uvicorn
from fastapi import FastAPI, Depends

# 導入子路由
from dependencies import get_query_token, get_token_header
from routers import user_router, item_router
from common import admin_router

# 主路由
app = FastAPI(
    # 聲明全局依賴項
    # 如果每個 APIRouter 都會用到這個依賴項，那么應該聲明為全局依賴項
    dependencies=[Depends(get_query_token)]
)

# 添加子路由
app.include_router(user_router)
app.include_router(item_router)
# 也可以在這里給子路由聲明 prefix、tags、dependencies、responses，而無需修改原始的 APIRouter
app.include_router(
    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!"}


if __name__ == "__main__":
    uvicorn.run(app="main:app", host="127.0.0.1", port=8080, debug=True, reload=True)

 

重點

• 使用 app.include_router() 可以將每個 APIRouter 添加到主 FastAPI 應用程序中，它將包括來自該路由器的所有路由作為它的一部分
• 它實際上會在內部為 APIRouter 中聲明的每個路徑操作創建一個路徑操作，因此，在幕后，它實際上會像所有東西都是同一個應用程序一樣工作
• 使用 app.include_router() 時，不必擔心性能，只需要幾微秒，並且只會在啟動時發生，所以不會影響性能

 

主程序-子路由的架構圖

• 這個架構還是比較簡單的，主程序直接包含子路由
• 但其實子路由還可以再包含一個子路由
• 個人把主程序也稱為主路由（感覺更好理解）

user_router = APIRouter(
    prefix="/users",
)

user_router.include_router(
    router=item_router
    prefix="/items"
)

 

FastAPI() 的 include_router() 源碼