FastAPI 高效能 RESTful API 開發

FastAPI 的高效能源於其非同步特性以及與 Starlette 和 Pydantic 的緊密整合。相較於 Flask，FastAPI 能夠更有效地處理大量併發請求，並透過型別提示實作自動資料驗證和序列化，減少開發時間和錯誤。其內建的互動式 API 檔案也簡化了開發和測試流程，提升整體開發效率。除了基本 API 構建外，FastAPI 也支援依賴注入，方便管理資料函式庫連線等資源，並提供簡潔的非同步任務處理機制。

使用 FastAPI 開發高效能的 RESTful API

FastAPI 是一個現代化的 Web 框架，專為使用 Python 3.6+ 開發高效能的 API 而設計。它利用 Python 的型別提示（Type Hints）來簡化 API 的開發流程，並提供出色的效能表現。本章將介紹 FastAPI 的核心功能、與 Flask 的比較，以及如何使用 FastAPI 構建 RESTful API。

從 Flask 遷移到 FastAPI

要從 Flask 或其他 Web 框架遷移到 FastAPI，需要了解其主要特點和優勢。FastAPI 利用 Python 的最新功能來促進 API 的開發，使其既快速又易於使用。

FastAPI 與 Flask 的主要差異

1. 效能

   • 非同步支援：FastAPI 根據 Starlette 和 Pydantic 構建，設計上支援非同步請求處理。搭配 Uvicorn 這個 ASGI 伺服器，可以原生支援非同步請求處理，使其效能優於根據 WSGI 的 Flask。

2. 型別提示與自動資料驗證

   • FastAPI 大量使用 Python 型別提示，不僅提高開發時的清晰度，還用於自動進行資料驗證和序列化。

   from fastapi import FastAPI
   from pydantic import BaseModel
   
   app = FastAPI()
   
   class Item(BaseModel):
       name: str
       description: str = None
       price: float
       tax: float = None
   
   @app.post("/items/")
   async def create_item(item: Item):
       return {"name": item.name, "price": item.price}
   

   在上述範例中，FastAPI 使用 Pydantic 模型（Item 類別）自動處理請求資料到 Python 型別的轉換、資料驗證，並生成檔案。

3. 依賴注入

   • FastAPI 提供了一個簡單而強大的依賴注入系統，允許在路由函式中注入可重用的依賴項。

   from fastapi import Depends, FastAPI
   
   def get_db():
       db = DBSession()
       try:
           yield db
       finally:
           db.close()
   
   @app.get("/items/")
   async def read_items(db = Depends(get_db)):
       items = db.get_items()
       return items
   

   上述程式碼片段展示瞭如何使用 Depends 建立一個依賴項，為每個請求檢索資料函式庫連線，並確保在請求完成後關閉連線。

4. 內建互動式 API 檔案

   • FastAPI 自動生成互動式 API 檔案，使用 Swagger UI 和 ReDoc。這些檔案包含即時 “Try it out” 功能，並根據程式碼自動生成，包含 API 的描述、引數型別等資訊。

使用 FastAPI 構建 RESTful API

FastAPI 的內建功能專為高效建立和管理 API 而設計。透過利用現代化的 Python 功能和 FastAPI 的內建功能，可以充分發揮 RESTful API 的能力。本文將介紹如何使用 FastAPI 構建 RESTful API，並展示它如何改進和簡化流程。

定義 RESTful 端點

FastAPI 使定義 RESTful 端點變得簡單直接。每個端點對應一個 HTTP 方法和路徑。使用裝飾器將 Python 函式連結到這些路徑和方法。

以下是一個使用 FastAPI 定義具有 GET 和 POST 方法的簡單 API 的範例：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.get("/")
async def read_root():
    return {"message": "Welcome to FastAPI"}

@app.post("/items/")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}

詳細解說：

1. @app.get("/")：定義了一個處理 GET 請求的端點，路徑為根目錄（"/"）。
2. @app.post("/items/")：定義了一個處理 POST 請求的端點，用於建立新的專案。
3. Item 類別：使用 Pydantic 的 BaseModel 定義了一個資料模型，用於驗證和解析請求中的 JSON 資料。
4. async def create_item(item: Item)：處理 POST 請求的函式，自動將請求的 JSON 資料轉換為 Item 例項。

FastAPI 技術深度解析與實務應用

FastAPI 是一個現代化的 Python 網頁框架，專為構建高效能、易於擴充套件的 API 而設計。其強大的功能和靈活性使其成為開發 RESTful 服務的理想選擇。

端點定義與資料處理

在 FastAPI 中，端點（endpoint）是用於處理特定 HTTP 請求的函式。以下是一個簡單的範例：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.post("/items/")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}

內容解密：

• @app.get("/") 和 @app.get("/items/{item_id}") 是用於讀取資料的端點。
• @app.post("/items/") 是用於建立資料的端點，使用 Pydantic 模型（Item）自動處理資料驗證和序列化。
• read_root、read_item 和 create_item 函式分別處理對應的 HTTP 請求。

路徑引數與查詢字串

FastAPI 提供了一種簡單的方法來定義路徑引數和查詢字串，這對於建立動態端點至關重要。

路徑引數

@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}

內容解密：

• read_user 函式使用 user_id 作為路徑引數。
• FastAPI 自動根據型別註解驗證和轉換輸入資料。

查詢引數

@app.get("/items/")
async def read_items(q: str = None):
    query = {"q": q}
    return query

內容解密：

• read_items 函式使用 q 作為查詢引數。
• 如果設定了預設值，則該引數是可選的；否則，它是必需的。

請求主體處理

對於更複雜的資料，FastAPI 允許您使用 Pydantic 模型輕鬆處理 JSON 主體，這提供了資料驗證和結構。

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "name": item.name, "price": item.price}

內容解密：

• update_item 函式使用 Pydantic 模型例項 item，從請求主體解析、驗證和序列化。

回應處理

FastAPI 提供了複雜的回應處理機制，允許對 HTTP 回應進行精細控制。

自定義狀態碼

@app.post("/items/", status_code=201)
async def create_item(item: Item):
    return item

內容解密：

• create_item 函式在路由裝飾器中直接指定 HTTP 狀態碼。

回應標頭

from fastapi import Response

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return Response(content=f"Item ID: {item_id}", media_type="text/plain")

內容解密：

• read_item 函式透過傳回 Response 物件來修改回應標頭。

錯誤處理

在 RESTful API 中處理錯誤時，FastAPI 允許您引發包含自定義標頭和內容的 HTTP 異常。

from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in item_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": item_db[item_id]}

內容解密：

• 如果找不到專案，則引發 HTTP 404 錯誤，提供清晰的錯誤訊息。

相依性注入

相依性注入是一種設計模式，有助於實作控制反轉（IoC）。它使系統更容易測試、管理和模組化。

定義相依性

from fastapi import Depends, HTTPException

def get_db():
    try:
        db = Database.connect()
        yield db
    finally:
        db.disconnect()

def get_current_user(db=Depends(get_db)):
    user_id = db.get_current_user_id()
    if not user_id:
        raise HTTPException(status_code=404, detail="User not found")
    return user_id

內容解密：

• get_db 是一個提供資料函式庫連線的相依性。
• get_current_user 是一個使用另一個相依性（get_db）從資料函式庫取得當前使用者 ID 的相依性。

在路由處理器中使用相依性

@app.get("/users/me")
def read_current_user(user_id: int = Depends(get_current_user)):
    return {"user_id": user_id}

內容解密：

• read_current_user 路由處理器依賴於 get_current_user 函式。
• FastAPI 在後台解析這些相依性並將結果（user_id）注入路徑操作函式。

大型應用程式中的相依性處理

在大型 FastAPI 應用程式中，您可能有多個相依性需要在應用程式的不同部分解析。FastAPI 允許您以層次結構方式構建這些相依性，其中高層次的相依性可以依賴於低層次的相依性。

def get_api_key(db=Depends(get_db)):
    key = db.get_api_key()
    if not key:
        raise HTTPException(status_code=403, detail="API key invalid")
    return key

@app.get("/data")
def read_data(api_key: str = Depends(get_api_key)):
    return {"data": "secret data"}

內容解密：

• read_data 取決於 get_api_key，而 get_api_key 又取決於 get_db。
• FastAPI 自動解析此層次結構，確保每個函式獲得其有效運作所需的相依性。

FastAPI 對相依性注入的內建支援，使用簡單的 Python 函式和 Depends 類別，簡化了原本可能是應用程式開發中複雜且繁瑣的部分。

FastAPI 進階資料函式庫整合與背景任務實作

使用 SQLAlchemy 提升資料函式庫操作能力

FastAPI 透過 SQLAlchemy 的 Object-Relational Mapping (ORM) 功能，實作強大的資料操作與擷取。SQLAlchemy 提供進階的資料函式倉管理功能，讓開發者能以 Python 程式碼與不同資料函式庫互動，並充分利用其查詢、交易管理等功能。

設定 SQLAlchemy 與 FastAPI

1. 安裝必要套件
   首先，安裝支援非同步操作的 SQLAlchemy 與 databases 套件：

   pip install sqlalchemy databases[sqlite]
   pip install asyncpg # 若使用 PostgreSQL，需安裝對應的驅動程式
   

2. 組態資料函式庫連線 URL
   定義資料函式庫連線字串，用於 SQLAlchemy 連線至資料函式庫：

   DATABASE_URL = "sqlite:///./test.db"
   # PostgreSQL 範例：postgresql://user:password@localhost/dbname
   

3. 建立資料函式庫與資料表
   使用 SQLAlchemy 的 ORM 定義模型，進而建立資料表：

   from sqlalchemy import create_engine, Column, Integer, String, MetaData
   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()
   
   class User(Base):
       __tablename__ = "users"
       id = Column(Integer, primary_key=True, index=True)
       username = Column(String, unique=True, index=True)
       email = Column(String, index=True)
       full_name = Column(String)
       disabled = Column(Boolean, default=False)
   
   # 建立資料表
   Base.metadata.create_all(bind=engine)
   

   內容解密：

   • create_engine：建立資料函式庫引擎，負責連線至指定的 SQLite 資料函式庫。
   • sessionmaker：建立 SessionLocal 類別，用於管理資料函式庫連線的生命週期。
   • declarative_base：建立基底類別 Base，用於定義 ORM 模型。
   • User 類別：定義 users 資料表結構，包含欄位如 id、username 等。
   • Base.metadata.create_all(bind=engine)：根據定義的模型建立對應的資料表。

將 SQLAlchemy 整合至 FastAPI

1. 資料函式庫 Session 相依性注入
   建立一個相依性函式 get_db，用於在路由處理函式中取得 Session：

   from fastapi import Depends, FastAPI, HTTPException
   from sqlalchemy.orm import Session
   
   app = FastAPI()
   
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   

   內容解密：

   • get_db 函式：使用 SessionLocal 建立一個新的資料函式庫 Session，並在請求結束後關閉。

2. 執行資料函式庫操作
   使用由 get_db 提供的 Session 執行 CRUD 操作：

   @app.get("/users/{user_id}", response_model=UserSchema)
   async def read_user(user_id: int, db: Session = Depends(get_db)):
       db_user = db.query(User).filter(User.id == user_id).first()
       if db_user is None:
           raise HTTPException(status_code=404, detail="User not found")
       return db_user
   

   內容解密：

   • read_user 路由：根據 user_id 從資料函式庫查詢使用者，若無結果則丟擲 404 錯誤。

3. 非同步處理資料函式庫操作
   若需完全的非同步處理，需使用 databases 套件搭配 SQLAlchemy Core：

   from databases import Database
   
   database = Database(DATABASE_URL)
   
   @app.on_event("startup")
   async def startup():
       await database.connect()
   
   @app.on_event("shutdown")
   async def shutdown():
       await database.disconnect()
   
   @app.get("/items/")
   async def read_items():
       query = items.select()
       return await database.fetch_all(query)
   

   內容解密：

   • database.connect() 與 database.disconnect()：在應用程式啟動與關閉時自動連線或斷開資料函式庫連線。
   • read_items 路由：非同步執行查詢並傳回結果。

實作背景任務

FastAPI 中的背景任務基礎

FastAPI 提供簡便的方式來定義背景任務，這些任務會在回應傳送給客戶端後執行，例如寄送電子郵件或處理檔案。

1. 定義背景任務
   使用 BackgroundTasks 類別新增背景任務：

   from fastapi import FastAPI, BackgroundTasks
   
   app = FastAPI()
   
   def write_log(message: str):
       with open("log.txt", "a") as log:
           log.write(f"{message}\n")
   
   @app.post("/send-notification/")
   async def send_notification(email: str, background_tasks: BackgroundTasks):
       background_tasks.add_task(write_log, message=f"notification sent to {email}")
       return {"message": "Notification sent in the background"}
   

   內容解密：

   • write_log 函式：將訊息寫入日誌檔案。
   • send_notification 路由：將寫入日誌的任務加入背景執行，並立即傳回回應。

更複雜的背景任務實作

對於較複雜或資源密集的任務，建議使用 Celery 結合訊息代理（如 RabbitMQ 或 Redis）進行處理。

1. 設定 Celery
   安裝 Celery 與訊息代理，並在 FastAPI 中進行設定：

from celery import Celery

def make_celery(app): celery = Celery( app.import_name, backend=app.config[‘CELERY_RESULT_BACKEND’], broker=app.config[‘CELERY_BROKER_URL’] ) celery.conf.update(app.config)

class ContextTask(celery.Task):
    def __call__(self, *args, **kwargs):
        with app.app_context():
            return self.run(*args, **kwargs)

celery.Task = ContextTask
return celery

celery = make_celery(app)

#### 內容解密：
- `make_celery` 函式：根據 FastAPI 應用程式組態建立 Celery 例項。
- `ContextTask` 類別：確保 Celery 任務在應用程式上下文中執行。

2. **定義 Celery 任務**  
```python
@celery.task()
def send_email(email: str):
    # 傳送電子郵件的邏輯
    pass

內容解密：

• send_email 任務：定義一個傳送電子郵件的 Celery 任務。

3. 從 FastAPI 中觸發 Celery 任務

@app.post("/send-email/")
async def send_email_endpoint(email: str):
    send_email.delay(email=email)
    return {"message": "Email is being sent in the background"}

內容解密：

• send_email.delay(email=email)：將傳送電子郵件的任務非同步加入 Celery佇列。