ORM 整合實戰：TypeORM、Prisma、SQLAlchemy、Django ORM 與 PostgreSQL 進階型別映射 | PostgreSQL
2026/07/23
ORM（Object-Relational Mapping）框架在應用程式物件模型與 PostgreSQL 關聯式資料模型之間搭建橋梁。本文完整解析 TypeORM、Prisma、SQLAlchemy、Django ORM、Drizzle 五大框架與 PostgreSQL 的型別映射、進階查詢技巧、N+1 問題解法，以及何時該跳出 ORM 直接使用 Raw SQL。
ORM 的核心抽象層次
ORM 在應用程式與資料庫之間引入多個抽象層：
應用程式層（Application Layer）
│
▼
物件/類別定義（Entity / Model Class）
│ ← ORM 核心抽象
▼
SQL 查詢生成器（Query Builder）
│ ← 框架差異最大的地方
▼
資料庫驅動（Database Driver）
Node.js: pg / postgres.js
Python: psycopg2 / asyncpg
│
▼
PostgreSQL Wire Protocol → PostgreSQL Backend
抽象層帶來的能力： 型別安全（編譯期捕捉欄位錯誤）、遷移管理（Schema 變更版本化）、關係處理（自動 JOIN、Lazy/Eager Loading）。
抽象層的代價： 可能生成低效 SQL（N+1 問題）、遮蔽 PostgreSQL 特有功能（Array 操作、全文搜尋、WINDOW 函式）。
各框架 PostgreSQL 特性支援概覽
| 特性 | TypeORM | Prisma | SQLAlchemy | Django ORM | Drizzle |
|---|---|---|---|---|---|
| JSON/JSONB | 部分支援 | 完整支援 | 完整支援 | 完整支援 | 完整支援 |
| Array 型別 | 支援 | 需 Raw | 完整支援 | ArrayField | 支援 |
| 全文搜尋 | 需 Raw | 需 Raw | 完整支援 | 原生支援 | 需 Raw |
| ENUM | 原生支援 | 原生支援 | 原生支援 | 原生支援 | 原生支援 |
| RETURNING | 支援 | 支援 | 支援 | 部分支援 | 支援 |
| LISTEN/NOTIFY | 需繞過 | 需繞過 | 需繞過 | 需繞過 | 需繞過 |
| 資料表繼承 | STI/CTI | 不支援 | 完整支援 | 部分支援 | 不支援 |
| 遷移工具 | 內建 | 內建 | Alembic | 內建 | Drizzle Kit |
TypeORM — TypeScript/NestJS 生態主流
連線與連線池配置
import { DataSource } from 'typeorm';
const AppDataSource = new DataSource({
type: 'postgres',
host: process.env.DB_HOST || 'localhost',
port: parseInt(process.env.DB_PORT || '5432'),
username: process.env.DB_USER,
password: process.env.DB_PASSWORD,
database: process.env.DB_NAME,
ssl: process.env.NODE_ENV === 'production'
? { rejectUnauthorized: true, ca: process.env.DB_SSL_CA }
: false,
extra: {
max: 20, // 連線池最大連線數
min: 5, // 最小閒置連線數
idleTimeoutMillis: 30000, // 閒置逾時
connectionTimeoutMillis: 5000, // 建立逾時
statement_timeout: 10000, // 查詢逾時
application_name: 'my-app', // pg_stat_activity 顯示名稱
},
entities: [__dirname + '/**/*.entity{.ts,.js}'],
migrations: [__dirname + '/migrations/**/*{.ts,.js}'],
synchronize: false, // 生產環境必須關閉
logging: ['query', 'error'],
});
Entity 與 PostgreSQL 型別映射
import {
Entity, PrimaryGeneratedColumn, Column,
CreateDateColumn, UpdateDateColumn, Index
} from 'typeorm';
@Entity('products')
export class Product {
@PrimaryGeneratedColumn('uuid')
id: string;
@Column({ type: 'varchar', length: 255 })
name: string;
// NUMERIC（金融計算，避免浮點誤差）
@Column({ type: 'numeric', precision: 15, scale: 4 })
price: number;
// JSONB（結構化但彈性的資料）
@Column({ type: 'jsonb', nullable: true })
metadata: Record<string, unknown> | null;
// Array（PostgreSQL 原生陣列）
@Column({ type: 'text', array: true, default: '{}' })
tags: string[];
// Enum（對應 PostgreSQL ENUM）
@Column({
type: 'enum',
enum: ['draft', 'published', 'archived'],
default: 'draft',
})
status: 'draft' | 'published' | 'archived';
@CreateDateColumn({ type: 'timestamptz' })
createdAt: Date;
@UpdateDateColumn({ type: 'timestamptz' })
updatedAt: Date;
}
QueryBuilder 操作 PostgreSQL 特性
@Injectable()
export class ProductRepository {
constructor(
@InjectRepository(Product)
private readonly repo: Repository<Product>,
) {}
// JSONB 查詢
async findByMetadata(key: string, value: string): Promise<Product[]> {
return this.repo.createQueryBuilder('p')
.where("p.metadata ->> :key = :value", { key, value })
.getMany();
}
// Array 操作（ANY）
async findByTag(tag: string): Promise<Product[]> {
return this.repo.createQueryBuilder('p')
.where(":tag = ANY(p.tags)", { tag })
.getMany();
}
// 全文搜尋（需 Raw SQL）
async fullTextSearch(query: string): Promise<Product[]> {
return this.repo.query(`
SELECT * FROM products
WHERE search_vector @@ plainto_tsquery('chinese', $1)
ORDER BY ts_rank(search_vector, plainto_tsquery('chinese', $1)) DESC
LIMIT 20
`, [query]);
}
// UPSERT with RETURNING
async upsertProduct(data: Partial<Product>): Promise<Product> {
const result = await this.repo.createQueryBuilder()
.insert()
.into(Product)
.values(data)
.orUpdate(['name', 'price', 'metadata'], ['id'])
.returning('*')
.execute();
return result.raw[0];
}
}
Prisma — 現代 TypeScript ORM 的型別安全代表
Schema 定義與 PostgreSQL 特性
generator client {
provider = "prisma-client-js"
previewFeatures = ["postgresqlExtensions", "fullTextSearch"]
}
datasource db {
provider = "postgresql"
url = env("DATABASE_URL")
directUrl = env("DIRECT_URL") // 搭配 PgBouncer 時必要
extensions = [pgcrypto, pg_trgm, unaccent]
}
enum OrderStatus {
PENDING
CONFIRMED
SHIPPED
DELIVERED
CANCELLED
}
model Product {
id String @id @default(dbgenerated("gen_random_uuid()")) @db.Uuid
name String @db.VarChar(255)
price Decimal @db.Decimal(15, 4)
metadata Json?
status OrderStatus @default(PENDING)
createdAt DateTime @default(now()) @db.Timestamptz(3)
updatedAt DateTime @updatedAt @db.Timestamptz(3)
@@index([metadata], type: Gin)
@@map("products")
}
Prisma 查詢 PostgreSQL 特性
import { PrismaClient, Prisma } from '@prisma/client';
const prisma = new PrismaClient();
// JSONB 查詢（Prisma JSON 過濾語法）
async function queryByMetadata() {
return prisma.product.findMany({
where: {
metadata: {
path: ['category'],
equals: 'electronics',
},
},
});
}
// 全文搜尋（需啟用 previewFeature）
async function fullTextSearch(query: string) {
return prisma.article.findMany({
where: {
OR: [
{ title: { search: query } },
{ content: { search: query } },
],
},
orderBy: {
_relevance: {
fields: ['title', 'content'],
search: query,
sort: 'desc',
},
},
});
}
// $queryRaw — 處理 Prisma 不支援的 PG 特性
async function rawArrayQuery(tags: string[]) {
return prisma.$queryRaw<Product[]>`
SELECT * FROM products
WHERE tags && ${tags}::text[]
ORDER BY created_at DESC
`;
}
// 交易（Serializable 隔離等級）
async function transfer(fromId: string, toId: string, amount: number) {
return prisma.$transaction(async (tx) => {
const from = await tx.account.update({
where: { id: fromId },
data: { balance: { decrement: amount } },
});
if (from.balance < 0) throw new Error('餘額不足');
return tx.account.update({
where: { id: toId },
data: { balance: { increment: amount } },
});
}, {
isolationLevel: Prisma.TransactionIsolationLevel.Serializable,
timeout: 10000,
});
}
Prisma 與 PgBouncer 整合
# .env
# 透過 PgBouncer（關閉 prepared statements）
DATABASE_URL="postgresql://user:pass@pgbouncer:6432/db?pgbouncer=true"
# 直連 PostgreSQL（用於 Migrate）
DIRECT_URL="postgresql://user:pass@postgres:5432/db"
SQLAlchemy — Python 生態最成熟的 ORM
Engine 與連線池配置
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from sqlalchemy.pool import QueuePool
engine = create_engine(
"postgresql+psycopg2://user:pass@localhost:5432/mydb",
poolclass=QueuePool,
pool_size=10,
max_overflow=20,
pool_timeout=30,
pool_recycle=1800,
pool_pre_ping=True,
connect_args={
"application_name": "my-python-app",
"options": "-c statement_timeout=10000",
"sslmode": "require",
},
)
SessionLocal = sessionmaker(
bind=engine, autocommit=False, autoflush=False,
expire_on_commit=False,
)
Model 定義與 PG Dialect 型別
from sqlalchemy import Column, String, Numeric, DateTime, Index, Enum as SAEnum
from sqlalchemy.dialects.postgresql import UUID, JSONB, ARRAY, TSVECTOR
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.sql import func
import uuid
from datetime import datetime
from typing import Optional
class Base(DeclarativeBase):
pass
class Product(Base):
__tablename__ = "products"
# UUID 主鍵
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True,
server_default=func.gen_random_uuid()
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
price: Mapped[float] = mapped_column(Numeric(15, 4), nullable=False)
# JSONB
metadata_: Mapped[Optional[dict]] = mapped_column('metadata', JSONB, nullable=True)
# ARRAY
tags: Mapped[Optional[list]] = mapped_column(ARRAY(String), nullable=True)
# PostgreSQL ENUM
status: Mapped[str] = mapped_column(
SAEnum('draft', 'published', 'archived', name='product_status'),
default='draft'
)
# TSVECTOR（全文搜尋向量）
search_vector: Mapped[Optional[str]] = mapped_column(TSVECTOR, nullable=True)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
__table_args__ = (
Index('ix_products_metadata_gin', metadata_, postgresql_using='gin'),
Index('ix_products_search_gin', search_vector, postgresql_using='gin'),
Index('ix_products_tags_gin', tags, postgresql_using='gin'),
)
SQLAlchemy 操作 PostgreSQL 特性
from sqlalchemy import select, func, text
from sqlalchemy.dialects.postgresql import insert
class ProductService:
def __init__(self, db):
self.db = db
# JSONB 查詢
def find_by_metadata(self, key: str, value: str):
return self.db.execute(
select(Product).where(
Product.metadata_[key].astext == value
)
).scalars().all()
# JSONB 包含查詢（@>）
def find_containing(self, criteria: dict):
return self.db.execute(
select(Product).where(
Product.metadata_.contains(criteria)
)
).scalars().all()
# Array 操作
def find_by_tag(self, tag: str):
return self.db.execute(
select(Product).where(Product.tags.contains([tag]))
).scalars().all()
# 全文搜尋
def full_text_search(self, query: str):
ts_query = func.plainto_tsquery('chinese', query)
return self.db.execute(
select(Product)
.where(Product.search_vector.op('@@')(ts_query))
.order_by(func.ts_rank(Product.search_vector, ts_query).desc())
).scalars().all()
# UPSERT（INSERT ... ON CONFLICT）
def upsert(self, data: dict):
stmt = insert(Product).values(**data)
stmt = stmt.on_conflict_do_update(
index_elements=['id'],
set_={'name': stmt.excluded.name, 'price': stmt.excluded.price}
).returning(Product)
return self.db.execute(stmt).fetchone()
Django ORM — PostgreSQL 專屬模組
Django 提供 django.contrib.postgres 專屬模組，對 PostgreSQL 特性支援最為完整：
PostgreSQL 專屬 Field
from django.db import models
from django.contrib.postgres.fields import ArrayField, HStoreField
from django.contrib.postgres.search import SearchVectorField
from django.contrib.postgres.indexes import GinIndex, BrinIndex
class Product(models.Model):
name = models.CharField(max_length=255)
price = models.DecimalField(max_digits=15, decimal_places=4)
# JSONField（自動使用 JSONB on PostgreSQL）
metadata = models.JSONField(null=True, blank=True)
# ArrayField（PostgreSQL 專屬）
tags = ArrayField(
models.CharField(max_length=50),
blank=True, default=list,
)
# HStoreField（需啟用 hstore 擴展）
properties = HStoreField(null=True, blank=True)
# SearchVectorField（全文搜尋向量）
search_vector = SearchVectorField(null=True, editable=False)
status = models.CharField(max_length=20, default='draft')
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
indexes = [
GinIndex(fields=['tags'], name='ix_products_tags_gin'),
GinIndex(fields=['metadata'], name='ix_products_metadata_gin'),
GinIndex(fields=['search_vector'], name='ix_products_search_gin'),
BrinIndex(fields=['created_at'], name='ix_products_created_brin'),
]
Django 全文搜尋（原生支援）
from django.contrib.postgres.search import (
SearchQuery, SearchRank, SearchVector, TrigramSimilarity,
)
from django.db.models import F
class ArticleSearchService:
# 搜尋排名
def ranked_search(self, query: str, limit: int = 20):
search_query = SearchQuery(query, config='chinese')
search_rank = SearchRank(F('search_vector'), search_query)
return (
Article.objects
.filter(search_vector=search_query)
.annotate(rank=search_rank)
.order_by('-rank')[:limit]
)
# 多欄位即時向量搜尋
def multi_field_search(self, query: str):
search_vector = (
SearchVector('title', weight='A', config='english') +
SearchVector('content', weight='B', config='english')
)
search_query = SearchQuery(query, config='english')
return (
Article.objects
.annotate(search=search_vector, rank=SearchRank(search_vector, search_query))
.filter(search=search_query)
.order_by('-rank')
)
# Trigram 相似度搜尋（需 pg_trgm 擴展）
def trigram_search(self, query: str, threshold: float = 0.3):
return (
Article.objects
.annotate(similarity=TrigramSimilarity('title', query))
.filter(similarity__gt=threshold)
.order_by('-similarity')
)
ArrayField 與 JSONField 查詢
# ArrayField 查詢
Product.objects.filter(tags__contains=['python']) # 包含
Product.objects.filter(tags__overlap=['python', 'js']) # 任一重疊
Product.objects.filter(tags__0='python') # 索引存取
# JSONField 查詢
Product.objects.filter(metadata__category='electronics') # 鍵值
Product.objects.filter(metadata__specs__ram__gte=16) # 巢狀路徑
Product.objects.filter(metadata__tags__contains=['urgent']) # JSON 陣列包含
Drizzle ORM — 輕量 SQL 建構器
import { pgTable, pgEnum, varchar, numeric, jsonb, timestamp, uuid } from 'drizzle-orm/pg-core';
import { sql, eq } from 'drizzle-orm';
export const productStatusEnum = pgEnum('product_status', [
'draft', 'published', 'archived',
]);
export const products = pgTable('products', {
id: uuid('id').primaryKey().default(sql`gen_random_uuid()`),
name: varchar('name', { length: 255 }).notNull(),
price: numeric('price', { precision: 15, scale: 4 }).notNull(),
metadata: jsonb('metadata'),
status: productStatusEnum('status').default('draft').notNull(),
createdAt: timestamp('created_at', { withTimezone: true }).defaultNow(),
}, (table) => ({
metadataGin: index('ix_products_metadata_gin').using('gin', table.metadata),
}));
// 查詢
const result = await db.select().from(products).where(eq(products.status, 'published'));
// JSONB（需 sql helper）
const byMeta = await db.select().from(products)
.where(sql`${products.metadata} ->> 'category' = ${'electronics'}`);
// UPSERT
await db.insert(products).values(data)
.onConflictDoUpdate({
target: products.id,
set: { name: sql`EXCLUDED.name`, price: sql`EXCLUDED.price` },
});
N+1 問題與解法
N+1 問題是 ORM 最常見的效能陷阱——查詢 N 筆主記錄後，對每筆記錄各發送 1 次查詢取得關聯資料。
問題識別
# Django N+1（每次迴圈觸發額外查詢）
orders = Order.objects.filter(status='pending') # 1 次查詢
for order in orders:
print(order.customer.name) # N 次查詢
應用程式端解法（Eager Loading）
# Django：select_related（JOIN，適合 FK/OneToOne）
orders = Order.objects.filter(status='pending').select_related('customer')
# Django：prefetch_related（分開查詢 + 合併，適合 M2M/反向 FK）
orders = Order.objects.filter(status='pending').prefetch_related('items', 'items__product')
// TypeORM：relations 選項
const orders = await orderRepo.find({
where: { status: 'pending' },
relations: ['customer', 'items', 'items.product'],
});
# SQLAlchemy：selectinload / joinedload
from sqlalchemy.orm import selectinload, joinedload
orders = session.execute(
select(Order)
.options(
joinedload(Order.customer),
selectinload(Order.items).selectinload(Item.product),
)
.where(Order.status == 'pending')
).scalars().unique().all()
PostgreSQL 端解法（JSON 聚合）
將 N+1 合併為單一查詢，充分利用 PostgreSQL 的 JSON 聚合能力：
SELECT
o.id, o.status,
row_to_json(c.*) AS customer,
COALESCE(
json_agg(
json_build_object('id', i.id, 'product', p.name)
ORDER BY i.id
) FILTER (WHERE i.id IS NOT NULL),
'[]'
) AS items
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.id
LEFT JOIN order_items i ON o.id = i.order_id
LEFT JOIN products p ON i.product_id = p.id
WHERE o.status = 'pending'
GROUP BY o.id, c.id;
連線池配置黃金法則
pool_size ≤ (max_connections - 預留連線) / 應用實例數
範例：max_connections=200, 預留=20, 4 個 Pod
→ 每實例最多 (200-20)/4 = 45 → 設定 pool_size=40
| ORM | 設定位置 | 關鍵參數 |
|---|---|---|
| TypeORM | DataSource.extra | max, min, idleTimeoutMillis |
| Prisma | 連線字串 Query String | connection_limit, pool_timeout |
| SQLAlchemy | create_engine() | pool_size, max_overflow, pool_recycle |
| Django | settings.DATABASES | CONN_MAX_AGE, CONN_HEALTH_CHECKS |
| Drizzle | Pool 設定 | max, min, idleTimeoutMillis |
何時跳出 ORM 使用 Raw SQL？
| 場景 | 原因 |
|---|---|
| WINDOW 函式（RANK、LAG、LEAD） | 多數 ORM 支援不完整 |
| 複雜 JSONB 操作（jsonb_set、jsonb_each） | ORM 語法無法表達完整路徑操作 |
| PostgreSQL 全文搜尋 | Prisma/TypeORM/Drizzle 支援有限 |
| EXCLUDE 約束（防止範圍重疊） | ORM Migration 無法直接生成 |
| 遞迴 CTE（樹狀結構遍歷） | ORM 幾乎都不支援 |
| COPY 命令（大量資料匯入） | 需直接呼叫 pg 驅動的 COPY API |
| LISTEN/NOTIFY（實時通知） | 需在 ORM 外管理持久連線 |
| 分區表 PARTITION 操作 | ORM Migration 通常無法處理 |
總結
選擇 ORM 框架與 PostgreSQL 整合的關鍵要點：
- TypeORM 適合 NestJS 生態，Decorator 語法直覺但進階 PG 特性需 Raw SQL
- Prisma 型別安全最強，Schema-first 開發體驗佳，但 Array/全文搜尋需 Raw
- SQLAlchemy 對 PostgreSQL 特性支援最完整，Core + ORM 雙層 API 靈活度最高
- Django ORM 內建
django.contrib.postgres專屬模組，全文搜尋原生支援最佳 - Drizzle 輕量且接近 SQL 語法，適合偏好「SQL-first」的 TypeScript 開發者
無論選擇哪個框架，都要掌握：正確配置連線池、使用 Eager Loading 避免 N+1、善用 EXPLAIN ANALYZE 檢查生成的 SQL、在 ORM 無法表達時果斷使用 Raw SQL。
下一篇，我們將探討 雲端託管 PostgreSQL——從 AWS RDS、Google Cloud SQL 到 Supabase，掌握各大雲平台的 PostgreSQL 服務差異與最佳配置。