1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
const std = @import("std");
const pq = @cImport(
@cInclude("libpq-fe.h"),
);
const Db = @import("main.zig");
const OpenError = Db.OpenError;
const PrepareError = Db.PrepareError;
const StepError = Db.StepError;
const ColumnError = Db.ColumnError;
const log = @import("log.zig").scoped_log_t(.postgres);
/// Postgres implementation
/// Single persistent connection implementation of postgres via libpq
pub const Postgres = @This();
allocator: std.mem.Allocator,
conn: *pq.PGconn,
/// Connect to a postgres database. URL is format accepted by libpq
pub fn open(allocator: std.mem.Allocator, url: [:0]const u8) OpenError!Db {
if (pq.PQisthreadsafe() == 0) {
log.err("Postgres#open: PQisthreadsafe returned 0, can't use libpq in this program", .{});
return OpenError.NotThreadSafe;
}
var maybe_conn: ?*pq.PGconn = pq.PQconnectdb(url);
if (pq.PQstatus(maybe_conn) != pq.CONNECTION_OK) {
log.err("Postgres#open: PQstatus returned error {}: {s}", .{ pq.PQstatus(maybe_conn), pq.PQerrorMessage(maybe_conn) });
return OpenError.Failed;
}
var pg = try allocator.create(Postgres);
pg.allocator = allocator;
pg.conn = maybe_conn.?;
return Db{ .ptr = pg, .vtable = .{
.prepare = prepare,
.step = step,
.column_i64 = column_i64,
.column_slice_const_u8 = column_slice_const_u8,
.close_stmt = close_stmt,
.close_db = close_db,
} };
}
fn prepare(db: *anyopaque, query: [:0]const u8) PrepareError!*anyopaque {
var self: *Postgres = @alignCast(@ptrCast(db));
var pg_stmt = try self.allocator.create(PgStmt);
pg_stmt.* = .{
.query = query,
.params = std.ArrayList([*c]const u8).init(self.allocator),
};
return pg_stmt;
}
const PgStmt = struct {
query: [:0]const u8,
params: std.ArrayList([*c]const u8),
c_res: ?*pq.PGresult = null,
did_exec: bool = false,
n_tuples: ?c_int = null,
n_fields: ?c_int = null,
res_index: c_int = -1,
};
// TODO
// fn bind
fn step(db: *anyopaque, stmt: *anyopaque) StepError!bool {
var self: *Postgres = @alignCast(@ptrCast(db));
var pgstmt: *PgStmt = @alignCast(@ptrCast(stmt));
if (!pgstmt.did_exec) {
const params = try pgstmt.params.toOwnedSlice();
pgstmt.c_res = pq.PQexecParams(self.conn, pgstmt.query, @intCast(params.len), null, params.ptr, null, null, 0);
const rs = pq.PQresultStatus(pgstmt.c_res);
if (rs != pq.PGRES_TUPLES_OK and rs != pq.PGRES_SINGLE_TUPLE and rs != pq.PGRES_COMMAND_OK) {
log.err("PQresultStatus {} error: {s}", .{ rs, pq.PQerrorMessage(self.conn) });
return StepError.Failed;
}
pgstmt.n_tuples = pq.PQntuples(pgstmt.c_res);
pgstmt.n_fields = pq.PQnfields(pgstmt.c_res);
pgstmt.did_exec = true;
}
pgstmt.res_index = pgstmt.res_index + 1;
return pgstmt.res_index < pgstmt.n_tuples.?;
}
fn column_i64(db: *anyopaque, stmt: *anyopaque, idx: u31) ColumnError!?i64 {
_ = db;
var pgstmt: *PgStmt = @alignCast(@ptrCast(stmt));
if (!pgstmt.did_exec) @panic("did_exec == false you must call exec then step before trying column");
if (pgstmt.res_index == -1) @panic("res_index == -1 you must call exec then step before trying column");
if (pq.PQgetisnull(pgstmt.c_res, pgstmt.res_index, idx) == 1) {
return null;
}
const value_c: [*c]const u8 = pq.PQgetvalue(pgstmt.c_res, pgstmt.res_index, idx);
const slice = std.mem.sliceTo(value_c, 0);
return try std.fmt.parseInt(i64, slice, 10);
}
fn column_slice_const_u8(db: *anyopaque, stmt: *anyopaque, idx: u31) ColumnError!?[:0]const u8 {
_ = db;
var pgstmt: *PgStmt = @alignCast(@ptrCast(stmt));
if (!pgstmt.did_exec) @panic("did_exec == false you must call exec then step before trying column");
if (pgstmt.res_index == -1) @panic("res_index == -1 you must call exec then step before trying column");
if (pq.PQgetisnull(pgstmt.c_res, pgstmt.res_index, idx) == 1) {
return null;
}
const value_c: ?[*:0]u8 = pq.PQgetvalue(pgstmt.c_res, pgstmt.res_index, idx);
const value_c_nonnull = value_c orelse return null;
return std.mem.sliceTo(value_c_nonnull, 0);
}
fn close_stmt(db: *anyopaque, stmt: *anyopaque) void {
var self: *Postgres = @alignCast(@ptrCast(db));
var pgstmt: *PgStmt = @alignCast(@ptrCast(stmt));
if (pgstmt.c_res != null) {
pq.PQclear(pgstmt.c_res.?);
}
pgstmt.params.deinit();
self.allocator.destroy(pgstmt);
}
fn close_db(db: *anyopaque) void {
var self: *Postgres = @alignCast(@ptrCast(db));
pq.PQfinish(self.conn);
self.allocator.destroy(self);
}
|
