Forge
search
pythonf8555a15
1"""Load libwitdb and bind C ABI (ADR-0003)."""
2
3from __future__ import annotations
4
5import os
6import platform
7import sys
8from ctypes import (
9 CDLL,
10 POINTER,
11 c_char_p,
12 c_int32,
13 c_int64,
14 c_size_t,
15 c_uint8,
16 c_uint32,
17 c_void_p,
18 create_string_buffer,
19)
20from functools import lru_cache
21from pathlib import Path
22
23WITDB_ABI_VERSION = 1
24
25WITDB_OK = 0
26WITDB_INVALID_ARGUMENT = 1
27WITDB_NOT_FOUND = 2
28WITDB_PASSWORD_REQUIRED = 3
29WITDB_WRONG_PASSWORD = 4
30WITDB_CONFIG_MISMATCH = 5
31WITDB_UNKNOWN_PROVIDER = 6
32WITDB_TXN_NOT_SUPPORTED = 7
33WITDB_TXN_ACTIVE = 8
34WITDB_STORE_ERROR = 9
35WITDB_INVALID_HANDLE = 10
36WITDB_SQL_ERROR = 11
37
38
39class WitDbNativeLib:
40 def __init__(self, dll: CDLL) -> None:
41 self._dll = dll
42 self._configure()
43
44 def _configure(self) -> None:
45 d = self._dll
46 d.witdb_abi_version.argtypes = []
47 d.witdb_abi_version.restype = c_uint32
48
49 d.witdb_last_error_message.argtypes = []
50 d.witdb_last_error_message.restype = c_char_p
51
52 d.witdb_open.argtypes = [c_char_p, c_char_p, c_uint8, POINTER(c_size_t)]
53 d.witdb_open.restype = c_uint32
54
55 d.witdb_close.argtypes = [c_size_t]
56 d.witdb_close.restype = c_uint32
57
58 d.witdb_get.argtypes = [
59 c_size_t,
60 POINTER(c_uint8),
61 c_uint32,
62 POINTER(POINTER(c_uint8)),
63 POINTER(c_uint32),
64 ]
65 d.witdb_get.restype = c_uint32
66
67 d.witdb_put.argtypes = [
68 c_size_t,
69 POINTER(c_uint8),
70 c_uint32,
71 POINTER(c_uint8),
72 c_uint32,
73 ]
74 d.witdb_put.restype = c_uint32
75
76 d.witdb_delete.argtypes = [
77 c_size_t,
78 POINTER(c_uint8),
79 c_uint32,
80 POINTER(c_uint8),
81 ]
82 d.witdb_delete.restype = c_uint32
83
84 d.witdb_txn_begin.argtypes = [c_size_t, POINTER(c_size_t)]
85 d.witdb_txn_begin.restype = c_uint32
86
87 d.witdb_txn_commit.argtypes = [c_size_t]
88 d.witdb_txn_commit.restype = c_uint32
89
90 d.witdb_txn_rollback.argtypes = [c_size_t]
91 d.witdb_txn_rollback.restype = c_uint32
92
93 d.witdb_txn_get.argtypes = [
94 c_size_t,
95 POINTER(c_uint8),
96 c_uint32,
97 POINTER(POINTER(c_uint8)),
98 POINTER(c_uint32),
99 ]
100 d.witdb_txn_get.restype = c_uint32
101
102 d.witdb_txn_put.argtypes = [
103 c_size_t,
104 POINTER(c_uint8),
105 c_uint32,
106 POINTER(c_uint8),
107 c_uint32,
108 ]
109 d.witdb_txn_put.restype = c_uint32
110
111 d.witdb_txn_delete.argtypes = [
112 c_size_t,
113 POINTER(c_uint8),
114 c_uint32,
115 POINTER(c_uint8),
116 ]
117 d.witdb_txn_delete.restype = c_uint32
118
119 d.witdb_buffer_free.argtypes = [c_void_p]
120 d.witdb_buffer_free.restype = None
121
122 if hasattr(d, "witdb_sql_exec"):
123 d.witdb_sql_exec.argtypes = [
124 c_size_t,
125 c_char_p,
126 c_char_p,
127 POINTER(c_int64),
128 POINTER(c_int32),
129 ]
130 d.witdb_sql_exec.restype = c_uint32
131
132 d.witdb_sql_query.argtypes = [
133 c_size_t,
134 c_char_p,
135 c_char_p,
136 POINTER(POINTER(c_uint8)),
137 POINTER(c_uint32),
138 ]
139 d.witdb_sql_query.restype = c_uint32
140
141 d.witdb_sql_commit.argtypes = [c_size_t]
142 d.witdb_sql_commit.restype = c_uint32
143
144 d.witdb_sql_rollback.argtypes = [c_size_t]
145 d.witdb_sql_rollback.restype = c_uint32
146
147 @property
148 def dll(self) -> CDLL:
149 return self._dll
150
151
152def _default_search_paths() -> list[Path]:
153 root = Path(__file__).resolve().parents[4]
154 rid = _runtime_rid()
155 candidates: list[Path] = []
156 env = os.environ.get("WITDB_NATIVE_PATH")
157 if env:
158 candidates.append(Path(env))
159 for tfm in ("net10.0", "net9.0"):
160 candidates.append(
161 root
162 / "witdatabase"
163 / "Sources"
164 / "Core"
165 / "OutWit.Database.Native"
166 / "bin"
167 / "Release"
168 / tfm
169 / rid
170 / "publish"
171 / _dll_name()
172 )
173 candidates.append(root / "native" / rid / _dll_name())
174 return candidates
175
176
177def _runtime_rid() -> str:
178 system = sys.platform
179 machine = platform.machine().lower()
180 if system == "win32":
181 return "win-x64" if machine in {"amd64", "x86_64"} else f"win-{machine}"
182 if system == "darwin":
183 return "osx-arm64" if machine == "arm64" else "osx-x64"
184 return "linux-x64" if machine in {"amd64", "x86_64"} else f"linux-{machine}"
185
186
187def _dll_name() -> str:
188 if sys.platform == "win32":
189 return "witdb.dll"
190 if sys.platform == "darwin":
191 return "libwitdb.dylib"
192 return "libwitdb.so"
193
194
195@lru_cache(maxsize=1)
196def load_native_lib() -> WitDbNativeLib | None:
197 for path in _default_search_paths():
198 try:
199 if path.is_file():
200 lib = WitDbNativeLib(CDLL(str(path)))
201 elif path.is_dir():
202 full = path / _dll_name()
203 if not full.is_file():
204 continue
205 lib = WitDbNativeLib(CDLL(str(full)))
206 else:
207 continue
208 if lib.dll.witdb_abi_version() != WITDB_ABI_VERSION:
209 continue
210 return lib
211 except OSError:
212 continue
213 return None
214
215
216def bytes_to_buf(data: bytes):
217 buf = create_string_buffer(data, len(data))
218 return buf, len(data)
219
View only · write via MCP/CIDE