kysely

Kysely — 类型安全的 SQL 查询构建器

概述

Kysely 是一个类型安全的 TypeScript SQL 查询构建器，可实现从数据库架构到查询结果的端到端类型安全。与 ORM 不同，它生成纯 SQL，让你在保持完美 TypeScript 类型推断的同时，全面掌控查询。

主要功能：

• 完整的类型推断（架构 → 查询 → 结果）
• 零运行时开销（编译为 SQL）
• 与数据库无关（PostgreSQL、MySQL、SQLite、MSSQL）
• 内置迁移系统
• 插件生态系统（CTE、JSON、地理空间）
• 按需集成原始 SQL

安装：

npm install kysely
# Database driver (choose one)
npm install pg              # PostgreSQL
npm install mysql2          # MySQL
npm install better-sqlite3  # SQLite

快速开始

1. 定义数据库架构类型

import { Generated, Selectable, Insertable, Updateable } from 'kysely';

// Table interface (all columns)
interface UserTable {
  id: Generated<number>;
  email: string;
  name: string | null;
  created_at: Generated<Date>;
  updated_at: Date;
}

interface PostTable {
  id: Generated<number>;
  user_id: number;
  title: string;
  content: string;
  published: Generated<boolean>;
  created_at: Generated<Date>;
}

// Database interface
interface Database {
  users: UserTable;
  posts: PostTable;
}

// Type-safe query result types
type User = Selectable<UserTable>;
type NewUser = Insertable<UserTable>;
type UserUpdate = Updateable<UserTable>;

2. 创建数据库实例

import { Kysely, PostgresDialect } from 'kysely';
import { Pool } from 'pg';

const db = new Kysely<Database>({
  dialect: new PostgresDialect({
    pool: new Pool({
      host: process.env.DB_HOST,
      database: process.env.DB_NAME,
      user: process.env.DB_USER,
      password: process.env.DB_PASSWORD,
      max: 10,
    }),
  }),
});

3. 类型安全的查询

// SELECT with full type inference
const users = await db
  .selectFrom('users')
  .select(['id', 'email', 'name'])
  .where('created_at', '>', new Date('2024-01-01'))
  .execute();
// Type: Array<{ id: number; email: string; name: string | null }>

// INSERT with type checking
const newUser: NewUser = {
  email: '[email protected]',
  name: 'Alice',
  updated_at: new Date(),
};

const inserted = await db
  .insertInto('users')
  .values(newUser)
  .returningAll()
  .executeTakeFirstOrThrow();
// Type: User

// UPDATE
await db
  .updateTable('users')
  .set({ name: 'Alice Updated', updated_at: new Date() })
  .where('id', '=', 1)
  .execute();

// DELETE
await db
  .deleteFrom('users')
  .where('email', 'like', '%@spam.com')
  .execute();

高级查询模式

类型安全的联接

// INNER JOIN
const usersWithPosts = await db
  .selectFrom('users')
  .innerJoin('posts', 'posts.user_id', 'users.id')
  .select([
    'users.id',
    'users.name',
    'posts.title',
    'posts.content',
  ])
  .execute();
// Type: Array<{ id: number; name: string | null; title: string; content: string }>

// LEFT JOIN with null handling
const usersWithOptionalPosts = await db
  .selectFrom('users')
  .leftJoin('posts', 'posts.user_id', 'users.id')
  .select([
    'users.id',
    'users.email',
    'posts.title',  // Type: string | null (from LEFT JOIN)
  ])
  .execute();

// Multiple joins
const complexQuery = await db
  .selectFrom('posts')
  .innerJoin('users', 'users.id', 'posts.user_id')
  .leftJoin('comments', 'comments.post_id', 'posts.id')
  .select([
    'posts.id as postId',
    'posts.title',
    'users.name as authorName',
    'comments.id as commentId',
  ])
  .execute();

聚合与分组

import { sql } from 'kysely';

// COUNT, AVG, SUM
const stats = await db
  .selectFrom('posts')
  .select([
    'user_id',
    db.fn.count<number>('id').as('post_count'),
    db.fn.avg<number>('views').as('avg_views'),
  ])
  .groupBy('user_id')
  .having(db.fn.count('id'), '>', 5)
  .execute();
// Type: Array<{ user_id: number; post_count: number; avg_views: number }>

// Complex aggregations with raw SQL
const advanced = await db
  .selectFrom('users')
  .select([
    'users.id',
    sql<number>`COUNT(DISTINCT posts.id)`.as('total_posts'),
    sql<Date>`MAX(posts.created_at)`.as('latest_post'),
  ])
  .leftJoin('posts', 'posts.user_id', 'users.id')
  .groupBy('users.id')
  .execute();

子查询

// Scalar subquery
const usersWithPostCount = await db
  .selectFrom('users')
  .select([
    'users.id',
    'users.name',
    (eb) =>
      eb
        .selectFrom('posts')
        .select(eb.fn.count<number>('id').as('count'))
        .whereRef('posts.user_id', '=', 'users.id')
        .as('post_count'),
  ])
  .execute();

// EXISTS subquery
const activeUsers = await db
  .selectFrom('users')
  .selectAll()
  .where((eb) =>
    eb.exists(
      eb
        .selectFrom('posts')
        .select('id')
        .whereRef('posts.user_id', '=', 'users.id')
        .where('created_at', '>', new Date('2024-01-01'))
    )
  )
  .execute();

// IN subquery
const usersInTopTier = await db
  .selectFrom('users')
  .selectAll()
  .where(
    'id',
    'in',
    db.selectFrom('posts')
      .select('user_id')
      .groupBy('user_id')
      .having(db.fn.count('id'), '>', 100)
  )
  .execute();

公用表表达式（CTE）

// WITH clause
const result = await db
  .with('popular_posts', (db) =>
    db
      .selectFrom('posts')
      .select(['id', 'user_id', 'title'])
      .where('views', '>', 1000)
  )
  .with('active_users', (db) =>
    db
      .selectFrom('users')
      .select(['id', 'email'])
      .where('last_login', '>', new Date('2024-01-01'))
  )
  .selectFrom('popular_posts')
  .innerJoin('active_users', 'active_users.id', 'popular_posts.user_id')
  .selectAll()
  .execute();

// Recursive CTE (organizational hierarchy)
interface OrgNode {
  id: number;
  name: string;
  parent_id: number | null;
  level: number;
}

const hierarchy = await db
  .withRecursive('org_tree', (db) =>
    db
      .selectFrom('departments')
      .select(['id', 'name', 'parent_id', sql<number>`0`.as('level')])
      .where('parent_id', 'is', null)
      .unionAll(
        db
          .selectFrom('departments')
          .innerJoin('org_tree', 'org_tree.id', 'departments.parent_id')
          .select([
            'departments.id',
            'departments.name',
            'departments.parent_id',
            sql<number>`org_tree.level + 1`.as('level'),
          ])
      )
  )
  .selectFrom('org_tree')
  .selectAll()
  .execute();

从数据库生成架构

使用 kysely-codegen

# Install
npm install --save-dev kysely-codegen

# Generate types from existing database
npx kysely-codegen --url "postgresql://user:pass@localhost:5432/mydb"

生成的输出：

// Generated by kysely-codegen
import type { ColumnType, Generated } from 'kysely';

export interface Database {
  users: UsersTable;
  posts: PostsTable;
  comments: CommentsTable;
}

export interface UsersTable {
  id: Generated<number>;
  email: string;
  name: string | null;
  created_at: Generated<Date>;
}

export interface PostsTable {
  id: Generated<number>;
  user_id: number;
  title: string;
  content: string;
  published: Generated<boolean>;
  created_at: Generated<Date>;
}

自定义类型映射

// Map database types to TypeScript types
interface CustomTypes {
  timestamp: Date;
  jsonb: unknown;
  numeric: string; // Preserve precision
  uuid: string;
}

interface ProductTable {
  id: ColumnType<string, string | undefined, string>; // SELECT, INSERT, UPDATE types
  metadata: ColumnType<Record<string, unknown>, string, string>; // JSON column
  price: ColumnType<number, number, number | undefined>; // Numeric
}

迁移

迁移设置

import { Kysely, Migrator, FileMigrationProvider } from 'kysely';
import { promises as fs } from 'fs';
import * as path from 'path';

const migrator = new Migrator({
  db,
  provider: new FileMigrationProvider({
    fs,
    path,
    migrationFolder: path.join(__dirname, 'migrations'),
  }),
});

// Run all pending migrations
async function migrateToLatest() {
  const { error, results } = await migrator.migrateToLatest();

  results?.forEach((it) => {
    if (it.status === 'Success') {
      console.log(`Migration "${it.migrationName}" executed successfully`);
    } else if (it.status === 'Error') {
      console.error(`Migration "${it.migrationName}" failed`);
    }
  });

  if (error) {
    console.error('Migration failed:', error);
    process.exit(1);
  }
}

// Rollback last migration
async function migrateDown() {
  const { error, results } = await migrator.migrateDown();
  // Handle results...
}

迁移文件

// migrations/001_create_users.ts
import { Kysely, sql } from 'kysely';

export async function up(db: Kysely<any>): Promise<void> {
  await db.schema
    .createTable('users')
    .addColumn('id', 'serial', (col) => col.primaryKey())
    .addColumn('email', 'varchar(255)', (col) => col.notNull().unique())
    .addColumn('name', 'varchar(255)')
    .addColumn('created_at', 'timestamp', (col) =>
      col.defaultTo(sql`CURRENT_TIMESTAMP`).notNull()
    )
    .execute();

  await db.schema
    .createIndex('users_email_idx')
    .on('users')
    .column('email')
    .execute();
}

export async function down(db: Kysely<any>): Promise<void> {
  await db.schema.dropTable('users').execute();
}

复杂迁移示例

// Add foreign key
export async function up(db: Kysely<any>): Promise<void> {
  await db.schema
    .createTable('posts')
    .addColumn('id', 'serial', (col) => col.primaryKey())
    .addColumn('user_id', 'integer', (col) =>
      col.references('users.id').onDelete('cascade').notNull()
    )
    .addColumn('title', 'varchar(500)', (col) => col.notNull())
    .addColumn('content', 'text')
    .execute();
}

// Alter table
export async function up(db: Kysely<any>): Promise<void> {
  await db.schema
    .alterTable('users')
    .addColumn('bio', 'text')
    .execute();

  await db.schema
    .alterTable('users')
    .modifyColumn('email', 'varchar(320)')
    .execute();
}

// Add enum column (PostgreSQL)
export async function up(db: Kysely<any>): Promise<void> {
  await sql`CREATE TYPE user_role AS ENUM ('admin', 'user', 'guest')`.execute(db);

  await db.schema
    .alterTable('users')
    .addColumn('role', sql`user_role`, (col) => col.defaultTo('user'))
    .execute();
}

事务

基本事务

// Automatic rollback on error
await db.transaction().execute(async (trx) => {
  await trx
    .insertInto('users')
    .values({ email: '[email protected]', name: 'Alice', updated_at: new Date() })
    .execute();

  await trx
    .insertInto('posts')
    .values({ user_id: 1, title: 'First Post', content: 'Hello' })
    .execute();
});

// Manual transaction control
const trx = await db.transaction().execute(async (trx) => {
  const user = await trx
    .insertInto('users')
    .values({ email: '[email protected]', name: 'Bob', updated_at: new Date() })
    .returningAll()
    .executeTakeFirstOrThrow();

  const post = await trx
    .insertInto('posts')
    .values({
      user_id: user.id,
      title: 'Bob\'s Post',
      content: 'Content',
    })
    .returningAll()
    .executeTakeFirstOrThrow();

  return { user, post };
});

隔离级别

import { IsolationLevel } from 'kysely';

// Read committed (default)
await db.transaction()
  .setIsolationLevel('read committed')
  .execute(async (trx) => {
    // Transaction logic
  });

// Serializable (strongest isolation)
await db.transaction()
  .setIsolationLevel('serializable')
  .execute(async (trx) => {
    const balance = await trx
      .selectFrom('accounts')
      .select('balance')
      .where('id', '=', accountId)
      .executeTakeFirstOrThrow();

    await trx
      .updateTable('accounts')
      .set({ balance: balance.balance - amount })
      .where('id', '=', accountId)
      .execute();
  });

原始 SQL 集成

使用 sql 模板标签

import { sql } from 'kysely';

// Raw SQL in SELECT
const result = await db
  .selectFrom('users')
  .select([
    'id',
    sql<string>`UPPER(name)`.as('uppercase_name'),
    sql<number>`EXTRACT(YEAR FROM created_at)`.as('year_created'),
  ])
  .execute();

// Raw SQL in WHERE
const filtered = await db
  .selectFrom('posts')
  .selectAll()
  .where(sql`LOWER(title)`, 'like', '%typescript%')
  .execute();

// Complex raw queries
const custom = await sql<{ total: number; avg_age: number }>`
  SELECT
    COUNT(*) as total,
    AVG(EXTRACT(YEAR FROM age(birth_date))) as avg_age
  FROM users
  WHERE active = true
`.execute(db);

完整原始查询

// Execute arbitrary SQL
const result = await sql`
  WITH ranked_posts AS (
    SELECT
      p.*,
      ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY views DESC) as rank
    FROM posts p
  )
  SELECT * FROM ranked_posts WHERE rank <= 3
`.execute(db);

// Parameterized raw queries
const email = '[email protected]';
const user = await sql<User>`
  SELECT * FROM users WHERE email = ${email}
`.execute(db);

插件生态系统

JSON 操作（PostgreSQL）

import { jsonBuildObject, jsonArrayFrom } from 'kysely/helpers/postgres';

// Build JSON objects
const usersWithPosts = await db
  .selectFrom('users')
  .select([
    'users.id',
    'users.name',
    jsonArrayFrom(
      db
        .selectFrom('posts')
        .select(['posts.id', 'posts.title', 'posts.content'])
        .whereRef('posts.user_id', '=', 'users.id')
    ).as('posts'),
  ])
  .execute();
// Result: { id: 1, name: "Alice", posts: [{ id: 1, title: "..." }] }

// JSON aggregation
const nested = await db
  .selectFrom('users')
  .select([
    'users.id',
    jsonBuildObject({
      name: 'users.name',
      email: 'users.email',
      postCount: sql<number>`(SELECT COUNT(*) FROM posts WHERE user_id = users.id)`,
    }).as('user_data'),
  ])
  .execute();

分页插件

import { SelectQueryBuilder } from 'kysely';

function paginate<DB, TB extends keyof DB, O>(
  query: SelectQueryBuilder<DB, TB, O>,
  page: number,
  pageSize: number
) {
  return query.limit(pageSize).offset((page - 1) * pageSize);
}

// Usage
const page = 2;
const pageSize = 20;

const users = await paginate(
  db.selectFrom('users').selectAll(),
  page,
  pageSize
).execute();

// With total count
async function paginateWithCount<DB, TB extends keyof DB, O>(
  query: SelectQueryBuilder<DB, TB, O>,
  page: number,
  pageSize: number
) {
  const [items, { count }] = await Promise.all([
    query.limit(pageSize).offset((page - 1) * pageSize).execute(),
    query.select(db.fn.count<number>('id').as('count')).executeTakeFirstOrThrow(),
  ]);

  return {
    items,
    total: count,
    page,
    pageSize,
    totalPages: Math.ceil(count / pageSize),
  };
}

全文搜索（PostgreSQL）

// GIN index for full-text search
export async function up(db: Kysely<any>): Promise<void> {
  await sql`
    ALTER TABLE posts
    ADD COLUMN search_vector tsvector
    GENERATED ALWAYS AS (
      to_tsvector('english', coalesce(title, '') || ' ' || coalesce(content, ''))
    ) STORED
  `.execute(db);

  await sql`
    CREATE INDEX posts_search_idx ON posts USING GIN (search_vector)
  `.execute(db);
}

// Full-text search query
const searchResults = await db
  .selectFrom('posts')
  .selectAll()
  .where(
    sql`search_vector`,
    '@@',
    sql`to_tsquery('english', ${query})`
  )
  .execute();

Kysely、Drizzle 与 Prisma 对比

功能对比

功能	Kysely	Drizzle	Prisma
类型安全	完整（架构 → 查询）	完整（架构 → 查询）	完整（生成的客户端）
SQL 控制	✅ 支持原始 SQL	✅ 支持原始 SQL	❌ 有限
包体积	~50kB	~30kB	~500kB+
迁移系统	✅ 内置	✅ 内置	✅ 强大的 CLI
查询性能	✅ 纯 SQL	✅ 纯 SQL	❌ 较慢（抽象层）
架构定义	TypeScript 类型	TypeScript 架构	Prisma 架构
是否需要代码生成	可选	否	✅ 需要
ORM 功能	❌ 仅查询构建器	部分（关系型）	✅ 完整 ORM
学习曲线	中等（需要 SQL 知识）	中等	简单（抽象了 SQL）
最适合	SQL 优先、复杂查询	类型安全的架构	快速原型开发

何时选择 Kysely

✅ 选择 Kysely 的情况：

• 你熟悉 SQL，并希望全面掌控查询
• 需要处理复杂查询（CTE、窗口函数、子查询）
• 性能至关重要（没有 ORM 开销）
• 从原始 SQL 迁移
• 经常需要使用原始 SQL 作为备用方案
• 使用现有数据库
• 包体积很重要（边缘函数）

❌ 选择 Drizzle 的情况：

• 希望使用声明式 TypeScript 架构
• 需要关系型查询功能
• 希望兼具 ORM 风格的易用性与 SQL 控制能力
• 正在开展全新的项目

❌ 选择 Prisma 的情况：

• 团队不熟悉 SQL
• 需要快速原型开发和迭代
• 需要强大的迁移工具
• 希望自动处理关系
• 更偏好声明式架构语言

从 Prisma 迁移

// Prisma
const users = await prisma.user.findMany({
  where: { createdAt: { gte: new Date('2024-01-01') } },
  include: { posts: true },
});

// Kysely equivalent
const users = await db
  .selectFrom('users')
  .select([
    'users.id',
    'users.email',
    jsonArrayFrom(
      db.selectFrom('posts')
        .selectAll()
        .whereRef('posts.user_id', '=', 'users.id')
    ).as('posts'),
  ])
  .where('created_at', '>=', new Date('2024-01-01'))
  .execute();

最佳实践

1. 先定义架构类型 - 使用 Generated、Selectable、Insertable、Updateable
2. 使用 kysely-codegen - 从现有数据库生成类型
3. 充分利用类型推断 - 让 TypeScript 推断结果类型
4. 使用事务 - 处理多步骤操作
5. 按需使用原始 SQL - 不要勉强查询构建器
6. 对大型结果进行分页 - 使用 LIMIT/OFFSET 或基于游标的分页
7. 为经常查询的列建立索引 - 性能优化由你负责
8. 测试迁移 - 同时测试向上和向下迁移
9. 使用 CTE 提高可读性 - 让复杂查询变得易于维护
10. 连接池 - 适当配置数据库连接池

常见陷阱

❌ 忘记执行查询：

// WRONG - returns query builder, not results
const users = db.selectFrom('users').selectAll();

// CORRECT
const users = await db.selectFrom('users').selectAll().execute();

❌ 未处理 LEFT JOIN 返回的 null：

// TypeScript knows posts.title can be null from LEFT JOIN
const result = await db
  .selectFrom('users')
  .leftJoin('posts', 'posts.user_id', 'users.id')
  .select(['users.name', 'posts.title'])
  .execute();
// posts.title type: string | null

❌ 自增列缺少 Generated：

// WRONG - TypeScript will require 'id' in INSERT
interface UserTable {
  id: number;  // Bad!
}

// CORRECT
interface UserTable {
  id: Generated<number>;  // INSERT doesn't require id
}

资源

• 文档：https://kysely.dev
• GitHub：https://github.com/kysely-org/kysely
• Discord：https://discord.gg/kysely
• kysely-codegen：https://github.com/RobinBlomberg/kysely-codegen
• 演练场：https://kysely-org.github.io/kysely-playground/

相关技能

使用 Kysely 时，可以考虑搭配以下技能：

• typescript-core：TypeScript 类型系统、高级模式和 tsconfig 优化
• database-migration：适用于生产数据库的安全架构演进模式
• Node.js backend：服务设置、连接池和数据库配置

TypeScript 类型系统快速参考（内联，方便独立使用）

// Kysely leverages advanced TypeScript features
import { Kysely, Generated, ColumnType } from 'kysely';

// Database interface with Generated types
interface Database {
  users: {
    id: Generated<number>;  // Auto-generated by database
    email: string;
    created_at: ColumnType<Date, string | undefined, never>;
    // ColumnType<SelectType, InsertType, UpdateType>
  };
}

// Type inference in queries
const db = new Kysely<Database>({ /* config */ });

// Full type safety - TypeScript knows return type
const users = await db
  .selectFrom('users')
  .select(['id', 'email'])
  .where('created_at', '>', new Date('2025-01-01'))
  .execute();
// Type: Array<{ id: number; email: string }>

// Conditional types for dynamic queries
type SelectFields<T> = {
  [K in keyof T]: T[K] extends ColumnType<infer S, any, any> ? S : T[K];
};

数据库迁移模式快速参考（内联，方便独立使用）

安全迁移原则：

1. 向后兼容 - 新代码可与旧架构配合使用
2. 可逆 - 需要时可以回滚迁移
3. 零停机 - 不会中断服务
4. 渐进式 - 小步修改，而非大规模重写

Kysely 迁移示例：

// migrations/001_add_full_name.ts
import { Kysely, sql } from 'kysely';

export async function up(db: Kysely<any>): Promise<void> {
  // Phase 1: Add new column (nullable initially)
  await db.schema
    .alterTable('users')
    .addColumn('full_name', 'varchar(255)')
    .execute();

  // Phase 2: Backfill data
  await db
    .updateTable('users')
    .set({
      full_name: sql`concat(first_name, ' ', last_name)`
    })
    .execute();

  // Phase 3: Make required (separate migration recommended)
  // await db.schema
  //   .alterTable('users')
  //   .alterColumn('full_name', (col) => col.setNotNull())
  //   .execute();
}

export async function down(db: Kysely<any>): Promise<void> {
  await db.schema
    .alterTable('users')
    .dropColumn('full_name')
    .execute();
}

常见安全迁移：

// Add index (concurrently for PostgreSQL)
await db.schema
  .createIndex('idx_users_email')
  .on('users')
  .column('email')
  .execute();

// Rename column (multi-phase approach)
// Phase 1: Add new column
await db.schema
  .alterTable('users')
  .addColumn('email_address', 'varchar(255)')
  .execute();

// Phase 2: Copy data
await db
  .updateTable('users')
  .set({ email_address: sql`email` })
  .execute();

// Phase 3: Drop old column (after deploy)
// await db.schema
//   .alterTable('users')
//   .dropColumn('email')
//   .execute();

// Change column type (add new, migrate, drop old)
await db.schema
  .alterTable('products')
  .addColumn('price_cents', 'integer')
  .execute();

await db
  .updateTable('products')
  .set({ price_cents: sql`cast(price * 100 as integer)` })
  .execute();

运行迁移：

// migrate.ts
import { Kysely, Migrator, FileMigrationProvider } from 'kysely';
import { promises as fs } from 'fs';
import path from 'path';

const migrator = new Migrator({
  db,
  provider: new FileMigrationProvider({
    fs,
    path,
    migrationFolder: path.join(__dirname, 'migrations'),
  }),
});

// Migrate to latest
const { error, results } = await migrator.migrateToLatest();

// Migrate up/down
await migrator.migrateUp();
await migrator.migrateDown();

// List pending migrations
const migrations = await migrator.getMigrations();

[如果同时部署，完整的 TypeScript 模式和迁移工作流可在相应技能中查看]

总结

• Kysely 是类型安全的 SQL 查询构建器，而非 ORM
• 从架构定义到查询结果均可进行完整的类型推断
• 零运行时开销 - 编译为纯 SQL
• 内置支持 up/down 的迁移系统
• 查询构建器无法满足需求时，可集成原始 SQL
• 提供用于 JSON、分页和全文搜索的插件生态系统
• 适用于熟悉 SQL 且需要类型安全的开发者
• 可替代 Prisma（完整 ORM）和 Drizzle（以架构为先）
• 非常适合复杂查询、现有数据库和性能关键型应用