Django 進階資料庫：連線池、Raw SQL 與交易管理 | Django 教學
2026/07/02
2026/05/25
當 Django 專案從開發階段邁向生產環境，資料庫操作的效能與可靠性就成為核心關注點。 連線池（Connection Pool） 減少重複建立連線的開銷， Raw SQL 讓你在 ORM 力有未逮時直接操作資料庫， 交易管理（Transaction Management） 確保資料的一致性與完整性。本篇將深入探討這些進階資料庫技術，幫助你打造高效能且穩定的 Django 後端。
連線池設定
每次 HTTP 請求建立新的資料庫連線都需要經歷 TCP 握手與認證，在高並發環境下這個開銷非常可觀。 連線池（Connection Pool） 透過重複使用現有連線來解決這個問題。
CONN_MAX_AGE：Django 內建連線持久化
Django 內建最簡單的連線持久化機制——CONN_MAX_AGE 參數：
# settings.py
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'mydb',
'USER': 'myuser',
'PASSWORD': 'mypassword',
'HOST': 'localhost',
'PORT': '5432',
'CONN_MAX_AGE': 600, # 連線存活 600 秒（10 分鐘）
}
}
| 設定值 | 行為 |
|---|---|
0（預設） | 每次請求結束後關閉連線 |
600 | 連線存活 10 分鐘，期間可重複使用 |
None | 連線永不過期（需注意資料庫端的逾時設定） |
需要注意的是，CONN_MAX_AGE 是 每個執行緒一個連線 的持久化機制，並非真正的連線池。如果使用 Gunicorn 搭配 4 Workers 和 2 threads，最多會有 8 個持久連線。
django-db-connection-pool：真正的連線池
如果需要更精細的連線池管理，可以使用第三方套件 django-db-connection-pool，它基於 SQLAlchemy 的連線池實作：
pip install django-db-connection-pool
# settings.py
DATABASES = {
'default': {
'ENGINE': 'dj_db_conn_pool.backends.postgresql',
'NAME': 'mydb',
'USER': 'myuser',
'PASSWORD': 'mypassword',
'HOST': 'localhost',
'PORT': '5432',
'POOL_OPTIONS': {
'POOL_SIZE': 10, # 連線池大小
'MAX_OVERFLOW': 10, # 超出 POOL_SIZE 時的最大額外連線數
'RECYCLE': 300, # 連線回收時間（秒）
},
}
}
pgBouncer：外部連線池
pgBouncer 是 PostgreSQL 專用的輕量級連線池代理，獨立於 Django 運行，適合大規模生產環境：
# /etc/pgbouncer/pgbouncer.ini
[databases]
mydb = host=127.0.0.1 port=5432 dbname=mydb
[pgbouncer]
listen_port = 6432
listen_addr = 127.0.0.1
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction # 以交易為單位分配連線
max_client_conn = 1000 # 最大客戶端連線數
default_pool_size = 20 # 每個資料庫的預設連線池大小
# settings.py（Django 連線到 pgBouncer）
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'mydb',
'HOST': '127.0.0.1',
'PORT': '6432', # pgBouncer 的端口，非 PostgreSQL 的 5432
'CONN_MAX_AGE': 0, # 搭配 pgBouncer 時設為 0，避免兩層池衝突
}
}
Raw SQL 查詢
雖然 Django ORM 能處理大部分查詢需求，但某些複雜場景仍需要直接撰寫 SQL。Django 提供兩種方式執行 Raw SQL。
Manager.raw()：回傳 Model 實例
Manager.raw() 適合查詢結果能對應到 Model 欄位的情境：
# 基本用法——結果映射為 Model 實例
users = User.objects.raw('SELECT * FROM auth_user WHERE is_active = %s', [True])
for user in users:
print(user.username) # 可直接存取 Model 屬性
# 搭配 JOIN 查詢
articles = Article.objects.raw('''
SELECT a.*, u.username AS author_name
FROM blog_article a
JOIN auth_user u ON a.author_id = u.id
WHERE a.status = %s
ORDER BY a.created_at DESC
''', ['published'])
for article in articles:
print(f"{article.title} by {article.author_name}")
connection.cursor()：完全控制 SQL
connection.cursor() 提供最大的彈性，適合回傳非 Model 格式的資料：
from django.db import connection
from django.utils import timezone
from datetime import timedelta
def get_monthly_stats():
"""查詢月度統計資料"""
with connection.cursor() as cursor:
cursor.execute("""
SELECT
DATE_TRUNC('month', created_at) AS month,
COUNT(*) AS order_count,
SUM(amount) AS total_revenue
FROM orders_order
WHERE created_at >= %s
GROUP BY 1
ORDER BY 1 DESC
""", [timezone.now() - timedelta(days=365)])
# 將結果轉為 dict 列表
columns = [col[0] for col in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
SQL 注入防護
使用 Raw SQL 時，永遠不要使用字串格式化拼接 SQL，必須使用參數化查詢：
from django.db import connection
user_id = 42
# 危險！SQL 注入漏洞
cursor.execute(f"SELECT * FROM auth_user WHERE id = {user_id}")
# 危險！同樣有漏洞
cursor.execute("SELECT * FROM auth_user WHERE id = %s" % user_id)
# 安全！參數化查詢
cursor.execute("SELECT * FROM auth_user WHERE id = %s", [user_id])
Django 會自動處理參數的跳脫與型別轉換，確保惡意輸入無法改變 SQL 語句的結構。
資料庫函式
Django ORM 提供豐富的 資料庫函式（Database Functions） ，讓你在不寫 Raw SQL 的情況下使用資料庫層級的運算。
常用函式一覽
from django.db.models.functions import (
Concat, Coalesce, Now, ExtractYear, ExtractMonth,
Upper, Lower, Length, Trim, Left, Right,
Cast, Round, Abs,
)
from django.db.models import Value, CharField, FloatField
# Concat：字串串接
from django.db.models.functions import Concat
users = User.objects.annotate(
full_name=Concat('first_name', Value(' '), 'last_name')
)
# Coalesce：取第一個非 NULL 值（類似 SQL 的 COALESCE）
articles = Article.objects.annotate(
display_name=Coalesce('nickname', 'username', Value('匿名'))
)
# Now：取得資料庫當前時間
from django.db.models.functions import Now
expired_items = Item.objects.filter(expires_at__lt=Now())
# ExtractYear / ExtractMonth：提取日期部分
articles_2026 = Article.objects.annotate(
year=ExtractYear('created_at')
).filter(year=2026)
monthly_stats = Article.objects.annotate(
month=ExtractMonth('created_at')
).values('month').annotate(count=Count('id'))
自訂資料庫函式
from django.db.models import Func
class ArrayLength(Func):
"""PostgreSQL array_length 函式"""
function = 'array_length'
template = '%(function)s(%(expressions)s, 1)'
# 使用自訂函式
products = Product.objects.annotate(
tag_count=ArrayLength('tags')
).filter(tag_count__gte=3)
交易管理（Transaction Management）
交易（Transaction） 確保一組資料庫操作要麼全部成功，要麼全部回滾，是維護資料一致性的關鍵機制。Django 透過 transaction.atomic() 提供簡潔的交易管理 API。
transaction.atomic() 基本用法
from django.db import transaction
# 方法一：with 語句（推薦）
def transfer_money(from_account_id, to_account_id, amount):
with transaction.atomic():
# select_for_update() 鎖定資料列，防止並發衝突
from_acc = Account.objects.select_for_update().get(pk=from_account_id)
to_acc = Account.objects.select_for_update().get(pk=to_account_id)
if from_acc.balance < amount:
raise ValueError("餘額不足")
from_acc.balance -= amount
to_acc.balance += amount
from_acc.save()
to_acc.save()
# 發生例外時，整個 block 自動回滾
# 方法二：裝飾器
@transaction.atomic
def create_order_with_items(order_data, items_data):
order = Order.objects.create(**order_data)
for item in items_data:
OrderItem.objects.create(order=order, **item)
return order
手動 Savepoint：巢狀交易
transaction.atomic() 支援巢狀使用，內層會建立 Savepoint（儲存點） ：
from django.db import transaction
def complex_operation():
with transaction.atomic(): # 外層：建立真正的交易
obj1 = Model1.objects.create(name='obj1')
try:
with transaction.atomic(): # 內層：建立 Savepoint
obj2 = Model2.objects.create(name='obj2')
raise ValueError("內部操作失敗") # 僅回滾此 Savepoint
except ValueError:
pass # obj2 被回滾，但 obj1 保留
obj3 = Model3.objects.create(name='obj3') # 繼續正常執行
# 最終結果：obj1 和 obj3 被提交，obj2 被回滾
on_commit：交易提交後執行
transaction.on_commit() 確保回呼函式只在交易成功提交後才執行，適合發送通知、觸發非同步任務等場景：
from django.db import transaction
def create_order(request):
with transaction.atomic():
order = Order.objects.create(
customer=request.user,
total=calculate_total(request)
)
# 確保 commit 後才發送通知（避免通知成功但 DB 回滾）
transaction.on_commit(
lambda: send_confirmation_email.delay(order.id)
)
return order
select_for_update() 悲觀鎖定
from django.db import transaction, DatabaseError
# NOWAIT：如果資料列已被鎖定，立即拋出例外
with transaction.atomic():
try:
product = Product.objects.select_for_update(nowait=True).get(pk=1)
product.stock -= 1
product.save()
except DatabaseError:
raise ValueError("庫存更新衝突，請重試")
# SKIP LOCKED：跳過已鎖定的資料列（適合任務佇列模式）
with transaction.atomic():
tasks = Task.objects.select_for_update(skip_locked=True).filter(
status='pending'
)[:10]
for task in tasks:
process_task(task)
多資料庫設定
Django 支援同時連接多個資料庫，透過 DATABASE_ROUTERS 自動決定讀寫路由。
DATABASES 多個 key
# settings.py
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'primary_db',
'HOST': 'primary.db.internal',
},
'replica': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'primary_db',
'HOST': 'replica.db.internal',
'TEST': {'MIRROR': 'default'}, # 測試時鏡像主庫
},
'analytics': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'analytics_db',
'HOST': 'analytics.db.internal',
},
}
DATABASE_ROUTERS = ['myapp.routers.ReadWriteRouter', 'myapp.routers.AnalyticsRouter']
自訂 Router：讀寫分離
# myapp/routers.py
class ReadWriteRouter:
"""讀寫分離路由：寫入導向 Primary，讀取導向 Replica"""
def db_for_read(self, model, **hints):
if model._meta.app_label not in ('analytics',):
return 'replica'
return None
def db_for_write(self, model, **hints):
return 'default'
def allow_relation(self, obj1, obj2, **hints):
db_set = {'default', 'replica'}
if obj1._state.db in db_set and obj2._state.db in db_set:
return True
return None
def allow_migrate(self, db, app_label, model_name=None, **hints):
return db == 'default'
class AnalyticsRouter:
"""Analytics App 路由至獨立資料庫"""
def db_for_read(self, model, **hints):
if model._meta.app_label == 'analytics':
return 'analytics'
return None
def db_for_write(self, model, **hints):
if model._meta.app_label == 'analytics':
return 'analytics'
return None
def allow_migrate(self, db, app_label, model_name=None, **hints):
if app_label == 'analytics':
return db == 'analytics'
return None
手動指定資料庫
# 使用 .using() 手動指定資料庫（繞過 Router）
User.objects.using('replica').filter(is_active=True)
Analytics.objects.using('analytics').create(event='page_view')
PostgreSQL 特定功能
PostgreSQL 是 Django 生產環境的首選資料庫，Django 為其提供了許多專屬功能。
特殊欄位類型
from django.contrib.postgres.fields import ArrayField, HStoreField
from django.db import models
class Product(models.Model):
name = models.CharField(max_length=200)
# ArrayField：儲存陣列資料
tags = ArrayField(
models.CharField(max_length=50),
default=list,
blank=True
)
# JSONField：儲存 JSON 資料（Django 3.1+ 內建，不限 PostgreSQL）
metadata = models.JSONField(default=dict, blank=True)
# HStoreField：儲存鍵值對（需啟用 hstore 擴充）
attributes = HStoreField(default=dict, blank=True)
# ArrayField 查詢
Product.objects.filter(tags__contains=['django']) # 包含 'django'
Product.objects.filter(tags__overlap=['django', 'python']) # 重疊
Product.objects.filter(tags__len=3) # 陣列長度為 3
# JSONField 查詢
Product.objects.filter(metadata__brand='Apple') # JSON 欄位查詢
Product.objects.filter(metadata__specs__ram__gte=16) # 巢狀欄位查詢
全文搜尋（Full-Text Search）
PostgreSQL 的全文搜尋功能強大，Django 透過 SearchVector 和 SearchQuery 提供完整支援：
from django.contrib.postgres.search import (
SearchVector, SearchQuery, SearchRank, TrigramSimilarity
)
# 基本全文搜尋
articles = Article.objects.annotate(
search=SearchVector('title', 'content')
).filter(search=SearchQuery('Django 部署'))
# 搜尋排名
vector = SearchVector('title', weight='A') + SearchVector('content', weight='B')
query = SearchQuery('Django 部署')
articles = Article.objects.annotate(
rank=SearchRank(vector, query)
).filter(rank__gte=0.1).order_by('-rank')
# 模糊搜尋（Trigram Similarity，需啟用 pg_trgm 擴充）
articles = Article.objects.annotate(
similarity=TrigramSimilarity('title', 'Djnago') # 拼錯也能搜到
).filter(similarity__gt=0.3).order_by('-similarity')
資料庫效能監控
在生產環境中，持續監控資料庫效能是預防問題的關鍵。
Django Debug Toolbar（開發環境）
# settings.py（僅開發環境）
if DEBUG:
INSTALLED_APPS += ['debug_toolbar']
MIDDLEWARE = ['debug_toolbar.middleware.DebugToolbarMiddleware'] + MIDDLEWARE
INTERNAL_IPS = ['127.0.0.1']
django-debug-toolbar 可以在每個頁面顯示執行的 SQL 查詢數量、耗時與重複查詢，是開發階段最實用的除錯工具。
django-silk（開發與測試環境）
# settings.py
INSTALLED_APPS += ['silk']
MIDDLEWARE = ['silk.middleware.SilkyMiddleware'] + MIDDLEWARE
django-silk 提供更詳細的效能分析，包含每個請求的 SQL 查詢、執行時間分佈與 Python 程式碼的效能剖析。
自訂 SQL 日誌（生產環境）
# settings.py
LOGGING = {
'version': 1,
'handlers': {
'console': {
'class': 'logging.StreamHandler',
},
},
'loggers': {
'django.db.backends': {
'handlers': ['console'],
'level': 'DEBUG', # 記錄所有 SQL 查詢
'propagate': False,
},
},
}
查詢效能監控指標
| 監控項目 | 建議閾值 | 工具 |
|---|---|---|
| 單次請求 SQL 查詢數 | < 20 | django-debug-toolbar |
| 單次查詢執行時間 | < 100ms | django-silk、EXPLAIN ANALYZE |
| 連線池使用率 | < 80% | pgBouncer SHOW POOLS |
| 慢查詢 | > 1s | PostgreSQL pg_stat_statements |
總結
進階資料庫操作是 Django 應用邁向生產環境的必備知識，正確的設定與使用能大幅提升效能與可靠性。以下整理本篇的重點：
- 連線池 減少重複建立連線的開銷——
CONN_MAX_AGE提供最簡單的連線持久化，django-db-connection-pool實現真正的連線池，pgBouncer適合大規模生產環境 - Raw SQL 讓你在 ORM 無法表達的場景直接操作資料庫——
Manager.raw()回傳 Model 實例，connection.cursor()提供完全控制，但必須使用參數化查詢防止 SQL 注入 - 資料庫函式 讓你在不寫 Raw SQL 的情況下使用資料庫層級運算——
Concat、Coalesce、Now、ExtractYear等函式涵蓋常見需求 - 交易管理 確保資料一致性——
transaction.atomic()支援巢狀 Savepoint，on_commit()確保提交後才執行副作用，select_for_update()實現悲觀鎖定 - 多資料庫設定 實現讀寫分離——
DATABASE_ROUTERS自動路由讀寫請求到不同資料庫 - PostgreSQL 特定功能 提供
ArrayField、JSONField、全文搜尋等強大功能，是 Django 生產環境的首選資料庫 - 資料庫效能監控 應從開發階段就開始——
django-debug-toolbar和django-silk幫助你即時發現效能瓶頸
下一篇我們將介紹 Django 的 套件生態系 ，盤點各類必裝與推薦的第三方套件，幫助你快速建立功能完整的 Django 專案。