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 將這些數字串流給客戶端。這種方式適用於需要逐步傳回大量資料的場景。
圖表說明
graph LR A[客戶端請求] --> B[FastAPI應用] B --> C[處理請求] C --> D[設定HTTP標頭] C --> E[讀取HTTP標頭] C --> F[控制回應狀態碼] C --> G[選擇回應型別] G --> H[JSON回應] G --> I[HTML回應] G --> J[StreamingResponse] J --> K[串流資料給客戶端]
圖表翻譯： 此圖表呈現了客戶端請求如何被 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`中，建立了一個檔案並將其插入到集合中。
