FastAPI 提供了便捷的方式管理 HTTP 標頭和回應，讓開發者能精確控制伺服器與客戶端的互動。除了設定自定義標頭，也能輕鬆讀取請求標頭，並依需求調整回應狀態碼，支援 JSON、HTML 和串流等多元回應格式，滿足不同應用場景。此外，FastAPI 也能良好地整合資料函式庫，無論是透過 DB-API 直接操作，或是使用 SQLAlchemy 和 PyMongo 等 ORM 工具，都能簡化資料函式庫互動流程。非同步支援更是 FastAPI 的一大亮點，aiosqlite 和 Motor 等非同步驅動程式，讓資料函式庫操作能以非同步方式執行，大幅提升應用程式效能，滿足高併發需求。
FastAPI 中的 HTTP 標頭與回應控制
在現代的 Web 開發中，HTTP 標頭與回應控制扮演著至關重要的角色。FastAPI 作為一個現代、快速（高效能）的 Web 框架，提供了強大的功能來處理 HTTP 請求和回應。本章將探討如何在 FastAPI 中使用 HTTP 標頭、控制回應狀態碼以及處理不同型別的回應。
設定 HTTP 回應標頭
在 FastAPI 中，您可以輕鬆地在回應中新增自定義的 HTTP 標頭。這可以透過在操作函式（Operation Function）中使用 Response 物件來實作。以下是一個簡單的例子：
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/header/")
def set_header():
data = {"message": "Hello World"}
headers = {"Custom-Header": "Custom-Value"}
return Response(content=str(data), headers=headers, media_type="application/json")
內容解密：
在上述範例中，我們匯入了 Response 物件並在操作函式中使用它來設定自定義的 HTTP 標頭。headers 引數接受一個字典，其中包含您想要設定的標頭名稱和值。透過這種方式，您可以根據應用需求靈活地新增或修改 HTTP 回應標頭。
讀取 HTTP 請求標頭
FastAPI 提供了 Header 類別來讀取客戶端請求中的 HTTP 標頭。以下是一個範例：
from typing import Optional
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/read_header/")
async def read_header(accept_language: Optional[str] = Header(None)):
return {"Language": accept_language}
內容解密：
在此範例中，我們使用 Header 類別來定義一個操作函式引數 accept_language，用於接收客戶端請求中的 accept-language 標頭值。FastAPI 自動將 HTTP 標頭名稱轉換為 Python 合法的變數名稱（例如，將 - 替換為 _）。這種機制使得處理 HTTP 標頭變得簡單直接。
控制 HTTP 回應狀態碼
HTTP 狀態碼是伺服器對客戶端請求回應的重要組成部分。FastAPI 允許您輕鬆地控制回應狀態碼。預設情況下，FastAPI 傳回 200 OK 狀態碼，但您可以根據需求更改它。
from fastapi import FastAPI, status
app = FastAPI()
@app.get("/hello/{name}", status_code=status.HTTP_201_CREATED)
async def sayhello(name: str):
return {"message": "Hello " + name}
內容解密：
在這個範例中，我們使用 status_code 引數來設定回應的 HTTP 狀態碼。FastAPI 提供了 status 模組，其中包含各種標準 HTTP 狀態碼的常數。這樣不僅使程式碼更具可讀性，也減少了硬編碼狀態碼可能帶來的錯誤。
處理不同型別的回應
FastAPI 支援多種回應型別，包括 JSON、HTML、純文字等。您可以根據需要選擇合適的回應型別。
JSON 回應
JSON 是 FastAPI 的預設回應格式。您可以直接傳回 Python 物件，FastAPI 將自動將其轉換為 JSON。
from fastapi import FastAPI
app = FastAPI()
@app.get("/json/")
def get_json():
data = {"message": "Hello World"}
return data
HTML 回應
對於需要傳回 HTML 的情況，FastAPI 提供了 HTMLResponse。
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
app = FastAPI()
@app.get("/html/", response_class=HTMLResponse)
def get_html():
data = """
<html>
<body>
<h3>Hello World</h3>
</body>
</html>
"""
return data
內容解密：
在此範例中，我們使用 HTMLResponse 作為回應類別，使 FastAPI 將傳回的內容視為 HTML。這種方式使得傳回靜態或動態 HTML 網頁變得簡單。
StreamingResponse
對於需要串流資料的應用場景，FastAPI 提供了 StreamingResponse。
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
async def generator():
for i in range(10):
yield f"Number {i}\n"
@app.get("/stream/")
async def get_stream():
return StreamingResponse(generator(), media_type="text/plain")
內容解密：
在此範例中，我們定義了一個非同步生成器函式 generator()，它產生一系列數字。然後，我們使用 StreamingResponse 將這些數字串流給客戶端。這種方式適用於需要逐步傳回大量資料的場景。
圖表說明
@startuml
skinparam backgroundColor #FEFEFE
skinparam sequenceArrowThickness 2
title FastAPI 高效處理 HTTP 與資料函式庫操作
actor "客戶端" as client
participant "API Gateway" as gateway
participant "認證服務" as auth
participant "業務服務" as service
database "資料庫" as db
queue "訊息佇列" as mq
client -> gateway : HTTP 請求
gateway -> auth : 驗證 Token
auth --> gateway : 認證結果
alt 認證成功
gateway -> service : 轉發請求
service -> db : 查詢/更新資料
db --> service : 回傳結果
service -> mq : 發送事件
service --> gateway : 回應資料
gateway --> client : HTTP 200 OK
else 認證失敗
gateway --> client : HTTP 401 Unauthorized
end
@enduml圖表翻譯： 此圖表呈現了客戶端請求如何被 FastAPI 應用處理的流程。首先，客戶端發起請求到達 FastAPI 應用。應用接著處理請求，並根據需求進行不同的操作，包括設定 HTTP 標頭、讀取客戶端傳來的 HTTP 標頭、控制回應的狀態碼，以及選擇適合的回應型別，如 JSON、HTML 或串流回應等。其中，串流回應會將資料逐步傳送給客戶端。整個流程體現了 FastAPI 在處理 HTTP 請求和回應方面的靈活性和強大功能。
使用資料函式庫於FastAPI應用程式
在前面的章節中，我們已經瞭解了FastAPI應用程式的基本結構及其運作方式。現在，我們將探討如何使用資料函式庫來實作持久化的CRUD（建立、讀取、更新、刪除）操作。為了實作這些操作，FastAPI應用程式需要與資料儲存和檢索系統進行互動。在本章中，我們將重點介紹如何使用關係型資料函式庫和NoSQL資料函式庫。
6.1 DB-API簡介
DB-API是Python的一個標準API，用於與資料函式庫進行互動。它為不同的資料函式庫提供了一個統一的介面，使得開發者可以輕鬆地在不同的資料函式庫之間切換。
6.2 aiosqlite模組
aiosqlite是一個非同步的SQLite資料函式庫驅動程式，它允許開發者在非同步的環境中使用SQLite資料函式庫。下面是一個簡單的例子，展示瞭如何使用aiosqlite來建立一個資料函式庫連線並執行查詢：
import aiosqlite
async def main():
async with aiosqlite.connect('example.db') as db:
await db.execute('CREATE TABLE example (id INTEGER PRIMARY KEY, name TEXT)')
await db.commit()
cursor = await db.execute('SELECT * FROM example')
rows = await cursor.fetchall()
for row in rows:
print(row)
#### 內容解密：
1. 首先，我們匯入了`aiosqlite`模組。
2. 然後，我們定義了一個非同步函式`main`，在其中建立了一個與SQLite資料函式庫的連線。
3. 使用`await db.execute()`執行SQL查詢，建立了一個名為`example`的表格。
4. 使用`await db.commit()`提交了更改。
5. 執行了`SELECT`查詢，並使用`await cursor.fetchall()`取得了所有結果。
6. 最後，遍歷並列印了查詢結果。
### 6.3 SQLAlchemy
SQLAlchemy是一個流行的Python SQL工具包和ORM（物件關聯對映）系統。它提供了一種高階別的SQL抽象，使得開發者可以使用Python程式碼來操作資料函式庫。
#### 6.3.1 SQLAlchemy的基本使用
下面是一個使用SQLAlchemy的簡單例子：
```python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///example.db')
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
new_user = User(name='John Doe', age=30)
session.add(new_user)
session.commit()
#### 內容解密：
1. 首先，我們匯入了必要的模組並建立了一個資料函式庫引擎。
2. 定義了一個`User`類別，它繼承自`declarative_base()`傳回的基礎類別，並定義了對應的表格結構。
3. 使用`Base.metadata.create_all(engine)`建立了資料函式庫表格。
4. 建立了一個會話類別`Session`並例項化了一個會話物件`session`。
5. 建立了一個新的`User`物件，並將其新增到會話中。
6. 使用`session.commit()`提交了更改。
### 6.4 非同步SQLAlchemy
SQLAlchemy 1.4版本開始支援非同步操作。下面是一個使用非同步SQLAlchemy的例子：
```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
engine = create_async_engine('sqlite+aiosqlite:///example.db')
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def main():
async with async_session() as session:
# 非同步操作
pass
#### 內容解密：
1. 首先，我們匯入了必要的模組並建立了一個非同步資料函式庫引擎。
2. 定義了一個非同步會話類別`async_session`。
3. 在非同步函式`main`中，使用`async with async_session() as session:`建立了一個非同步會話。
### 6.5 PyMongo與MongoDB
PyMongo是一個Python的MongoDB驅動程式。下面是一個簡單的使用PyMongo的例子：
```python
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['example_db']
collection = db['example_collection']
document = {'name': 'John Doe', 'age': 30}
collection.insert_one(document)
#### 內容解密：
1. 首先，我們匯入了`MongoClient`類別並建立了一個客戶端物件。
2. 取得了一個資料函式庫物件和一個集合物件。
3. 建立了一個檔案並將其插入到集合中。
### 6.6 Motor與MongoDB
Motor是一個非同步的MongoDB驅動程式。下面是一個簡單的使用Motor的例子：
```python
from motor import motor_asyncio
client = motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017/')
db = client['example_db']
collection = db['example_collection']
async def main():
document = {'name': 'John Doe', 'age': 30}
await collection.insert_one(document)
#### 內容解密：
1. 首先，我們匯入了`AsyncIOMotorClient`類別並建立了一個非同步客戶端物件。
2. 取得了一個資料函式庫物件和一個集合物件。
3. 在非同步函式`main`中，建立了一個檔案並將其插入到集合中。
