Repository navigation
Expand file tree
/
Copy pathstore_options.py
More file actions
136 lines (94 loc) · 5.2 KB
/
Copy pathstore_options.py
File metadata and controls
136 lines (94 loc) · 5.2 KB
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
127
128
129
130
131
132
133
134
135
136
from objectbox.c import *
from objectbox.model.model import Model
class StoreOptions:
""" A RAII wrapper to the C API for setting the store options. """
_c_handle: Optional[int]
def __init__(self):
self._c_handle = obx_opt()
def _free(self):
if self._c_handle is not None:
obx_opt_free(self._c_handle)
self._c_handle = None
def directory(self, path: str):
obx_opt_directory(self._c_handle, c_str(path))
def max_db_size_in_kb(self, size_in_kb: int):
obx_opt_max_db_size_in_kb(self._c_handle, size_in_kb)
def max_data_size_in_kb(self, size_in_kb: int):
obx_opt_max_data_size_in_kb(self._c_handle, size_in_kb)
def file_mode(self, file_mode: int):
obx_opt_file_mode(self._c_handle, file_mode)
def max_readers(self, max_readers: int):
obx_opt_max_readers(self._c_handle, max_readers)
def no_reader_thread_locals(self, flag: bool):
obx_opt_no_reader_thread_locals(self._c_handle, flag)
def model(self, model: Model):
model.validate_ids_assigned()
c_model = model._create_c_model()
obx_opt_model(self._c_handle, c_model)
def model_bytes(self, bytes_: bytes):
obx_opt_model_bytes(self._c_handle, bytes_, len(bytes_))
def model_bytes_direct(self, bytes_: bytes):
# The bytes are not copied by the C library; keep them alive until the store is opened (options are freed then)
self._model_bytes_direct = bytes_
obx_opt_model_bytes_direct(self._c_handle, bytes_, len(bytes_))
def validate_on_open_pages(self, page_limit: int, flags: int):
raise NotImplementedError # TODO
def validate_on_open_kv(self, flags: OBXValidateOnOpenKvFlags):
raise NotImplementedError # TODO
def put_padding_mode(self, mode: OBXPutPaddingMode):
obx_opt_put_padding_mode(self._c_handle, mode)
def read_schema(self, value: bool):
obx_opt_read_schema(self._c_handle, value)
def use_previous_commit(self, value: bool):
obx_opt_use_previous_commit(self._c_handle, value)
def read_only(self, value: bool):
obx_opt_read_only(self._c_handle, value)
def debug_flags(self, flags: OBXDebugFlags):
obx_opt_debug_flags(self._c_handle, flags)
def add_debug_flags(self, flags: OBXDebugFlags):
obx_opt_add_debug_flags(self._c_handle, flags)
def async_max_queue_length(self, value: int):
obx_opt_async_max_queue_length(self._c_handle, value)
def async_throttle_at_queue_length(self, value: int):
obx_opt_async_throttle_at_queue_length(self._c_handle, value)
def async_throttle_micros(self, value: int):
obx_opt_async_throttle_micros(self._c_handle, value)
def async_max_in_tx_duration(self, micros: int):
obx_opt_async_max_in_tx_duration(self._c_handle, micros)
def async_max_in_tx_operations(self, value: int):
obx_opt_async_max_in_tx_operations(self._c_handle, value)
def async_pre_txn_delay(self, delay_micros: int):
obx_opt_async_pre_txn_delay(self._c_handle, delay_micros)
def async_pre_txn_delay4(self, delay_micros: int, delay2_micros: int, min_queue_length_for_delay2: int):
obx_opt_async_pre_txn_delay4(self._c_handle, delay_micros, delay2_micros, min_queue_length_for_delay2)
def async_post_txn_delay(self, delay_micros: int):
obx_opt_async_post_txn_delay(self._c_handle, delay_micros)
def async_post_txn_delay5(self, delay_micros: int, delay2_micros: int, min_queue_length_for_delay2: int,
subtract_processing_time: bool):
obx_opt_async_post_txn_delay5(self._c_handle, delay_micros, delay2_micros, min_queue_length_for_delay2,
subtract_processing_time)
def async_minor_refill_threshold(self, queue_length: int):
obx_opt_async_minor_refill_threshold(self._c_handle, queue_length)
def async_minor_refill_max_count(self, value: int):
obx_opt_async_minor_refill_max_count(self._c_handle, value)
def async_max_tx_pool_size(self, value: int):
obx_opt_async_max_tx_pool_size(self._c_handle, value)
def async_object_bytes_max_cache_size(self, value: int):
obx_opt_async_object_bytes_max_cache_size(self._c_handle, value)
def async_object_bytes_max_size_to_cache(self, value: int):
obx_opt_async_object_bytes_max_size_to_cache(self._c_handle, value)
def log_callback(self, value: Callable[[OBXLogLevel, str],None]):
self._c_log_cb = obx_log_callback_fn(lambda level, message, size, userdata: value(level, message.decode('utf-8')))
obx_opt_log_callback(self._c_handle, self._c_log_cb, None)
return self
def backup_restore(self, backup_file: str, flags: OBXBackupRestoreFlags):
raise NotImplementedError # TODO
def get_directory(self) -> str:
dir_bytes = obx_opt_get_directory(self._c_handle)
return dir_bytes.decode('utf-8')
def get_max_db_size_in_kb(self) -> int:
return obx_opt_get_max_db_size_in_kb(self._c_handle)
def get_max_data_size_in_kb(self) -> int:
return obx_opt_get_max_data_size_in_kb(self._c_handle)
def get_debug_flags(self) -> OBXDebugFlags:
return obx_opt_get_debug_flags(self._c_handle)