###############################################################################
#
# The MIT License (MIT)
#
# Copyright (c) typedef int GmbH
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
###############################################################################
"""Transactions"""
import struct
from time import time_ns as walltime
import zlmdb.lmdb as lmdb
from typing import Optional
[docs]
class TransactionStats(object):
"""
Value class for holding transaction statistics.
"""
def __init__(self):
[docs]
self._started = walltime()
@property
[docs]
def started(self):
"""
:return: start time in ns since epoch
"""
return self._started
@property
[docs]
def duration(self):
"""
:return: duration in ns
"""
if self._started:
return walltime() - self._started
else:
return 0
[docs]
def reset(self):
"""
:return:
"""
self.puts = 0
self.dels = 0
self._started = walltime()
[docs]
class Transaction(object):
"""
Transactions in zLMDB are always run under an instance of this class.
"""
def __init__(self, db, write=False, buffers=False, stats=None):
"""
:param db:
:type db: zlmdb.Database
:param write:
:type write: bool
:param stats:
:type stats: TransactionStats
"""
[docs]
self._buffers = buffers
[docs]
self._txn: Optional[lmdb.Transaction] = None
[docs]
def __enter__(self):
assert self._txn is None
self._txn = lmdb.Transaction(
self._db._env, write=self._write, buffers=self._buffers
)
return self
[docs]
def __exit__(self, exc_type, exc_value, traceback):
assert self._txn is not None
# https://docs.python.org/3/reference/datamodel.html#object.__exit__
# If the context was exited without an exception, all three arguments will be None.
if exc_type is None:
if self._log:
cnt = 0
for op, key in self._log:
_key = struct.pack(">H", 0)
_data = struct.pack(">H", op) + key
self._txn.put(_key, _data)
cnt += 1
self._txn.commit()
else:
self._txn.abort()
self._txn = None
[docs]
def id(self):
"""
:return:
"""
assert self._txn is not None
return self._txn.id()
[docs]
def get(self, key):
"""
:param key:
:return:
"""
assert self._txn is not None
return self._txn.get(key)
[docs]
def put(self, key, data, overwrite=True):
"""
:param key:
:param data:
:param overwrite:
:return:
"""
assert self._txn is not None
# store the record, returning True if it was written, or False to indicate the key
# was already present and overwrite=False.
was_written = self._txn.put(key, data, overwrite=overwrite)
if was_written:
if self._stats:
self._stats.puts += 1
if self._log:
self._log.append((Transaction.PUT, key))
return was_written
[docs]
def delete(self, key):
"""
:param key:
:return:
"""
assert self._txn is not None
was_deleted = self._txn.delete(key)
if was_deleted:
if self._stats:
self._stats.dels += 1
if self._log:
self._log.append((Transaction.DEL, key))
return was_deleted