系统编程

Zig 数据库访问与轻量 ORM：SQLite、PostgreSQL 与自定义 SQL

Zig 通过 C ABI 绑定可无缝调用 SQLite 和 PostgreSQL，配合参数化查询防注入、连接池设计与事务封装，构建安全高效的数据库访问层。本文系统讲解 SQLite CRUD、PostgreSQL 连接池、轻量 ORM 结构体映射、迁移脚本设计与连接生命周期管理。

Plume 6 分钟阅读 2585 字

引言

数据库访问是服务端开发的核心环节。Zig 没有官方 ORM 框架，但借助 C ABI 互操作、显式分配器管理和 comptime 泛型，可以用极少的代码封装出类型安全、零开销的数据库访问层。本文覆盖 SQLite 原生 C 绑定、PostgreSQL libpq 连接、参数化查询防注入、连接池、事务与轻量 ORM 结构体映射。

前置：/zig-sqlite-storage/（SQLite 基础绑定）、/zig-c-interoperability/（@cImport 与 C 类型映射）、/zig-memory-management/（分配器生命周期）。

目录

1. SQLite CRUD 与参数化查询

SQLite 是最合适 Zig 入门的数据库：单文件、零配置、C API 简洁。通过 @cImport 导入 sqlite3.h 后即可操作。

1.1 初始化与打开

const std = @import("std");
const c = @cImport({
    @cInclude("sqlite3.h");
});

pub const DbError = error{OpenFailed, PrepareFailed, StepFailed};

pub const SQLiteDb = struct {
    db: ?*c.sqlite3,

    pub fn open(path: [:0]const u8) DbError!SQLiteDb {
        var db: ?*c.sqlite3 = null;
        const rc = c.sqlite3_open(path, &db);
        if (rc != c.SQLITE_OK) return error.OpenFailed;
        return SQLiteDb{ .db = db };
    }

    pub fn close(self: *SQLiteDb) void {
        if (self.db) |ptr| {
            _ = c.sqlite3_close(ptr);
            self.db = null;
        }
    }
};

1.2 参数化查询防注入

绝不拼接 SQL 字符串。使用预编译语句（prepared statement）和参数绑定：

pub fn insertUser(self: *SQLiteDb, name: []const u8, age: i32) !void {
    const sql = "INSERT INTO users(name, age) VALUES(?, ?)";
    var stmt: ?*c.sqlite3_stmt = null;

    // 预编译 SQL
    if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK) {
        return error.PrepareFailed;
    }
    defer _ = c.sqlite3_finalize(stmt);

    // 绑定参数（索引从 1 开始）
    _ = c.sqlite3_bind_text(stmt, 1, name.ptr, @intCast(name.len), c.SQLITE_STATIC);
    _ = c.sqlite3_bind_int(stmt, 2, age);

    // 执行
    if (c.sqlite3_step(stmt) != c.SQLITE_DONE) {
        return error.StepFailed;
    }
}

参数绑定自动处理转义：name 中的 ' OR '1'='1 会被当作纯文本存储，不会破坏 SQL 结构。

1.3 查询与结果遍历

pub const User = struct {
    id: i32,
    name: []const u8,
    age: i32,
};

pub fn queryUsers(
    self: *SQLiteDb,
    allocator: std.mem.Allocator,
    min_age: i32,
) ![]User {
    const sql = "SELECT id, name, age FROM users WHERE age >= ?";
    var stmt: ?*c.sqlite3_stmt = null;
    if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK) {
        return error.PrepareFailed;
    }
    defer _ = c.sqlite3_finalize(stmt);

    _ = c.sqlite3_bind_int(stmt, 1, min_age);

    var list = std.ArrayList(User).init(allocator);
    defer list.deinit();

    while (c.sqlite3_step(stmt) == c.SQLITE_ROW) {
        const id = c.sqlite3_column_int(stmt, 0);
        const name_ptr = c.sqlite3_column_text(stmt, 1);
        const name_len = c.sqlite3_column_bytes(stmt, 1);
        const age = c.sqlite3_column_int(stmt, 2);

        // 拷贝到 allocator，避免 SQLite 内部缓冲区被释放后悬垂
        const name_copy = try allocator.dupe(u8, name_ptr[0..@intCast(name_len)]);
        try list.append(User{
            .id = id,
            .name = name_copy,
            .age = age,
        });
    }

    return list.toOwnedSlice();
}

注意：sqlite3_column_text 返回的指针生命周期与 statement 绑定，需要 dupe 拷贝到 allocator 管理的内存。

2. PostgreSQL 连接与 libpq 绑定

2.1 连接建立

const pq = @cImport({
    @cInclude("libpq-fe.h");
});

pub const PgConn = struct {
    conn: ?*pq.PGconn,

    pub fn connect(conninfo: [:0]const u8) !PgConn {
        const conn = pq.PQconnectdb(conninfo);
        if (pq.PQstatus(conn) != pq.CONNECTION_OK) {
            std.log.err("PG 连接失败: {s}", .{pq.PQerrorMessage(conn)});
            pq.PQfinish(conn);
            return error.ConnectionFailed;
        }
        return PgConn{ .conn = conn };
    }

    pub fn deinit(self: *PgConn) void {
        if (self.conn) |c| pq.PQfinish(c);
        self.conn = null;
    }
};

2.2 参数化查询

libpq 使用 $N 占位符：

pub fn insertProduct(self: *PgConn, name: []const u8, price: f64) !void {
    const sql = "INSERT INTO products(name, price) VALUES($1, $2)";
    const name_z = try std.mem.joinZ(std.heap.page_allocator, "", &.{name});
    defer std.heap.page_allocator.free(name_z);

    const values = [_][*c]const u8{
        name_z.ptr,
        std.fmt.allocPrintZ(std.heap.page_allocator, "{d}", .{price}) catch unreachable,
    };
    defer std.heap.page_allocator.free(values[1]);

    const result = pq.PQexecParams(
        self.conn,
        sql,
        2,              // nParams
        null,           // paramTypes (让 PG 推断)
        &values,
        null,           // paramLengths (文本模式下忽略)
        null,           // paramFormats (0 = 文本)
        0,              // resultFormat
    );
    defer pq.PQclear(result);

    if (pq.PQresultStatus(result) != pq.PGRES_COMMAND_OK) {
        std.log.err("插入失败: {s}", .{pq.PQerrorMessage(self.conn)});
        return error.ExecFailed;
    }
}

3. 连接池设计

连接池避免频繁创建/销毁 TCP 连接的开销。Zig 中用 std.ArrayList + Mutex 即可实现轻量池。

const std = @import("std");

pub fn Pool(comptime Conn: type, comptime max_size: usize) type {
    return struct {
        const Self = @This();

        allocator: std.mem.Allocator,
        mutex: std.Thread.Mutex,
        available: std.ArrayList(Conn),
        factory: *const fn (std.mem.Allocator) anyerror!Conn,

        pub fn init(allocator: std.mem.Allocator, factory: anytype) !Self {
            var available = std.ArrayList(Conn).init(allocator);
            // 预热：预先创建一批连接
            var i: usize = 0;
            while (i < max_size / 2) : (i += 1) {
                const conn = try factory(allocator);
                try available.append(conn);
            }
            return Self{
                .allocator = allocator,
                .mutex = .{},
                .available = available,
                .factory = factory,
            };
        }

        pub fn acquire(self: *Self) !Conn {
            self.mutex.lock();
            defer self.mutex.unlock();

            if (self.available.items.len > 0) {
                return self.available.pop();
            }
            // 池为空时创建新连接（不严格限制上限的简化版）
            return self.factory(self.allocator);
        }

        pub fn release(self: *Self, conn: Conn) void {
            self.mutex.lock();
            defer self.mutex.unlock();
            if (self.available.items.len < max_size) {
                self.available.append(conn) catch {};
            } else {
                // 池已满，直接关闭
                // conn.deinit(); // 视 Conn 类型而定
                _ = conn;
            }
        }

        pub fn deinit(self: *Self) void {
            for (self.available.items) |conn| {
                _ = conn; // conn.deinit();
            }
            self.available.deinit();
        }
    };
}

这是简化示例。生产级连接池还需添加：健康检查（ping 检测）、超时回收（idle timeout）、阻塞等待（条件变量）和最大连接数限制。

4. 事务封装与 errdefer

Zig 的 errdefer 是事务回滚的语义利器：函数任何位置返回错误时自动执行回滚，成功时显式提交。

pub fn transferFunds(
    self: *SQLiteDb,
    from_id: i32,
    to_id: i32,
    amount: i32,
) !void {
    // 开始事务
    _ = c.sqlite3_exec(self.db, "BEGIN", null, null, null);

    // 任意位置返回 error 时自动回滚
    errdefer _ = c.sqlite3_exec(self.db, "ROLLBACK", null, null, null);

    // 扣款
    const debit = "UPDATE accounts SET balance = balance - ? WHERE id = ?";
    try self.execParams(debit, .{ amount, from_id });

    // 入账
    const credit = "UPDATE accounts SET balance = balance + ? WHERE id = ?";
    try self.execParams(credit, .{ amount, to_id });

    // 记录流水
    const log_sql = "INSERT INTO transactions(from_id, to_id, amount) VALUES(?,?,?)";
    try self.execParams(log_sql, .{ from_id, to_id, amount });

    // 全部成功，提交事务
    _ = c.sqlite3_exec(self.db, "COMMIT", null, null, null);
}

errdefer 在成功路径上不执行，在错误路径上按逆序执行。这是 Zig 结构化错误处理最优雅的实践之一。

5. 轻量 ORM：结构体表映射

5.1 反射生成 CRUD

利用 comptime 反射，可以从结构体自动生成 INSERT 和 SELECT：

pub fn Repository(comptime T: type) type {
    return struct {
        pub fn insert(
            self: *SQLiteDb,
            allocator: std.mem.Allocator,
            item: T,
        ) !void {
            const info = @typeInfo(T).Struct;
            var cols = std.ArrayList([]const u8).init(allocator);
            var placeholders = std.ArrayList(u8).init(allocator);
            defer cols.deinit();
            defer placeholders.deinit();

            inline for (info.fields) |field| {
                try cols.append(field.name);
                try placeholders.appendSlice("?,");
            }

            const sql = try std.fmt.allocPrint(allocator, "INSERT INTO {s}({s}) VALUES({s})", .{
                @typeName(T),
                try std.mem.join(allocator, ",", cols.items),
                placeholders.items[0 .. placeholders.items.len - 1], // 去掉末尾逗号
            });
            defer allocator.free(sql);

            // ... 绑定参数并执行
            _ = item;
            _ = self;
        }
    };
}

5.2 显式映射

对于复杂查询，显式映射更清晰可控：

pub fn mapRowToUser(stmt: ?*c.sqlite3_stmt, allocator: std.mem.Allocator) !User {
    const id = c.sqlite3_column_int(stmt, 0);
    const name_ptr = c.sqlite3_column_text(stmt, 1);
    const name_len = c.sqlite3_column_bytes(stmt, 1);
    const age = c.sqlite3_column_int(stmt, 2);

    return User{
        .id = id,
        .name = try allocator.dupe(u8, name_ptr[0..@intCast(name_len)]),
        .age = age,
    };
}

6. 迁移脚本与 Schema 管理

6.1 版本化迁移文件

migrations/
  001_init.sql
  002_add_index.sql
  003_add_orders.sql

6.2 迁移执行器

pub fn migrate(self: *SQLiteDb, allocator: std.mem.Allocator, dir: []const u8) !void {
    // 创建迁移记录表
    _ = c.sqlite3_exec(self.db,
        "CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY)",
        null, null, null);

    // 读取迁移目录
    var d = try std.fs.cwd().openDir(dir, .{ .iterate = true });
    defer d.close();

    var iter = d.iterate();
    while (try iter.next()) |entry| {
        if (!std.mem.endsWith(u8, entry.name, ".sql")) continue;
        const version = try std.fmt.parseInt(i32, entry.name[0..3], 10);

        // 检查是否已执行
        if (try self.versionApplied(version)) continue;

        // 读取并执行 SQL
        const content = try d.readFileAlloc(allocator, entry.name, 1024 * 1024);
        defer allocator.free(content);

        if (c.sqlite3_exec(self.db, content.ptr, null, null, null) != c.SQLITE_OK) {
            std.log.err("迁移 {s} 失败", .{entry.name});
            return error.MigrateFailed;
        }

        try self.recordVersion(version);
        std.log.info("已应用迁移: {s}", .{entry.name});
    }
}

7. 速查表

需求手段
绑定 SQLite@cImport(@cInclude("sqlite3.h")) + -lsqlite3
绑定 PostgreSQL@cImport(@cInclude("libpq-fe.h")) + -lpq
防 SQL 注入预编译语句 + sqlite3_bind_* / PQexecParams
事务BEGIN + errdefer ROLLBACK + COMMIT
连接池ArrayList(Conn) + Mutex + acquire/release
ORM 映射comptime @typeInfo 遍历字段生成 SQL
迁移版本号文件 + schema_migrations 记录表
查询结果dupe 拷贝到 allocator 管理所有权

8. 一句话记忆

Zig 调数据库 = @cImport 导入 C API + 预编译参数化防注入 + errdefer 自动回滚事务 + allocator dup 查询结果管理所有权；comptime 泛型生成轻量 ORM，连接池 Mutex + ArrayList 够用，迁移脚本文件版本号 + schema_migrations 记录表。

相关阅读

  • /zig-sqlite-storage/ — SQLite 基础绑定与预编译语句
  • /zig-c-interoperability/ — C ABI 类型映射
  • /zig-comptime-programming/ — 编译期反射与泛型

延伸阅读

  • /zig-memory-management/ — allocator 生命周期与查询结果所有权
  • /zig-json-serialization/ — 查询结果序列化为 JSON API
  • /zig-http-server/ — 数据库驱动的 HTTP API 服务
  • [[zig]] — Zig 系统编程专题
// 完整示例：SQLite + 参数化查询 + 事务 + 结构体映射

const std = @import("std");
const c = @cImport({ @cInclude("sqlite3.h"); });

const User = struct {
    id: i32,
    name: []const u8,
    age: i32,
};

const Db = struct {
    db: ?*c.sqlite3,

    pub fn open(path: [:0]const u8) !Db {
        var db: ?*c.sqlite3 = null;
        if (c.sqlite3_open(path, &db) != c.SQLITE_OK) return error.OpenFailed;
        return Db{ .db = db };
    }
    pub fn close(self: *Db) void {
        if (self.db) |ptr| _ = c.sqlite3_close(ptr);
        self.db = null;
    }

    pub fn exec(self: *Db, sql: [:0]const u8) !void {
        if (c.sqlite3_exec(self.db, sql, null, null, null) != c.SQLITE_OK)
            return error.ExecFailed;
    }

    pub fn insertUser(self: *Db, name: []const u8, age: i32) !void {
        const sql = "INSERT INTO users(name, age) VALUES(?,?)";
        var stmt: ?*c.sqlite3_stmt = null;
        if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK)
            return error.PrepareFailed;
        defer _ = c.sqlite3_finalize(stmt);
        _ = c.sqlite3_bind_text(stmt, 1, name.ptr, @intCast(name.len), c.SQLITE_STATIC);
        _ = c.sqlite3_bind_int(stmt, 2, age);
        if (c.sqlite3_step(stmt) != c.SQLITE_DONE) return error.StepFailed;
    }

    pub fn queryUsers(self: *Db, allocator: std.mem.Allocator) ![]User {
        const sql = "SELECT id, name, age FROM users";
        var stmt: ?*c.sqlite3_stmt = null;
        if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK)
            return error.PrepareFailed;
        defer _ = c.sqlite3_finalize(stmt);

        var list = std.ArrayList(User).init(allocator);
        defer list.deinit();
        while (c.sqlite3_step(stmt) == c.SQLITE_ROW) {
            const name_ptr = c.sqlite3_column_text(stmt, 1);
            const name_len = c.sqlite3_column_bytes(stmt, 1);
            try list.append(.{
                .id = c.sqlite3_column_int(stmt, 0),
                .name = try allocator.dupe(u8, name_ptr[0..@intCast(name_len)]),
                .age = c.sqlite3_column_int(stmt, 2),
            });
        }
        return list.toOwnedSlice();
    }
};

pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    var db = try Db.open(":memory:");
    defer db.close();

    // 建表
    try db.exec("CREATE TABLE users(id INTEGER PRIMARY KEY, name TEXT, age INTEGER)");

    // 插入（参数化，防注入）
    try db.insertUser("Alice", 30);
    try db.insertUser("Bob', 'hacker'); DROP TABLE users;--", 25); // 安全：参数化

    // 查询
    const users = try db.queryUsers(allocator);
    defer {
        for (users) |u| allocator.free(u.name);
        allocator.free(users);
    }

    for (users) |u| {
        std.debug.print("User: id={d}, name={s}, age={d}\n", .{ u.id, u.name, u.age });
    }
}

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「系统编程」更多文章

  1. Zig 可观测性：结构化日志、OpenTelemetry 与指标采集
  2. Zig WebSocket 与实时通信：服务端推送与帧解析
  3. Zig 密码学与安全编程：AES、ChaCha20Poly1305 与 TLS