與應用程式整合

資料庫最終要與應用程式協作。本篇介紹 PostgreSQL 與常用語言的整合方式與最佳實踐。

---

一、 Node.js / TypeScript

pg 套件（原生）

npm install pg

import { Pool, Client } from "pg";

// 連線池（推薦用於 Web 應用）
const pool = new Pool({
  host: "localhost",
  port: 5432,
  database: "mydb",
  user: "admin",
  password: "secret",
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
});

// 查詢
async function getUsers() {
  const result = await pool.query("SELECT * FROM users WHERE active = $1", [
    true,
  ]);
  return result.rows;
}

// 交易
async function transfer(fromId: number, toId: number, amount: number) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    await client.query(
      "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
      [amount, fromId]
    );
    await client.query(
      "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
      [amount, toId]
    );
    await client.query("COMMIT");
  } catch (e) {
    await client.query("ROLLBACK");
    throw e;
  } finally {
    client.release();
  }
}

Prisma ORM

npm install prisma @prisma/client
npx prisma init

// schema.prisma
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  name      String?
  posts     Post[]
  createdAt DateTime @default(now())
}

model Post {
  id        Int      @id @default(autoincrement())
  title     String
  content   String?
  published Boolean  @default(false)
  author    User     @relation(fields: [authorId], references: [id])
  authorId  Int
}

import { PrismaClient } from "@prisma/client";

const prisma = new PrismaClient();

// 查詢
const users = await prisma.user.findMany({
  where: { name: { contains: "Alice" } },
  include: { posts: true },
});

// 建立
const user = await prisma.user.create({
  data: {
    email: "alice@example.com",
    name: "Alice",
    posts: {
      create: { title: "First Post" },
    },
  },
});

// 交易
await prisma.$transaction([
  prisma.account.update({
    where: { id: 1 },
    data: { balance: { decrement: 100 } },
  }),
  prisma.account.update({
    where: { id: 2 },
    data: { balance: { increment: 100 } },
  }),
]);

Drizzle ORM

npm install drizzle-orm postgres
npm install -D drizzle-kit

// schema.ts
import {
  pgTable,
  serial,
  text,
  timestamp,
  boolean,
  integer,
} from "drizzle-orm/pg-core";

export const users = pgTable("users", {
  id: serial("id").primaryKey(),
  email: text("email").notNull().unique(),
  name: text("name"),
  createdAt: timestamp("created_at").defaultNow(),
});

export const posts = pgTable("posts", {
  id: serial("id").primaryKey(),
  title: text("title").notNull(),
  authorId: integer("author_id").references(() => users.id),
});

// 使用
import { drizzle } from "drizzle-orm/postgres-js";
import postgres from "postgres";

const sql = postgres(process.env.DATABASE_URL!);
const db = drizzle(sql);

const result = await db.select().from(users).where(eq(users.name, "Alice"));

---

二、 Python

psycopg2 / psycopg3

pip install psycopg2-binary  # 或 psycopg

import psycopg2
from psycopg2.extras import RealDictCursor

# 連線
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="admin",
    password="secret"
)

# 查詢
with conn.cursor(cursor_factory=RealDictCursor) as cur:
    cur.execute("SELECT * FROM users WHERE active = %s", (True,))
    users = cur.fetchall()

# 交易
try:
    with conn.cursor() as cur:
        cur.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (100, 1))
        cur.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (100, 2))
    conn.commit()
except Exception as e:
    conn.rollback()
    raise

conn.close()

SQLAlchemy

pip install sqlalchemy psycopg2-binary

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    name = Column(String)
    active = Column(Boolean, default=True)

# 連線
engine = create_engine('postgresql://admin:secret@localhost/mydb')
Session = sessionmaker(bind=engine)
session = Session()

# 查詢
users = session.query(User).filter(User.active == True).all()

# 建立
user = User(email='alice@example.com', name='Alice')
session.add(user)
session.commit()

---

三、 連線池最佳實踐

Node.js

const pool = new Pool({
  max: 20, // 最大連線數
  idleTimeoutMillis: 30000, // 閒置超時
  connectionTimeoutMillis: 2000, // 連線超時
});

// 優雅關閉
process.on("SIGTERM", async () => {
  await pool.end();
  process.exit(0);
});

使用 PgBouncer

# pgbouncer.ini
[databases]
mydb = host=localhost port=5432

[pgbouncer]
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 20

// 應用程式連接 PgBouncer
const pool = new Pool({
  host: "pgbouncer-host",
  port: 6432,
  database: "mydb",
});

---

四、 ORM vs 原生 SQL

特性	ORM	原生 SQL
開發速度	快	慢
學習曲線	需要學習 ORM	需要精通 SQL
效能	可能較差	可精確控制
複雜查詢	困難	靈活
可移植性	通常可跨 DB	特定 DB

建議

• 簡單 CRUD：用 ORM
• 複雜報表：用原生 SQL 或 Query Builder
• 效能關鍵：用原生 SQL

// Prisma 也可以執行原生 SQL
const result = await prisma.$queryRaw`
  SELECT 
    u.name,
    COUNT(p.id) as post_count
  FROM users u
  LEFT JOIN posts p ON u.id = p.author_id
  GROUP BY u.id
  HAVING COUNT(p.id) > 10
`;

---

五、 Migration 工具

Prisma Migrate

# 建立 migration
npx prisma migrate dev --name add_email_field

# 部署到生產
npx prisma migrate deploy

Drizzle Kit

# 產生 migration
npx drizzle-kit generate:pg

# 執行 migration
npx drizzle-kit push:pg

Flyway

# V1__create_users.sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) NOT NULL UNIQUE,
    created_at TIMESTAMP DEFAULT NOW()
);

# 執行
flyway migrate

Alembic (Python)

# 初始化
alembic init alembic

# 產生 migration
alembic revision --autogenerate -m "add users table"

# 執行
alembic upgrade head

---

六、 型別安全

Prisma

自動根據 schema 產生 TypeScript 型別。

Kysely (TypeScript Query Builder)

import { Kysely, PostgresDialect } from "kysely";

interface Database {
  users: {
    id: number;
    email: string;
    name: string | null;
  };
}

const db = new Kysely<Database>({
  dialect: new PostgresDialect({ pool }),
});

// 完全型別安全
const user = await db
  .selectFrom("users")
  .select(["id", "email"])
  .where("id", "=", 1)
  .executeTakeFirst();

zapatos

import * as db from "zapatos/db";

// 從資料庫 schema 自動產生型別
const users = await db.select("users", { active: true }).run(pool);

---

七、 效能技巧

1. 批量插入

// 不好：逐筆插入
for (const user of users) {
  await pool.query("INSERT INTO users (name) VALUES ($1)", [user.name]);
}

// 好：批量插入
const values = users.map((u, i) => `($${i + 1})`).join(",");
const params = users.map((u) => u.name);
await pool.query(`INSERT INTO users (name) VALUES ${values}`, params);

2. Prepared Statements

// pg 套件自動使用 prepared statements
// 相同查詢只會解析一次

3. 串流大量資料

import QueryStream from "pg-query-stream";

const query = new QueryStream("SELECT * FROM large_table");
const stream = client.query(query);

stream.on("data", (row) => {
  // 逐列處理，不會爆記憶體
});

---

八、 錯誤處理

import { DatabaseError } from "pg";

try {
  await pool.query("INSERT INTO users (email) VALUES ($1)", [
    "alice@example.com",
  ]);
} catch (e) {
  if (e instanceof DatabaseError) {
    switch (e.code) {
      case "23505": // unique_violation
        throw new Error("Email already exists");
      case "23503": // foreign_key_violation
        throw new Error("Referenced record not found");
      case "23514": // check_violation
        throw new Error("Validation failed");
      default:
        throw e;
    }
  }
  throw e;
}

---

九、 測試資料庫

使用 Docker

# docker-compose.test.yml
services:
  test-db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: test
    ports:
      - "5433:5432"

使用測試容器

import { PostgreSqlContainer } from "@testcontainers/postgresql";

describe("Database Tests", () => {
  let container: PostgreSqlContainer;
  let pool: Pool;

  beforeAll(async () => {
    container = await new PostgreSqlContainer().start();
    pool = new Pool({ connectionString: container.getConnectionUri() });
  });

  afterAll(async () => {
    await pool.end();
    await container.stop();
  });
});

---

總結

語言	推薦工具
Node.js	pg + Prisma/Drizzle
Python	psycopg2 + SQLAlchemy

最佳實踐：

1. 使用連線池
2. 適當使用 ORM 和原生 SQL
3. 使用 Migration 管理 schema
4. 處理資料庫錯誤
5. 型別安全

---

進階挑戰

1. 使用 Prisma + TypeScript 建立一個 CRUD API，並加入交易處理
2. 比較原生 pg 模組和 Prisma ORM 在批量插入場景的效能差異
3. 設計一個 Database Migration 流程，支援 rollback

---

延伸閱讀與資源

• node-postgres (pg) 官方文件
• Prisma 官方文件
• psycopg 官方文件

---

← 上一章：常用擴充套件 | 返回專題首頁