Cloudflare D1 API 完全指南Worker 查询方法、批量事务与高级特性实战【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills本文是cloudflare-deploy技能中 D1 API 参考 的深度展开面向在 Cloudflare Workers 中使用 D1基于 SQLite 的服务端无服务器数据库的开发者。文章系统讲解从安全的预编译语句、四种查询执行方法.all()/.first()/.run()/.raw()、单次往返的批量事务到付费计划专属的 Sessions API、读复制与 REST API 远程访问最后给出错误处理与本地调试的完整实战方案。读完你能够安全、高效地在 Worker 中编写生产级 D1 数据访问代码并具备通过 HTTP 在非 Worker 场景下操作 D1 的能力。D1 与 Worker 绑定env.DB从哪来在阅读 API 之前先明确env.DB的由来。D1 通过 Cloudflare Workers 的绑定binding机制注入运行时环境。在 wrangler.jsonc 配置 中声明d1_databases数组其中binding字段就是env上对应的属性名{ name: your-worker-name, main: src/index.ts, compatibility_date: 2025-01-01, // Use current date for new projects d1_databases: [ { binding: DB, // Env variable name database_name: your-db-name, // Human-readable name database_id: your-database-id, // UUID from dashboard/CLI migrations_dir: migrations // Optional: default is migrations }, // Read replica (paid plans only) { binding: DB_REPLICA, database_name: your-db-name, database_id: your-database-id // Same ID, different binding }, // Multiple databases { binding: ANALYTICS_DB, database_name: analytics-db, database_id: yyy-yyy-yyy } ] }D1 的 binding 类型是D1Database来自cloudflare/workers-types包参见 Bindings API 参考 中的类型对照表。声明好配置后用npx wrangler types可自动生成类型安全的Env接口interface Env { DB: D1Database; // Primary (writes) DB_REPLICA: D1Database; // Replica (reads) } export default { async fetch(request: Request, env: Env, ctx: ExecutionContext): PromiseResponse { const result await env.DB.prepare(SELECT * FROM users).all(); return Response.json(result.results); } }D1 的核心设计哲学是面向 per-user / per-tenant / per-entity 的数据库模式即一个实体一个库而非维护单个超大库见 D1 总览。这解释了为什么 API 设计偏向轻量、快速、低延迟的 SQLite 语义。预编译语句安全底线没有任何例外D1 API 的第一原则是所有带外部输入的 SQL 必须使用预编译语句Prepared Statement。直接做字符串拼接等于把 SQL 注入漏洞暴露给攻击者。// ❌ NEVER: Direct string interpolation (SQL injection risk) const result await env.DB.prepare(SELECT * FROM users WHERE id ${userId}).all(); // ✅ CORRECT: Prepared statements with bind() const result await env.DB.prepare(SELECT * FROM users WHERE id ?).bind(userId).all(); // Multiple parameters const result await env.DB.prepare(SELECT * FROM users WHERE email ? AND active ?).bind(email, true).all();使用规则SQL 文本中所有动态值一律以?占位符出现随后通过.bind()按顺序绑定参数.bind()支持多个参数类型可以是字符串、数字、布尔值等布尔值注意SQLite 原生没有布尔类型实际存储为 INTEGER1/0。虽然 API 层面允许bind(true)但 gotchas 文档 明确提示与布尔列打交道时绑定1或0更稳妥日期时间同理SQLite 没有原生 DATE/TIME 类型建议用 TEXTISO 8601或 INTEGERUnix 时间戳存储。在 gotchas 文档 中SQL Injection Vulnerability 被列为最常见错误之一原因是字符串插值解决方案就是prepare()bind()的组合。这条规则对 SELECT 和写操作一视同仁没有例外。查询执行方法.all()/.first()/.run()/.raw()D1 提供四种查询执行方法各自对应不同的返回形态和适用场景。.all()取全部行const { results, success, meta } await env.DB.prepare(SELECT * FROM users WHERE active ?).bind(true).all(); // results: Array of row objects; success: boolean // meta: { duration: number, rows_read: number, rows_written: number }返回结构包含三部分results行对象数组每个元素对应一行字段名即列名success布尔值指示查询是否成功meta执行元数据duration是耗时毫秒、rows_read是读取的行数、rows_written是写入的行数。meta.duration常用于调试查询性能见下文测试与调试章节。.first()取首行或单列// .first() - Returns first row or null const user await env.DB.prepare(SELECT * FROM users WHERE id ?).bind(userId).first(); // .first(columnName) - Returns single column value const email await env.DB.prepare(SELECT email FROM users WHERE id ?).bind(userId).first(email); // Returns string | number | null无参数调用返回第一个行对象无结果时返回null传入列名参数时返回该列的单个值类型为string | number | null。适合按 ID 取某个字段这类查询。.run()执行写操作const result await env.DB.prepare(UPDATE users SET last_login ? WHERE id ?).bind(Date.now(), userId).run(); // result.meta: { duration, rows_read, rows_written, last_row_id, changes }.run()专用于 INSERT / UPDATE / DELETE不返回行数据但meta额外携带两个关键字段last_row_id刚插入行的自增 ID在插入后立即回读场景中非常有用changes受影响的行数可用于判断是否真的发生了更新。.raw()返回嵌套数组const rawResults await env.DB.prepare(SELECT id, name FROM users).raw(); // [[1, Alice], [2, Bob]].raw()返回数组的数组没有列名映射序列化开销更小。原文档特别标注它efficient for large datasets——在大数据集场景下比对象数组更省内存和带宽。提示在 D1 模式与最佳实践 中条件动态查询如多条件搜索也是通过prepare() 动态拼接?占位符 bind(...params)实现的这与上述四种方法完全兼容。批量操作单次往返的原子事务env.DB.batch()接受一个预编译语句数组在一次网络往返内按顺序执行且默认具备原子性——要么全部成功要么全部失败// Execute multiple queries in single round trip (atomic transaction) const results await env.DB.batch([ env.DB.prepare(SELECT * FROM users WHERE id ?).bind(1), env.DB.prepare(SELECT * FROM posts WHERE author_id ?).bind(1), env.DB.prepare(UPDATE users SET last_access ? WHERE id ?).bind(Date.now(), 1) ]); // results is array: [result1, result2, result3]两个高频用法同一语句、不同参数的批量执行例如批量查询或批量插入// Batch with same prepared statement, different params const userIds [1, 2, 3]; const stmt env.DB.prepare(SELECT * FROM users WHERE id ?); const results await env.DB.batch(userIds.map(id stmt.bind(id)));批量插入来自 patterns 文档async function bulkInsertUsers(users: Array{ name: string; email: string }, env: Env) { const stmt env.DB.prepare(INSERT INTO users (name, email) VALUES (?, ?)); const batch users.map(user stmt.bind(user.name, user.email)); return await env.DB.batch(batch); }需要注意平台限制详见 configuration.md 的 Plan Tiers 表和 gotchas.md免费计划单批最多1,000 条语句付费计划10,000 条超出会报 Batch size exceeded需要分块处理for (let i 0; i stmts.length; i MAX_BATCH) await env.DB.batch(stmts.slice(i, i MAX_BATCH))。事务用 batch 实现全有或全无D1 的事务模型很特别没有独立的BEGIN/COMMIT语句事务就是batch()本身。D1 将传入的语句数组作为一个原子单元执行// D1 executes batch() as atomic transaction - all succeed or all fail const results await env.DB.batch([ env.DB.prepare(INSERT INTO accounts (id, balance) VALUES (?, ?)).bind(1, 100), env.DB.prepare(INSERT INTO accounts (id, balance) VALUES (?, ?)).bind(2, 200), env.DB.prepare(UPDATE accounts SET balance balance - ? WHERE id ?).bind(50, 1), env.DB.prepare(UPDATE accounts SET balance balance ? WHERE id ?).bind(50, 2) ]);上述转账示例中扣款与入账在一个 batch 中完成任一步失败则整体回滚保证账目一致。这也呼应了 patterns 文档中的分页方案——把COUNT(*)和数据查询放进同一个batch()减少往返async function getUsers({ page, pageSize }: { page: number; pageSize: number }, env: Env) { const offset (page - 1) * pageSize; const [countResult, dataResult] await env.DB.batch([ env.DB.prepare(SELECT COUNT(*) as total FROM users), env.DB.prepare(SELECT * FROM users ORDER BY created_at DESC LIMIT ? OFFSET ?).bind(pageSize, offset) ]); return { data: dataResult.results, total: countResult.results[0].total, page, pageSize, totalPages: Math.ceil(countResult.results[0].total / pageSize) }; }Sessions API突破 30 秒的超长任务付费计划普通 D1 查询有30 秒超时免费与付费一致。对于建大索引、ANALYZE、批量数据迁移这类重型操作需要借助Sessions API开启最长15 分钟的长连接会话const session env.DB.withSession({ timeout: 600 }); // 10 min (1-900s) try { await session.prepare(CREATE INDEX idx_large ON big_table(column)).run(); await session.prepare(ANALYZE).run(); } finally { session.close(); // CRITICAL: always close to prevent leaks }关键要点withSession({ timeout })的timeout单位为秒取值范围1–900即最长 15 分钟session上的 API 与env.DB基本一致prepare()、run()、batch()都可用session.close()必须执行——文档以 CRITICAL: always close to prevent leaks 强调否则会造成资源泄漏最佳实践是try/finally结构gotchas 中 Session not closed / resource leak 正是对应此坑典型场景迁移Migrations、ANALYZE、大型索引创建、批量数据转换。来自 patterns 文档的进阶示例——用 session 分批转换大表数据async function transformLargeDataset(env: Env) { const session env.DB.withSession({ timeout: 900 }); // 15 min max try { const BATCH_SIZE 1000; let offset 0; while (true) { const rows await session.prepare(SELECT id, data FROM legacy LIMIT ? OFFSET ?).bind(BATCH_SIZE, offset).all(); if (rows.results.length 0) break; const updates rows.results.map(row session.prepare(UPDATE legacy SET new_data ? WHERE id ?).bind(transform(row.data), row.id) ); await session.batch(updates); offset BATCH_SIZE; } } finally { session.close(); } }读复制读写分离降低延迟付费计划读复制Read Replication将 SELECT 查询路由到地理上最近的副本降低读取延迟写入始终走主库Primary。架构上通过给同一个数据库配置两个 binding 实现——DB作为主库、DB_REPLICA作为副本见前文 wrangler.jsonc 配置示例副本使用同一个database_id、不同 binding 名interface Env { DB: D1Database; // Primary (writes) DB_REPLICA: D1Database; // Replica (reads) } // Reads: use replica const user await env.DB_REPLICA.prepare(SELECT * FROM users WHERE id ?).bind(userId).first(); // Writes: use primary await env.DB.prepare(UPDATE users SET last_login ? WHERE id ?).bind(Date.now(), userId).run(); // Read-after-write: use primary for consistency (replication lag 100ms-2s) await env.DB.prepare(INSERT INTO posts (title) VALUES (?)).bind(title).run(); const post await env.DB.prepare(SELECT * FROM posts WHERE title ?).bind(title).first(); // Primary关键取舍是复制延迟副本与主库之间存在100ms–2s的复制延迟。因此适合走副本分析看板、搜索结果、公开查询等可接受最终一致性的场景见 patterns 文档的读复制模式必须走主库读后写read-after-write、金融交易、身份认证等要求强一致的场景——写完之后立即读一定用主库否则可能读到旧数据。patterns 文档给出一个完整的 REST 风格读写分离实现GET 请求走DB_REPLICAPOST 写入走DB并用result.meta.last_row_id从主库回读新记录返回 201。错误处理success 标志、异常与约束冲突D1 的错误处理需要同时照顾两层返回结果里的success标志以及抛出的异常。基于返回值与异常的双重防护async function getUser(userId: number, env: Env): PromiseResponse { try { const result await env.DB.prepare(SELECT * FROM users WHERE id ?).bind(userId).all(); if (!result.success) return new Response(Database error, { status: 500 }); if (result.results.length 0) return new Response(User not found, { status: 404 }); return Response.json(result.results[0]); } catch (error) { return new Response(Internal error, { status: 500 }); } }这套模式把数据库自身失败500、记录不存在404与未捕获异常500三种情况清晰区分开。约束冲突映射为 409唯一约束UNIQUE冲突在 D1 中会以异常抛出错误信息包含UNIQUE constraint failed字样可据此映射为 HTTP 409 Conflict// Constraint violations try { await env.DB.prepare(INSERT INTO users (email, name) VALUES (?, ?)).bind(email, name).run(); } catch (error) { if (error.message?.includes(UNIQUE constraint failed)) return new Response(Email exists, { status: 409 }); throw error; }这与 gotchas.md 中 UNIQUE constraint failed 条目的建议一致捕获错误并返回 409。同文档还列出了其他高频错误no such table迁移未执行或 binding 名不匹配、查询超时30s 超限需拆小查询或加索引、外键约束失败需要PRAGMA foreign_keys ON;并配合ON DELETE CASCADE等可作为错误处理分支设计的参考清单。REST API在 Worker 之外访问 D1D1 不只限于 Worker 内部使用。通过 Cloudflare API 的 HTTP 端点可以在任何外部服务Node 脚本、CI/CD、管理工具等中执行查询只需账号 ID、数据库 ID 和 API Token。单条查询// Single query const response await fetch( https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/d1/database/${DATABASE_ID}/query, { method: POST, headers: { Authorization: Bearer ${CLOUDFLARE_API_TOKEN}, Content-Type: application/json }, body: JSON.stringify({ sql: SELECT * FROM users WHERE id ?, params: [userId] }) } ); const { result, success, errors } await response.json(); // result: [{ results: [...], success: true, meta: {...} }]请求体为{ sql, params }params对应?占位符。注意响应中的result是一个数组内部元素结构与 Worker API 一致results/success/meta这在解析时容易踩坑。批量查询// Batch queries via HTTP const response await fetch( https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/d1/database/${DATABASE_ID}/query, { method: POST, headers: { Authorization: Bearer ${CLOUDFLARE_API_TOKEN}, Content-Type: application/json }, body: JSON.stringify([ { sql: SELECT * FROM users WHERE id ?, params: [1] }, { sql: SELECT * FROM posts WHERE author_id ?, params: [1] } ]) } );HTTP 层的批量模式对应 Worker 中的batch()把请求体从单个对象换成对象数组即可。典型使用场景服务器端脚本、CI/CD 迁移、管理工具、非 Worker 集成。对应的命令行替代方案来自 D1 总览 的 CLI Commands包括wrangler d1 execute db-name --remote --commandSELECT * FROM users和wrangler d1 execute db-name --remote --file./schema.sql适合交互式调试和脚本化执行。测试与调试从 Vitest 到查询计划集成测试Vitest unstable_dev在本地用 Vitest 配合 wrangler 的unstable_dev启动真实 Worker 做端到端测试// Vitest with unstable_dev import { unstable_dev } from wrangler; describe(D1, () { let worker: AwaitedReturnTypetypeof unstable_dev; beforeAll(async () { worker await unstable_dev(src/index.ts); }); afterAll(async () { await worker.stop(); }); it(queries users, async () { expect((await worker.fetch(/users)).status).toBe(200); }); });性能与执行计划分析用meta.duration量化单次查询耗时用EXPLAIN QUERY PLAN检查索引是否生效// Debug query performance const result await env.DB.prepare(SELECT * FROM users).all(); console.log(Duration:, result.meta.duration, ms); // Query plan analysis const plan await env.DB.prepare(EXPLAIN QUERY PLAN SELECT * FROM users WHERE email ?).bind(email).all();本地数据库直查本地开发时wrangler dev --persist-to./.wrangler/state见 configuration.md数据库是落在磁盘上的真实 SQLite 文件可直接用sqlite3客户端检查表结构# Inspect local database sqlite3 .wrangler/state/v3/d1/database-id.sqlite .tables; .schema users; PRAGMA table_info(users);补充写入生产前必读的边界条件为了让本文中的 API 用法在真实环境中不翻车补充几条来自 gotchas.md 的硬性边界布尔与日期SQLite 无布尔/日期原生类型布尔绑定1/0日期用 TEXTISO 8601或 INTEGERUnix 时间戳行大小上限 1 MB免费与付费一致大文件请存 R2D1 只存 URL/KeyBLOB 在导出时可能损坏同样建议走 R2数据库大小免费 500 MB、付费 10 GB/库临近上限时按 per-tenant/per-user 水平拆分、归档旧数据或升级付费计划本地与生产差异本地是单文件 SQLite生产是分布式 D1性能与限制不同上线前务必用wrangler d1 migrations apply db-name --remote在生产端验证迁移N1 查询循环内逐条查询是反面教材应改用 JOIN 或batch()patterns 文档给出了 JOIN 替代方案。总结本文以 api.md 为骨架完整覆盖了 D1 的编程接口全貌预编译语句是安全底线.all()/.first()/.run()/.raw()四种方法覆盖读写与大数据量场景batch()兼作批量优化与原子事务Sessions API 与读复制是付费计划下突破超时、优化延迟的两大利器REST API 让非 Worker 场景也能操作 D1最后配合测试、调试与 gotchas 边界形成一套可落地的生产实践。继续深入可参考同目录下的 configuration.mdwrangler.jsonc 配置、迁移与索引策略、patterns.md分页、缓存、多租户等模式与 gotchas.md常见错误排查它们与本文互为补充共同构成完整的 D1 使用手册。【免费下载链接】skillsSkills Catalog for Codex项目地址: https://gitcode.com/GitHub_Trending/skills4/skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考