資料庫連線
FastAPI 本身不內建資料庫 ORM 或連線驅動，通常搭配各資料庫官方／官方維護的 Python 套件：
| 資料庫 | 套件 | 特性 |
|---|---|---|
| MySQL | `mysql-connector-python` | MySQL 官方 Connector/Python |
| PostgreSQL | `psycopg` | PostgreSQL 官方推薦的 Python adapter |
| MongoDB | `pymongo` | MongoDB 官方 Python Driver |
以下範例都使用同步 Driver，因此 FastAPI 路由使用一般 `def`，避免在 `async def` 中直接執行同步資料庫操作而阻塞事件迴圈。
---
## 1. MySQL：`mysql-connector-python`
### 安裝
```bash
pip install fastapi uvicorn mysql-connector-python
```
### 範例
```python
# mysql_app.py
import os
import mysql.connector
from mysql.connector import pooling
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
# 建立連線池，而不是每次 request 都重新連線
app.state.mysql_pool = pooling.MySQLConnectionPool(
pool_name="fastapi_pool",
pool_size=5,
host=os.getenv("MYSQL_HOST", "localhost"),
port=int(os.getenv("MYSQL_PORT", "3306")),
user=os.getenv("MYSQL_USER", "root"),
password=os.getenv("MYSQL_PASSWORD", "password"),
database=os.getenv("MYSQL_DATABASE", "testdb"),
)
yield
# mysql-connector 的連線會在取出的 connection.close()
# 時歸還連線池；應用程式結束時不需逐一處理
app.state.mysql_pool = None
app = FastAPI(lifespan=lifespan)
@app.get("/users")
def get_users(request: Request):
pool = request.app.state.mysql_pool
conn = pool.get_connection()
try:
cursor = conn.cursor(dictionary=True)
cursor.execute(
"SELECT id, name, email FROM users ORDER BY id LIMIT 20"
)
rows = cursor.fetchall()
cursor.close()
return rows
finally:
# 歸還連線池
conn.close()
```
執行：
```bash
export MYSQL_HOST=localhost
export MYSQL_USER=root
export MYSQL_PASSWORD=secret
export MYSQL_DATABASE=testdb
uvicorn mysql_app:app --reload
```
`users` 資料表可以先建立：
```sql
CREATE TABLE users (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL,
email VARCHAR(255) NOT NULL
);
```
---
## 2. PostgreSQL：`psycopg`
`psycopg` 是 Psycopg 3，建議使用連線池。
### 安裝
```bash
pip install fastapi uvicorn "psycopg[binary,pool]"
```
### 範例
```python
# postgres_app.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql://postgres:password@localhost:5432/testdb",
)
@asynccontextmanager
async def lifespan(app: FastAPI):
pool = ConnectionPool(
conninfo=DATABASE_URL,
min_size=1,
max_size=5,
kwargs={"row_factory": dict_row},
open=False,
)
pool.open()
app.state.pg_pool = pool
yield
pool.close()
app = FastAPI(lifespan=lifespan)
@app.get("/users")
def get_users(request: Request):
pool: ConnectionPool = request.app.state.pg_pool
# 使用完畢後，連線會歸還到 pool
with pool.connection() as conn:
rows = conn.execute(
"""
SELECT id, name, email
FROM users
ORDER BY id
LIMIT 20
"""
).fetchall()
return rows
```
執行：
```bash
export DATABASE_URL="postgresql://postgres:secret@localhost:5432/testdb"
uvicorn postgres_app:app --reload
```
建立資料表：
```sql
CREATE TABLE users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(255) NOT NULL
);
```
使用參數時不要直接拼接 SQL 字串，應使用參數化查詢：
```python
with pool.connection() as conn:
user = conn.execute(
"SELECT id, name FROM users WHERE id = %s",
(user_id,),
).fetchone()
```
---
## 3. MongoDB：`pymongo`
PyMongo 的 `MongoClient` 本身會管理連線池，因此通常整個 FastAPI 應用程式共用一個 `MongoClient`。
### 安裝
```bash
pip install fastapi uvicorn pymongo
```
### 範例
```python
# mongo_app.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from pymongo import MongoClient
MONGO_URL = os.getenv(
"MONGO_URL",
"mongodb://localhost:27017",
)
@asynccontextmanager
async def lifespan(app: FastAPI):
client = MongoClient(MONGO_URL)
# 啟動時確認 MongoDB 可連線
client.admin.command("ping")
app.state.mongo_client = client
app.state.mongo_db = client["testdb"]
yield
# 應用程式關閉時關閉 client
client.close()
app = FastAPI(lifespan=lifespan)
@app.get("/users")
def get_users(request: Request):
db = request.app.state.mongo_db
collection = db["users"]
documents = collection.find(
{},
{
"_id": 0,
"name": 1,
"email": 1,
},
).sort("name", 1).limit(20)
return list(documents)
```
執行：
```bash
export MONGO_URL="mongodb://localhost:27017"
uvicorn mongo_app:app --reload
```
新增資料的範例：
```python
@app.post("/users")
def create_user(request: Request, name: str, email: str):
collection = request.app.state.mongo_db["users"]
result = collection.insert_one({
"name": name,
"email": email,
})
return {
"id": str(result.inserted_id),
"name": name,
"email": email,
}
```
---
## 重要注意事項
### 1. 不要每次請求都建立新的資料庫連線
不建議這樣：
```python
@app.get("/users")
def get_users():
conn = mysql.connector.connect(...)
...
```
應在應用程式啟動時建立：
- MySQL：連線池
- PostgreSQL：`ConnectionPool`
- MongoDB：單一 `MongoClient`
### 2. 同步 Driver 使用一般 `def`
上述套件範例都是同步操作，因此使用：
```python
@app.get("/")
def endpoint():
...
```
FastAPI 會將一般 `def` 路由放到 thread pool 執行。
若使用：
```python
@app.get("/")
async def endpoint():
...
```
卻直接呼叫同步資料庫 Driver，可能阻塞 event loop。若應用程式需要完全非同步，可以考慮：
- MySQL：使用非同步相容的 Driver
- PostgreSQL：使用 Psycopg 3 的 async API
- MongoDB：使用 PyMongo 的非同步 API（依安裝版本而定）
### 3. 使用環境變數保存連線資訊
不要把密碼直接提交到 Git。常見設定方式：
```bash
MYSQL_PASSWORD=secret
DATABASE_URL=postgresql://user:password@localhost:5432/dbname
MONGO_URL=mongodb://user:password@localhost:27017
```
實務上也可再加入：
- Pydantic Settings 管理設定
- SQLAlchemy 管理 MySQL/PostgreSQL ORM
- Alembic 管理 SQL migration
- MongoDB index 與 schema validation
- 交易、重試與健康檢查機制
相關學習地圖、教學課程
Python 後端工程、資料庫