数据库集成 - wangshengliang

🙋 全栈应用需要数据库。如何在 Next.js 中高效、安全地进行数据库操作？

ORM 选择#

特性	Prisma	Drizzle
类型安全	✅	✅
Schema 定义	.prisma 文件	TypeScript
学习曲线	中等	较低
查询风格	链式 API	SQL-like
Bundle 大小	较大	较小
Edge 支持	需要适配器	原生支持

Prisma 集成#

安装与初始化#

# 安装 Prisma
pnpm add prisma @prisma/client
pnpm add -D prisma

# 初始化
npx prisma init

Schema 定义#

generator client {
  provider = "prisma-client-js"
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

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

model Post {
  id        String   @id @default(cuid())
  title     String
  content   String?
  published Boolean  @default(false)
  author    User     @relation(fields: [authorId], references: [id])
  authorId  String
  tags      Tag[]
  createdAt DateTime @default(now())
  updatedAt DateTime @updatedAt

  @@index([authorId])
}

model Tag {
  id    String @id @default(cuid())
  name  String @unique
  posts Post[]
}

客户端实例#

import { PrismaClient } from '@prisma/client'

const globalForPrisma = globalThis as unknown as {
  prisma: PrismaClient | undefined
}

export const prisma =
  globalForPrisma.prisma ??
  new PrismaClient({
    log:
      process.env.NODE_ENV === 'development'
        ? ['query', 'error', 'warn']
        : ['error'],
  })

if (process.env.NODE_ENV !== 'production') {
  globalForPrisma.prisma = prisma
}

🔶 重要：在开发环境使用全局变量避免热重载创建多个连接。

数据库迁移#

# 创建迁移
npx prisma migrate dev --name init

# 生产环境部署
npx prisma migrate deploy

# 重置数据库（开发用）
npx prisma migrate reset

# 生成客户端
npx prisma generate

基础 CRUD 操作#

'use server'

import { prisma } from '@/lib/prisma'
import { revalidatePath } from 'next/cache'

// 创建
export async function createUser(data: {
  email: string
  name: string
  password: string
}) {
  const user = await prisma.user.create({
    data: {
      email: data.email,
      name: data.name,
      password: data.password, // 实际应该哈希处理
    },
  })

  revalidatePath('/users')
  return user
}

// 查询
export async function getUsers() {
  return prisma.user.findMany({
    select: {
      id: true,
      email: true,
      name: true,
      createdAt: true,
      _count: {
        select: { posts: true },
      },
    },
    orderBy: { createdAt: 'desc' },
  })
}

// 查询单个
export async function getUser(id: string) {
  return prisma.user.findUnique({
    where: { id },
    include: {
      posts: {
        where: { published: true },
        orderBy: { createdAt: 'desc' },
      },
    },
  })
}

// 更新
export async function updateUser(
  id: string,
  data: { name?: string; email?: string }
) {
  const user = await prisma.user.update({
    where: { id },
    data,
  })

  revalidatePath('/users')
  revalidatePath(`/users/${id}`)
  return user
}

// 删除
export async function deleteUser(id: string) {
  await prisma.user.delete({
    where: { id },
  })

  revalidatePath('/users')
}

关联查询#

// 获取文章及作者
export async function getPostWithAuthor(id: string) {
  return prisma.post.findUnique({
    where: { id },
    include: {
      author: {
        select: { id: true, name: true, email: true },
      },
      tags: true,
    },
  })
}

// 获取用户的所有文章
export async function getUserPosts(userId: string) {
  return prisma.post.findMany({
    where: { authorId: userId },
    include: { tags: true },
    orderBy: { createdAt: 'desc' },
  })
}

// 创建文章并关联标签
export async function createPost(data: {
  title: string
  content: string
  authorId: string
  tags: string[]
}) {
  return prisma.post.create({
    data: {
      title: data.title,
      content: data.content,
      authorId: data.authorId,
      tags: {
        connectOrCreate: data.tags.map((name) => ({
          where: { name },
          create: { name },
        })),
      },
    },
    include: { tags: true },
  })
}

事务处理#

// 转账示例
export async function transfer(fromId: string, toId: string, amount: number) {
  return prisma.$transaction(async (tx) => {
    // 扣除发送方余额
    const sender = await tx.user.update({
      where: { id: fromId },
      data: { balance: { decrement: amount } },
    })

    if (sender.balance < 0) {
      throw new Error('余额不足')
    }

    // 增加接收方余额
    await tx.user.update({
      where: { id: toId },
      data: { balance: { increment: amount } },
    })

    // 记录交易
    await tx.transaction.create({
      data: { fromId, toId, amount },
    })

    return { success: true }
  })
}

Drizzle 集成#

安装#

# PostgreSQL
pnpm add drizzle-orm postgres
pnpm add -D drizzle-kit

# 或 MySQL
pnpm add drizzle-orm mysql2
pnpm add -D drizzle-kit

Schema 定义#

import { pgTable, text, timestamp, boolean, uuid } from 'drizzle-orm/pg-core'
import { relations } from 'drizzle-orm'

export const users = pgTable('users', {
  id: uuid('id').defaultRandom().primaryKey(),
  email: text('email').notNull().unique(),
  name: text('name'),
  password: text('password').notNull(),
  createdAt: timestamp('created_at').defaultNow().notNull(),
  updatedAt: timestamp('updated_at').defaultNow().notNull(),
})

export const posts = pgTable('posts', {
  id: uuid('id').defaultRandom().primaryKey(),
  title: text('title').notNull(),
  content: text('content'),
  published: boolean('published').default(false).notNull(),
  authorId: uuid('author_id')
    .notNull()
    .references(() => users.id),
  createdAt: timestamp('created_at').defaultNow().notNull(),
  updatedAt: timestamp('updated_at').defaultNow().notNull(),
})

export const tags = pgTable('tags', {
  id: uuid('id').defaultRandom().primaryKey(),
  name: text('name').notNull().unique(),
})

export const postsToTags = pgTable('posts_to_tags', {
  postId: uuid('post_id')
    .notNull()
    .references(() => posts.id),
  tagId: uuid('tag_id')
    .notNull()
    .references(() => tags.id),
})

// 定义关系
export const usersRelations = relations(users, ({ many }) => ({
  posts: many(posts),
}))

export const postsRelations = relations(posts, ({ one, many }) => ({
  author: one(users, {
    fields: [posts.authorId],
    references: [users.id],
  }),
  tags: many(postsToTags),
}))

客户端实例#

import { drizzle } from 'drizzle-orm/postgres-js'
import postgres from 'postgres'
import * as schema from './schema'

const connectionString = process.env.DATABASE_URL!

// 连接池配置
const client = postgres(connectionString, {
  max: 10,
  idle_timeout: 20,
  connect_timeout: 10,
})

export const db = drizzle(client, { schema })

配置文件#

import type { Config } from 'drizzle-kit'

export default {
  schema: './db/schema.ts',
  out: './drizzle',
  dialect: 'postgresql',
  dbCredentials: {
    url: process.env.DATABASE_URL!,
  },
} satisfies Config

迁移命令#

# 生成迁移
npx drizzle-kit generate

# 执行迁移
npx drizzle-kit migrate

# 推送 schema（开发用）
npx drizzle-kit push

# 打开 Studio
npx drizzle-kit studio

基础 CRUD 操作#

'use server'

import { db } from '@/db'
import { users, posts } from '@/db/schema'
import { eq, desc } from 'drizzle-orm'
import { revalidatePath } from 'next/cache'

// 创建
export async function createUser(data: {
  email: string
  name: string
  password: string
}) {
  const [user] = await db
    .insert(users)
    .values({
      email: data.email,
      name: data.name,
      password: data.password,
    })
    .returning()

  revalidatePath('/users')
  return user
}

// 查询所有
export async function getUsers() {
  return db.select().from(users).orderBy(desc(users.createdAt))
}

// 查询单个
export async function getUser(id: string) {
  const [user] = await db.select().from(users).where(eq(users.id, id))
  return user
}

// 更新
export async function updateUser(
  id: string,
  data: { name?: string; email?: string }
) {
  const [user] = await db
    .update(users)
    .set({ ...data, updatedAt: new Date() })
    .where(eq(users.id, id))
    .returning()

  revalidatePath('/users')
  return user
}

// 删除
export async function deleteUser(id: string) {
  await db.delete(users).where(eq(users.id, id))
  revalidatePath('/users')
}

关联查询#

// 使用 with 进行关联查询
export async function getUserWithPosts(id: string) {
  const result = await db.query.users.findFirst({
    where: eq(users.id, id),
    with: {
      posts: {
        orderBy: desc(posts.createdAt),
      },
    },
  })

  return result
}

// 手动 JOIN
export async function getPostsWithAuthors() {
  return db
    .select({
      post: posts,
      author: {
        id: users.id,
        name: users.name,
        email: users.email,
      },
    })
    .from(posts)
    .leftJoin(users, eq(posts.authorId, users.id))
    .orderBy(desc(posts.createdAt))
}

事务处理#

export async function createPostWithTags(data: {
  title: string
  content: string
  authorId: string
  tagNames: string[]
}) {
  return db.transaction(async (tx) => {
    // 创建文章
    const [post] = await tx
      .insert(posts)
      .values({
        title: data.title,
        content: data.content,
        authorId: data.authorId,
      })
      .returning()

    // 创建或获取标签
    for (const name of data.tagNames) {
      const [tag] = await tx
        .insert(tags)
        .values({ name })
        .onConflictDoNothing()
        .returning()

      const existingTag =
        tag ||
        (await tx.query.tags.findFirst({
          where: eq(tags.name, name),
        }))

      if (existingTag) {
        await tx.insert(postsToTags).values({
          postId: post.id,
          tagId: existingTag.id,
        })
      }
    }

    return post
  })
}

连接池管理#

Serverless 环境#

// lib/db.ts (Prisma with connection pooling)
import { PrismaClient } from '@prisma/client'
import { Pool } from '@neondatabase/serverless'
import { PrismaNeon } from '@prisma/adapter-neon'

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
const adapter = new PrismaNeon(pool)

export const prisma = new PrismaClient({ adapter })

使用外部连接池#

# 直接连接（用于迁移）
DATABASE_URL="postgresql://user:password@host:5432/db"

# 连接池（用于应用）
DATABASE_URL_POOLED="postgresql://user:password@pooler.host:6543/db?pgbouncer=true"

// 开发环境使用直接连接，生产使用连接池
const connectionString =
  process.env.NODE_ENV === 'production'
    ? process.env.DATABASE_URL_POOLED
    : process.env.DATABASE_URL

在 Server Components 中使用#

// Next.js 15.x
import { prisma } from '@/lib/prisma'
import Link from 'next/link'

export default async function UsersPage() {
  const users = await prisma.user.findMany({
    select: {
      id: true,
      name: true,
      email: true,
      _count: { select: { posts: true } },
    },
    orderBy: { createdAt: 'desc' },
  })

  return (
    <div>
      <h1 className="text-2xl font-bold mb-4">用户列表</h1>
      <ul className="space-y-2">
        {users.map((user) => (
          <li key={user.id} className="p-4 border rounded">
            <Link href={`/users/${user.id}`}>
              <h2 className="font-medium">{user.name || '未命名'}</h2>
              <p className="text-gray-500">{user.email}</p>
              <p className="text-sm">{user._count.posts} 篇文章</p>
            </Link>
          </li>
        ))}
      </ul>
    </div>
  )
}

数据验证#

使用 Zod#

import { z } from 'zod'

export const createUserSchema = z.object({
  email: z.string().email('无效的邮箱格式'),
  name: z.string().min(2, '名称至少 2 个字符').max(50),
  password: z.string().min(8, '密码至少 8 个字符'),
})

export const updateUserSchema = createUserSchema.partial()

export type CreateUserInput = z.infer<typeof createUserSchema>
export type UpdateUserInput = z.infer<typeof updateUserSchema>

'use server'

import { prisma } from '@/lib/prisma'
import { createUserSchema, updateUserSchema } from '@/lib/validations/user'
import { hash } from 'bcryptjs'

export async function createUser(formData: FormData) {
  const rawData = {
    email: formData.get('email'),
    name: formData.get('name'),
    password: formData.get('password'),
  }

  const result = createUserSchema.safeParse(rawData)

  if (!result.success) {
    return { errors: result.error.flatten().fieldErrors }
  }

  const hashedPassword = await hash(result.data.password, 12)

  try {
    const user = await prisma.user.create({
      data: {
        ...result.data,
        password: hashedPassword,
      },
    })

    return { user }
  } catch (error) {
    if ((error as any).code === 'P2002') {
      return { errors: { email: ['邮箱已被注册'] } }
    }
    throw error
  }
}

性能优化#

选择性查询#

// ✅ 只查询需要的字段
const users = await prisma.user.findMany({
  select: {
    id: true,
    name: true,
    // 不查询 password 等敏感字段
  },
})

// ❌ 查询所有字段
const users = await prisma.user.findMany()

分页查询#

export async function getPaginatedPosts(page: number, limit: number = 10) {
  const [posts, total] = await Promise.all([
    prisma.post.findMany({
      skip: (page - 1) * limit,
      take: limit,
      orderBy: { createdAt: 'desc' },
    }),
    prisma.post.count(),
  ])

  return {
    posts,
    pagination: {
      page,
      limit,
      total,
      pages: Math.ceil(total / limit),
    },
  }
}

缓存数据库查询#

import { cache } from 'react'

// 使用 React cache 去重
export const getUser = cache(async (id: string) => {
  return prisma.user.findUnique({ where: { id } })
})

// 使用 unstable_cache 持久化
import { unstable_cache } from 'next/cache'

export const getCachedUser = unstable_cache(
  async (id: string) => {
    return prisma.user.findUnique({ where: { id } })
  },
  ['user'],
  { revalidate: 60, tags: ['users'] }
)

常见问题#

🤔 Q: Prisma 和 Drizzle 怎么选？

Prisma：功能全面，生态成熟，适合快速开发
Drizzle：轻量高效，SQL-like 语法，适合追求性能

🤔 Q: 如何处理数据库连接超时？

配置连接池参数：

// Prisma
new PrismaClient({
  datasources: {
    db: {
      url: process.env.DATABASE_URL,
    },
  },
})

// Drizzle with postgres.js
postgres(url, {
  max: 10,
  idle_timeout: 20,
  connect_timeout: 10,
})

🤔 Q: Edge Runtime 如何使用数据库？

使用支持 Edge 的数据库驱动，如 Neon、PlanetScale 或 Turso。

下一篇将介绍错误处理，学习如何优雅地处理应用中的各种错误。

-EOF-