python如何通過FastAPI構(gòu)建復(fù)雜的Web?API
通過FastAPI構(gòu)建復(fù)雜的Web API
構(gòu)建復(fù)雜的 Web API 通常涉及到多個方面，包括良好的架構(gòu)設(shè)計(jì)、清晰的路由組織、數(shù)據(jù)驗(yàn)證與處理、安全措施、性能優(yōu)化等。使用 FastAPI 構(gòu)建復(fù)雜 Web API 的關(guān)鍵在于充分利用其提供的工具和特性，同時遵循軟件工程的最佳實(shí)踐。以下是一個詳細(xì)的指南，幫助你使用 FastAPI 構(gòu)建一個復(fù)雜且高效的 Web API。
1. 項(xiàng)目結(jié)構(gòu)
為你的項(xiàng)目創(chuàng)建一個合理的文件夾結(jié)構(gòu)，這有助于代碼的組織和維護(hù)。例如：
my_project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── dependencies.py
│ ├── models.py
│ ├── schemas.py
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── items.py
│ │ └── users.py
│ ├── services/
│ │ ├── __init__.py
│ │ ├── item_service.py
│ │ └── user_service.py
│ ├── database.py
│ └── config.py
└── tests/
├── __init__.py
└── test_main.py
2. 依賴管理
使用 requirements.txt 或者更現(xiàn)代的 poetry 來管理項(xiàng)目的依賴關(guān)系。確保所有開發(fā)和生產(chǎn)環(huán)境所需的庫都被列出。
# 使用 poetry 初始化項(xiàng)目并添加依賴 poetry init poetry add fastapi uvicorn sqlalchemy alembic bcrypt passlib[bcrypt] pydantic email-validator
3. 配置管理
使用環(huán)境變量或配置文件來管理應(yīng)用程序的不同設(shè)置（如數(shù)據(jù)庫連接字符串、密鑰等）?？梢越柚?Pydantic 的 BaseSettings 類。
# config.py
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = "My Complex API"
admin_email: str
items_per_user: int = 50
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
class Config:
env_file = ".env"
settings = Settings()
4. 數(shù)據(jù)庫集成
選擇合適的 ORM（如 SQLAlchemy）并與 FastAPI 集成。創(chuàng)建數(shù)據(jù)庫模型，并考慮使用 Alembic 進(jìn)行數(shù)據(jù)庫遷移。
# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
5. 定義模型與模式
創(chuàng)建數(shù)據(jù)模型用于映射到數(shù)據(jù)庫表，以及 Pydantic 模型用于請求體和響應(yīng)的驗(yàn)證。
# models.py
from sqlalchemy import Column, Integer, String
from .database import Base
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String, index=True)
# schemas.py
from pydantic import BaseModel
class ItemCreate(BaseModel):
name: str
description: str = None
class Item(BaseModel):
id: int
name: str
description: str = None
class Config:
orm_mode = True
6. 服務(wù)層
為了保持控制器（路由處理函數(shù)）的簡潔性，將業(yè)務(wù)邏輯分離到服務(wù)層中。
# services/item_service.py
from typing import List
from sqlalchemy.orm import Session
from ..models import Item as ItemModel
from ..schemas import ItemCreate, Item
def get_items(db: Session, skip: int = 0, limit: int = 10) -> List[Item]:
return db.query(ItemModel).offset(skip).limit(limit).all()
def create_item(db: Session, item: ItemCreate) -> Item:
db_item = ItemModel(**item.dict())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
7. 路由分組
將相關(guān)路徑操作分組到不同的路由器模塊中，以提高代碼的可讀性和可維護(hù)性。
# routers/items.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ... import schemas, services
from ...dependencies import get_db
router = APIRouter()
@router.get("/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
items = services.get_items(db, skip=skip, limit=limit)
return items
@router.post("/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
return services.create_item(db=db, item=item)
8. 主應(yīng)用入口
在主應(yīng)用文件中導(dǎo)入并包含各個路由器。
# main.py from fastapi import FastAPI from .routers import items, users app = FastAPI() app.include_router(items.router) app.include_router(users.router)
9. 安全性
實(shí)現(xiàn)身份驗(yàn)證和授權(quán)機(jī)制，例如使用 JWT Token 進(jìn)行用戶認(rèn)證。
# dependencies.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from . import models, schemas, config
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, config.settings.secret_key, algorithms=[config.settings.algorithm])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = schemas.TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(db, username=token_data.username)
if user is None:
raise credentials_exception
return user
10. 測試
編寫單元測試和集成測試來確保你的 API 按預(yù)期工作。FastAPI 提供了方便的測試客戶端 TestClient。
# tests/test_main.py
from fastapi.testclient import TestClient
from my_project.app.main import app
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
11. 部署
考慮使用 Docker 容器化你的應(yīng)用程序，并通過 CI/CD 管道自動部署到云平臺或服務(wù)器上。
# Dockerfile FROM tiangolo/uvicorn-gunicorn-fastapi:python3.9 COPY ./app /app WORKDIR /app RUN pip install --no-cache-dir -r requirements.txt CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]
總結(jié)
通過以上步驟可以使用 FastAPI 構(gòu)建復(fù)雜的 Web API。
FastAPI支持的工具
FastAPI 提供了豐富的工具和特性，使得開發(fā)高效、安全且易于維護(hù)的 Web API 變得更加簡單。除了核心功能外，還有一些重要的工具和特性可以幫助你構(gòu)建更強(qiáng)大的應(yīng)用程序。以下是 FastAPI 的一些重要工具和特性：
1. 依賴注入系統(tǒng)
FastAPI 內(nèi)置了一個強(qiáng)大且靈活的依賴注入系統(tǒng)，這使得你可以輕松地管理和重用代碼邏輯，比如身份驗(yàn)證、數(shù)據(jù)庫連接等。
from fastapi import Depends, FastAPI
app = FastAPI()
async def get_db():
db = DBSession()
try:
yield db
finally:
db.close()
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
items = db.query(Item).all()
return items
2. Pydantic 模型
Pydantic 是一個數(shù)據(jù)驗(yàn)證庫，它允許你定義數(shù)據(jù)模型，并自動處理數(shù)據(jù)解析和驗(yàn)證。FastAPI 使用 Pydantic 來確保請求體、查詢參數(shù)和其他輸入符合預(yù)期格式。
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
3. 中間件
中間件是在請求到達(dá)路由處理函數(shù)之前或響應(yīng)發(fā)送給客戶端之后執(zhí)行的代碼。你可以使用中間件來添加通用的行為，如日志記錄、身份驗(yàn)證、CORS 處理等。
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
4. 異常處理
FastAPI 提供了內(nèi)置的異常處理機(jī)制，并允許你自定義異常處理器來處理特定類型的錯誤。
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": exc.errors(), "body": exc.body},
)
5. 背景任務(wù)
如果你需要在返回響應(yīng)后繼續(xù)執(zhí)行某些操作（如發(fā)送電子郵件），可以使用背景任務(wù)。
from fastapi import BackgroundTasks
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, f"Notification sent to {email}\n")
return {"message": "Notification sent in the background"}
6. 靜態(tài)文件和模板
FastAPI 支持服務(wù)靜態(tài)文件（如 HTML、CSS、JavaScript 文件）以及渲染模板，這對于創(chuàng)建完整的 Web 應(yīng)用程序非常有用。
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/")
async def read_root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
7. WebSocket 支持
FastAPI 支持 WebSocket 連接，這對于實(shí)現(xiàn)實(shí)時雙向通信非常有用，例如聊天應(yīng)用或?qū)崟r更新的數(shù)據(jù)展示。
from fastapi import WebSocket, WebSocketDisconnect
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message text was: {data}")
except WebSocketDisconnect:
pass
8. 環(huán)境變量與配置管理
使用 pydantic 的 BaseSettings 類來管理應(yīng)用程序的配置和環(huán)境變量，這有助于分離配置和代碼。
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = "Awesome API"
admin_email: str
items_per_user: int = 50
class Config:
env_file = ".env"
settings = Settings()
9. 測試工具
FastAPI 提供了方便的測試工具，包括測試客戶端 TestClient 和模擬器，幫助你在本地快速測試 API。
from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
10. OpenAPI 和 Swagger UI
FastAPI 自動生成 OpenAPI 規(guī)范，并提供交互式的 API 文檔界面（Swagger UI 和 ReDoc）。這不僅方便開發(fā)者調(diào)試 API，也便于用戶了解如何使用你的 API。
# 自動生成并提供交互式文檔 # 訪問 http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc
11. 速率限制
雖然 FastAPI 本身不直接提供速率限制功能，但可以通過中間件或其他擴(kuò)展來實(shí)現(xiàn)這一特性，以防止濫用或保護(hù)服務(wù)器資源。
from fastapi import FastAPI, HTTPException, status
from fastapi.middleware import Middleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI(middleware=[
Middleware(TrustedHostMiddleware, allowed_hosts=["example.com"])
])
# 或者使用第三方庫如 fastapi-limiter 實(shí)現(xiàn)速率限制
這些工具和特性共同構(gòu)成了 FastAPI 強(qiáng)大而靈活的生態(tài)系統(tǒng)，使得開發(fā)者能夠更高效地構(gòu)建現(xiàn)代 Web API。
FastAPI 路由操作
FastAPI 支持所有標(biāo)準(zhǔn)的 HTTP 方法（也稱為“路由操作”或“HTTP 動詞”），這些方法定義了客戶端與服務(wù)器之間的交互類型。以下是 FastAPI 支持的主要路由操作：
1. GET
用于請求從服務(wù)器獲取信息，通常用于查詢資源。
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
2. POST
用于向指定資源提交數(shù)據(jù)，常用于創(chuàng)建新資源或發(fā)送表單數(shù)據(jù)。
@app.post("/items/")
async def create_item(item: Item):
return item
3. PUT
用于更新現(xiàn)有資源，通常整個資源會被替換。
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, "item": item}
4. DELETE
用于刪除指定的資源。
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
return {"item_id": item_id}
5. PATCH
用于對資源進(jìn)行部分更新，只修改指定字段。
@app.patch("/items/{item_id}")
async def patch_item(item_id: int, item: ItemUpdate):
return {"item_id": item_id, "item": item}
6. OPTIONS
用于描述目標(biāo)資源的通信選項(xiàng)。它返回服務(wù)器支持的方法列表。
@app.options("/items/")
async def describe_options():
return {"Allow": "GET, POST, PUT, DELETE, PATCH, OPTIONS"}
7. HEAD
類似于 GET 請求，但不返回消息體，僅用于獲取響應(yīng)頭信息。
@app.head("/items/{item_id}")
async def head_item(item_id: int):
# 處理邏輯，但不會返回響應(yīng)體
pass
8. TRACE
用于回顯服務(wù)器收到的請求，主要用于測試或診斷。
@app.trace("/trace")
async def trace_request():
return {"method": "TRACE"}
路由參數(shù)
除了上述的 HTTP 方法外，F(xiàn)astAPI 還允許你定義路徑參數(shù)、查詢參數(shù)、請求體等，以更靈活地處理不同的請求場景。
路徑參數(shù)
直接在路徑中定義參數(shù)，如 /items/{item_id} 中的 item_id。
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
查詢參數(shù)
通過 URL 的查詢字符串傳遞參數(shù)，例如 /items/?skip=0&limit=10。
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
請求體
使用 Pydantic 模型來解析和驗(yàn)證 JSON 請求體。
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
async def create_item(item: Item):
return item
額外功能
- 路徑操作函數(shù)：可以是同步函數(shù)（
def）或異步函數(shù)（async def），根據(jù)你的需求選擇。 - 依賴注入：可以在路徑操作函數(shù)中添加依賴項(xiàng)，以便于共享狀態(tài)或執(zhí)行預(yù)處理邏輯。
- 中間件：為所有請求添加額外的行為，比如日志記錄、身份驗(yàn)證等。
- 自定義響應(yīng)：你可以返回不同類型的響應(yīng)，包括 JSONResponse、HTMLResponse 等。
- WebSocket：FastAPI 同樣支持 WebSocket 連接，可以用來實(shí)現(xiàn)實(shí)時雙向通信。
以上就是 FastAPI 支持的主要路由操作。FastAPI 的設(shè)計(jì)使得它不僅支持標(biāo)準(zhǔn)的 HTTP 方法，還提供了強(qiáng)大的工具來幫助開發(fā)者構(gòu)建現(xiàn)代的 Web API。
總結(jié)
到此這篇關(guān)于python如何通過FastAPI構(gòu)建復(fù)雜的Web API的文章就介紹到這了,更多相關(guān)python FastAPI構(gòu)建Web API內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Django框架實(shí)現(xiàn)在線考試系統(tǒng)的示例代碼
這篇文章主要介紹了Django框架實(shí)現(xiàn)在線考試系統(tǒng)的示例代碼，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-11-11
Python實(shí)現(xiàn)爬取網(wǎng)頁中動態(tài)加載的數(shù)據(jù)
這篇文章主要介紹了Python實(shí)現(xiàn)爬取網(wǎng)頁中動態(tài)加載的數(shù)據(jù)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-08-08
這篇文章主要介紹了Python 防止死鎖的方法，文中講解非常細(xì)致，代碼幫助大家更好的理解和學(xué)習(xí)，感興趣的朋友可以了解下2020-07-07
python_tkinter彈出對話框創(chuàng)建2
這篇文章主要介紹了python_tkinter彈出對話框創(chuàng)建，上以篇文章我們簡單的對對話框創(chuàng)建做了簡單介紹，本文將繼續(xù)更多相關(guān)內(nèi)容，需要的小伙伴可以參考一下2022-03-03
python用reduce和map把字符串轉(zhuǎn)為數(shù)字的方法
最近在復(fù)習(xí)高階函數(shù)的時候，有一道題想了半天解不出來。于是上午搜索資料，看了下別人的解法，發(fā)現(xiàn)學(xué)習(xí)編程，思維真的很重要。下面這篇文章就來給大家介紹了python利用reduce和map把字符串轉(zhuǎn)為數(shù)字的思路及方法，有需要的朋友們可以參考借鑒，下面來一起看看吧。2016-12-12
python使用MQTT給硬件傳輸圖片的實(shí)現(xiàn)方法
最近因需要用python寫一個微服務(wù)來用MQTT給硬件傳輸圖片，其中python用的是flask框架。這篇文章主要介紹了python使用MQTT給硬件傳輸圖片,需要的朋友可以參考下2019-05-05
Django模板標(biāo)簽中url使用詳解(url跳轉(zhuǎn)到指定頁面)
這篇文章主要介紹了Django模板標(biāo)簽中url使用詳解(url跳轉(zhuǎn)到指定頁面)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-03-03
Python實(shí)現(xiàn)提取Excel指定關(guān)鍵詞的行數(shù)據(jù)
這篇文章主要為大家介紹了如何利用Python實(shí)現(xiàn)提取Excel指定關(guān)鍵詞的行數(shù)據(jù)，文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以跟隨小編一起動手試一試2022-03-03
如何解決Pycharm運(yùn)行報錯No Python interpreter selected
這篇文章主要介紹了如何解決Pycharm運(yùn)行時No Python interpreter selected問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-05-05
numpy中生成隨機(jī)數(shù)的幾種常用函數(shù)(小結(jié))
這篇文章主要介紹了numpy中生成隨機(jī)數(shù)的幾種常用函數(shù)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-08-08