Nucypher
/
nucypher
mirror of https://github.com/nucypher/nucypher.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

delete datastore modules

pull/2988/head
Kieran Prasch 2022-10-26 15:37:18 +01:00
parent a895b0d72e
commit 646246ae3f
8 changed files with 0 additions and 1081 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
nucypher/datastore/__init__.py
View File

@ -1,16 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""

210
nucypher/datastore/base.py
View File

@ -1,210 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
import msgpack
from typing import Any, Callable, Iterable, NamedTuple, Union
class DBWriteError(Exception):
"""
Exception class for when db writes fail.
"""
pass
class RecordField(NamedTuple):
"""
A `RecordField` represents a field as part of a record in the datastore.
The field is given a type via `field_type`. Additionally, a `RecordField`
has three optional parameters: `encode`, `decode`, and `query_filter`.
The `field_type` is the Python type that the field should be when being
accessed from the datastore.
The optional `encode` is any callable that takes the field value (as a
`field_type`) as an argument and returns it as `bytes`. This should always
be implemented if the `field_type` is not a natively msgpack'able type.
Care should be taken to ensure that the encoded data can be decoded and
usable in _another_ language other than Python to ensure future
interoperability.
The optional `decode` is any callable that takes the unpack'd encoded
field value and returns the `field_type`. If you implement `encode`, you
will probably always want to provide a `decode`.
"""
field_type: Any
encode: Callable[[Any], bytes] = lambda field: field
decode: Callable[[bytes], Any] = lambda field: field
class DatastoreRecord:
def __new__(cls, *args, **kwargs):
# Set default class attributes for the new instance
cls.__writeable = None
cls.__storagekey = f'{cls.__name__}:{{record_field}}:{{record_id}}'
return super().__new__(cls)
def __init__(self,
db_transaction: 'lmdb.Transaction',
record_id: Union[int, str],
writeable: bool = False) -> None:
self._record_id = record_id
self.__db_transaction = db_transaction
self.__writeable = writeable
def __setattr__(self, attr: str, value: Any) -> None:
"""
This method is called when setting attributes on the class. We override
this method to serialize the value being set to the attribute, and then
we _write_ it to the database.
When `__writeable` is `None`, we only set attributes on the instance.
When `__writeable` is `False`, we raise a `TypeError`.
Finally, when `__writeable` is `True`, we get the `RecordField` for
the corresponding `attr` and check that the `value` being set is
the correct type via its `RecordField.field_type`. If the type is not
correct, we raise a `TypeError`.
If the type is correct, we then serialize it to bytes via its
`RecordField.encode` function and pack it with msgpack. Then the value
gets written to the database. If the value is unable to be written,
this will raise a `DBWriteError`.
"""
# When writeable is None (meaning, it hasn't been __init__ yet), then
# we allow any attribute to be set on the instance.
# HOT LAVA -- causes a recursion if this check isn't present.
if self.__writeable is None:
super().__setattr__(attr, value)
# Datastore records are not writeable/mutable by default, so we
# raise a TypeError in the event that writeable is False.
elif self.__writeable is False:
raise TypeError("This datastore record isn't writeable.")
# A datastore record is only mutated iff writeable is True.
elif self.__writeable is True:
record_field = self.__get_record_field(attr)
# We delete records by setting the record to `None`.
if value is None:
return self.__delete_record(attr)
if not type(value) == record_field.field_type:
raise TypeError(f'Given record is type {type(value)}; expected {record_field.field_type}')
field_value = msgpack.packb(record_field.encode(value))
self.__write_raw_record(attr, field_value)
def __getattr__(self, attr: str) -> Any:
"""
This method is called when accessing attributes that don't exist on the
class. We override this method to _read_ from the database and return
a deserialized record.
We deserialize records by calling the record's respective `RecordField.decode`
function. If the deserialized type doesn't match the type defined by
its `RecordField.field_type`, then this method will raise a `TypeError`.
"""
# Handle __getattr__ look ups for private fields
# HOT LAVA -- causes a recursion if this check isn't present.
if attr.startswith('_'):
return super().__getattr__(attr)
# Get the corresponding RecordField and retrieve the raw value from
# the db, unpack it, then use the `RecordField` to deserialize it.
record_field = self.__get_record_field(attr)
field_value = record_field.decode(msgpack.unpackb(self.__retrieve_raw_record(attr)))
if not type(field_value) == record_field.field_type:
raise TypeError(f"Decoded record was type {type(field_value)}; expected {record_field.field_type}")
return field_value
def __retrieve_raw_record(self, record_field: str) -> bytes:
"""
Retrieves a raw record, as bytes, from the database given a `record_field`.
If the record doesn't exist, this method raises an `AttributeError`.
"""
key = self.__storagekey.format(record_field=record_field, record_id=self._record_id).encode()
field_value = self.__db_transaction.get(key, default=None)
if field_value is None:
raise AttributeError(f"No {record_field} record found for ID: {self._record_id}.")
return field_value
def __write_raw_record(self, record_field: str, value: bytes) -> None:
"""
Writes a raw record, as bytes, to the database given a `record_field`
and a `value`.
If the record is unable to be written, this method raises a `DBWriteError`.
"""
key = self.__storagekey.format(record_field=record_field, record_id=self._record_id).encode()
if not self.__db_transaction.put(key, value, overwrite=True):
raise DBWriteError(f"Couldn't write the record (key: {key}) to the database.")
def __delete_record(self, record_field: str) -> None:
"""
Deletes the record from the datastore.
"""
key = self.__storagekey.format(record_field=record_field, record_id=self._record_id).encode()
if not self.__db_transaction.delete(key) and self.__db_transaction.get(key) is not None:
# We do this check to ensure that the key was actually deleted.
raise DBWriteError(f"Couldn't delete the record (key: {key}) from the database.")
def __get_record_field(self, attr: str) -> 'RecordField':
"""
Uses `getattr` to return the `RecordField` object for a given
attribute.
These objects are accessed via class attrs as `_{attribute}`. If the
`RecordField` doesn't exist for a given `attr`, then this method will
raise a `TypeError`.
"""
try:
record_field = getattr(self, f'_{attr}')
except AttributeError:
raise TypeError(f'No valid RecordField found on {self} for {attr}.')
return record_field
def __eq__(self, other):
"""
WARNING: Records are only considered unique per their record IDs in this method.
In some cases these records may have the same IDs. This comparison is useful when iterating over the
datastore _as a subset of record type_ (as we do in queries), but not
useful when comparing two records of different type.
"""
# We want to be able to compare this record to other IDs for the
# set operation in the query without instantiating a whole record.
if type(other) in (int, str):
return self._record_id == other
return self._record_id == other._record_id
def __hash__(self):
"""
WARNING: Records are only considered unique per their record IDs in this method.
In some cases these records may have the same IDs. This comparison is useful when iterating over the
datastore _as a subset of record type_ (as we do in queries), but not
useful when comparing two records of different type.
"""
return hash(self._record_id)
def delete(self):
"""
Deletes the entire record.
This works by iterating over class variables, identifying the record fields,
and then deleting them.
"""
for class_var in type(self).__dict__:
if type(type(self).__dict__[class_var]) == RecordField:
setattr(self, class_var[1:], None)

234
nucypher/datastore/datastore.py
View File

@ -1,234 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
from pathlib import Path
import lmdb
from contextlib import contextmanager, suppress
from functools import partial
from typing import Any, Callable, Generator, List, NamedTuple, Optional, Type, Union
from nucypher.datastore.base import DatastoreRecord, DBWriteError
DatastoreQueryResult = Generator[List[Type['DatastoreRecord']], None, None]
class RecordNotFound(Exception):
"""
Exception class for Datastore calls for objects that don't exist.
"""
pass
class DatastoreTransactionError(Exception):
"""
Exception class for errors during transactions in the datastore.
"""
pass
class DatastoreKey(NamedTuple):
"""
Used for managing keys when querying the datastore.
"""
record_type: Optional[str] = None
record_field: Optional[str] = None
record_id: Optional[Union[bytes, int]] = None
@classmethod
def from_bytestring(cls, key_bytestring: bytes) -> 'DatastoreKey':
key_parts = key_bytestring.decode().split(':')
with suppress(ValueError):
# If the ID can be an int, we convert it
key_parts[-1] = int(key_parts[-1])
return cls(*key_parts)
def compare_key(self, key_bytestring: bytes) -> bool:
"""
This method compares a key to another key given a key's bytestring.
Usually, the `key_bytestring` will be a query key, and the `self`
key will be a key in the `Datastore`.
The logic below offers precedence when performing matches on a query.
We _prefer_ the `other_key` over `self`.
As such, if `other_key` doesn't specify a key attr (it will be None),
we will take the key attr conferred by `self`.
Specifically, this allows us to match partial keys to specific keys,
where the `Datastore` will _always_ return specific keys, but queries
will almost always be partial keys.
"""
other_key = DatastoreKey.from_bytestring(key_bytestring)
return self.record_type == (other_key.record_type or self.record_type) and \
self.record_field == (other_key.record_field or self.record_field) and \
self.record_id == (other_key.record_id or self.record_id)
class Datastore:
"""
A persistent storage layer for arbitrary data for use by NuCypher characters.
"""
# LMDB has a `map_size` arg that caps the total size of the database.
# We can set this arbitrarily high (1TB) to prevent any run-time crashes.
LMDB_MAP_SIZE = 1_000_000_000_000
def __init__(self, db_path: Path) -> None:
"""
Initializes a Datastore object by path.
:param db_path: Filepath to a lmdb database.
"""
self.db_path = db_path
self.__db_env = lmdb.open(str(db_path), map_size=self.LMDB_MAP_SIZE)
@contextmanager
def describe(self,
record_type: Type['DatastoreRecord'],
record_id: Union[int, str],
writeable: bool = False) -> Type['DatastoreRecord']:
"""
This method is used to perform CRUD operations on the datastore within
the safety of a context manager by returning an instance of the
`record_type` identified by the `record_id` provided.
When `writeable` is `False`, the record returned by this method
cannot be used for any operations that write to the datastore. If an
attempt is made to retrieve a non-existent record whilst `writeable`
is `False`, this method raises a `RecordNotFound` error.
When `writeable` is `True`, the record can be used to perform writes
on the datastore. In the event an error occurs during the write, the
transaction will be aborted and no data will be written, and a
`DatastoreTransactionError` will be raised.
If the record is used outside the scope of the context manager, any
writes or reads will error.
"""
with suppress(ValueError):
# If the ID can be converted to an int, we do it.
record_id = int(record_id)
with self.__db_env.begin(write=writeable) as datastore_tx:
record = record_type(datastore_tx, record_id, writeable=writeable)
try:
yield record
except (AttributeError, TypeError, DBWriteError) as tx_err:
# Handle `RecordNotFound` cases when `writeable` is `False`.
if not writeable and isinstance(tx_err, AttributeError):
raise RecordNotFound(tx_err)
raise DatastoreTransactionError(f'An error was encountered during the transaction (no data was written): {tx_err}')
finally:
# Now we ensure that the record is not writeable
record.__dict__['_DatastoreRecord__writeable'] = False
@contextmanager
def query_by(self,
record_type: Type['DatastoreRecord'],
filter_func: Optional[Callable[[Union[Any, Type['DatastoreRecord']]], bool]] = None,
filter_field: str = "",
writeable: bool = False,
) -> DatastoreQueryResult:
"""
Performs a query on the datastore for the record by `record_type`.
An optional `filter_func` callable will take the decoded field
specified by the optional arg `filter_field` (see below) for the given
`record_type` iff the `filter_field` has been provided.
If no `filter_field` has been provided, then the `filter_func` will
receive a _readonly_ `DatastoreRecord`.
An optional `filter_field` can be provided as a `str` to perform a
query on a specific field for a `record_type`. This will cause the
`filter_func` to receive the decoded `filter_field` per the `record_type`.
Additionally, providing a `filter_field` will limit the query to
iterating over only the subset of records specific to that field.
If records can't be found, this method will raise `RecordNotFound`.
"""
valid_records = set()
with self.__db_env.begin(write=writeable) as datastore_tx:
db_cursor = datastore_tx.cursor()
# Set the cursor to the closest key (if it exists) by the query params.
#
# By providing a `filter_field`, the query will immediately be
# limited to the subset of keys for the `filter_field`.
query_key = f'{record_type.__name__}:{filter_field}'.encode()
if not db_cursor.set_range(query_key):
# The cursor couldn't identify any records by the key
raise RecordNotFound(f"No records exist for the key from the specified query parameters: '{query_key}'")
# Check if the record at the cursor is valid for the query
curr_key = DatastoreKey.from_bytestring(db_cursor.key())
if not curr_key.compare_key(query_key):
raise RecordNotFound(f"No records exist for the key from the specified query parameters: '{query_key}'")
# Everything checks out, let's begin iterating!
# We begin by comparing the current key to the query key.
# If the key doesn't match the query key, we know that there are
# no records for the query because lmdb orders the keys lexicographically.
# Ergo, if the current key doesn't match the query key, we know
# we have gone beyond the relevant keys and can `break` the loop.
# Additionally, if the record is already in the `valid_records`
# set (identified by the `record_id`, we call `continue`.
for db_key in db_cursor.iternext(keys=True, values=False):
curr_key = DatastoreKey.from_bytestring(db_key)
if not curr_key.compare_key(query_key):
break
elif curr_key.record_id in valid_records:
continue
record = partial(record_type, datastore_tx, curr_key.record_id)
# We pass the field to the filter_func if `filter_field` and
# `filter_func` are both provided. In the event that the
# given `filter_field` doesn't exist for the record or the
# `filter_func` returns `False`, we call `continue`.
if filter_field and filter_func:
try:
field = getattr(record(writeable=False), filter_field)
except (TypeError, AttributeError):
continue
else:
if not filter_func(field):
continue
# If only a filter_func is given, we pass a readonly record to it.
# Likewise to the above, if `filter_func` returns `False`, we
# call `continue`.
elif filter_func:
if not filter_func(record(writeable=False)):
continue
# Finally, having a record that satisfies the above conditional
# constraints, we can add the record to the set
valid_records.add(record(writeable=writeable))
# If after the iteration we have no records, we raise `RecordNotFound`
if len(valid_records) == 0:
raise RecordNotFound(f"No records exist for the key from the specified query parameters: '{query_key}'")
# We begin the context manager try/finally block
try:
# At last, we yield the queried records
yield list(valid_records)
except (AttributeError, TypeError, DBWriteError) as tx_err:
# Handle `RecordNotFound` cases when `writeable` is `False`.
if not writeable and isinstance(tx_err, AttributeError):
raise RecordNotFound(tx_err)
raise DatastoreTransactionError(f'An error was encountered during the transaction (no data was written): {tx_err}')
finally:
for record in valid_records:
record.__dict__['_DatastoreRecord__writeable'] = False

27
nucypher/datastore/models.py
View File

@ -1,27 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
from nucypher_core.umbral import PublicKey
from nucypher.datastore.base import DatastoreRecord, RecordField
class ReencryptionRequest(DatastoreRecord):
_bob_verifying_key = RecordField(
PublicKey,
encode=bytes,
decode=PublicKey.from_bytes)

49
nucypher/datastore/queries.py
View File

@ -1,49 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
import functools
from typing import Callable, List, Type
from nucypher.datastore.base import DatastoreRecord
from nucypher.datastore.datastore import Datastore, DatastoreQueryResult, RecordNotFound
from nucypher.datastore.models import ReencryptionRequest
def unwrap_records(func: Callable[..., DatastoreQueryResult]) -> Callable[..., List[Type['DatastoreRecord']]]:
"""
Used to safely unwrap results of a query.
Suitable only for reading `DatastoreRecord`s. Use `find_*` functions if you want to modify records.
Since results returned by `Datastore.query_by()` are lazy (wrapped in a `@contextmanager` generator)
we have to unwrap them and handle `RecordNotFound` error, if any. `DatastoreRecord`s are not writable
after unwrapping, because exiting `@contextmanager` is also closing `Datastore` transaction.
"""
@functools.wraps(func)
def wrapper(*args, **kwargs) -> List[Type['DatastoreRecord']]:
try:
with func(*args, **kwargs) as results:
return results
except RecordNotFound:
return []
return wrapper
@unwrap_records
def get_reencryption_requests(ds: Datastore) -> List[ReencryptionRequest]:
return ds.query_by(ReencryptionRequest)

130
tests/mock/datastore.py
View File

@ -1,130 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
from bisect import bisect_left
from contextlib import contextmanager
from pathlib import Path
import lmdb
from threading import Lock
from constant_sorrow.constants import MOCK_DB
def mock_lmdb_open(db_path: Path, map_size=10485760):
if db_path == MOCK_DB:
return MockEnvironment()
else:
return lmdb.Environment(str(db_path), map_size=map_size)
class MockEnvironment:
def __init__(self):
self._storage = {}
self._lock = Lock()
@contextmanager
def begin(self, write=False):
with self._lock:
with MockTransaction(self, write=write) as tx:
yield tx
class MockTransaction:
def __init__(self, env, write=False):
self._env = env
self._storage = dict(env._storage)
self._write = write
self._invalid = False
def __enter__(self):
if self._invalid:
raise lmdb.Error()
return self
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.abort()
else:
self.commit()
def put(self, key, value, overwrite=True):
if self._invalid:
raise lmdb.Error()
assert self._write
if not overwrite and key in self._storage:
return False
self._storage[key] = value
return True
def get(self, key, default=None):
if self._invalid:
raise lmdb.Error()
return self._storage.get(key, default)
def delete(self, key):
if self._invalid:
raise lmdb.Error()
assert self._write
if key in self._storage:
del self._storage[key]
return True
else:
return False
def commit(self):
if self._invalid:
raise lmdb.Error()
self._invalidate()
self._env._storage = self._storage
def abort(self):
self._invalidate()
self._storage = self._env._storage
def _invalidate(self):
self._invalid = True
def cursor(self):
return MockCursor(self)
class MockCursor:
def __init__(self, tx):
self._tx = tx
# TODO: assuming here that the keys are not changed while the cursor exists.
# Any way to enforce it?
self._keys = list(sorted(tx._storage))
self._pos = None
def set_range(self, key):
pos = bisect_left(self._keys, key)
if pos == len(self._keys):
self._pos = None
return False
else:
self._pos = pos
return True
def key(self):
return self._keys[self._pos]
def iternext(self, keys=True, values=True):
return iter(self._keys[self._pos:])

57
tests/unit/datastore/conftest.py
View File

@ -1,57 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
from constant_sorrow.constants import MOCK_DB
import lmdb
import pytest
import shutil
import tempfile
from nucypher.datastore import datastore
@pytest.fixture(scope='function')
def mock_or_real_datastore(request):
if request.param:
yield datastore.Datastore(MOCK_DB)
else:
temp_path = tempfile.mkdtemp()
yield datastore.Datastore(temp_path)
shutil.rmtree(temp_path)
@pytest.fixture(scope='function')
def mock_or_real_lmdb_env(request):
if request.param:
yield lmdb.open(MOCK_DB)
else:
temp_path = tempfile.mkdtemp()
yield lmdb.open(temp_path)
shutil.rmtree(temp_path)
def pytest_generate_tests(metafunc):
if 'mock_or_real_datastore' in metafunc.fixturenames:
values = [False, True]
ids = ['real_datastore', 'mock_datastore']
metafunc.parametrize('mock_or_real_datastore', values, ids=ids, indirect=True)
if 'mock_or_real_lmdb_env' in metafunc.fixturenames:
values = [False, True]
ids = ['real_lmdb_env', 'mock_lmdb_env']
metafunc.parametrize('mock_or_real_lmdb_env', values, ids=ids, indirect=True)

358
tests/unit/datastore/test_datastore.py
View File

@ -1,358 +0,0 @@
"""
This file is part of nucypher.
nucypher is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
nucypher is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with nucypher. If not, see <https://www.gnu.org/licenses/>.
"""
import lmdb
import msgpack
import pytest
import tempfile
from datetime import datetime
from nucypher.datastore import datastore
from nucypher.datastore.base import DatastoreRecord, RecordField
class TestRecord(DatastoreRecord):
__test__ = False # For pytest
_test = RecordField(bytes)
_test_date = RecordField(datetime,
encode=lambda val: datetime.isoformat(val).encode(),
decode=lambda val: datetime.fromisoformat(val.decode()))
def test_datastore_create():
temp_path = tempfile.mkdtemp()
storage = datastore.Datastore(temp_path)
assert storage.LMDB_MAP_SIZE == 1_000_000_000_000
assert storage.db_path == temp_path
assert storage._Datastore__db_env.path() == temp_path
def test_datastore_describe(mock_or_real_datastore):
storage = mock_or_real_datastore
#
# Tests for `Datastore.describe`
#
# Getting writeable access to a record can be done by setting `writeable` to `True`.
# `writeable` is, by default, `False`.
# In the event a record doesn't exist, this will raise a `RecordNotFound` error iff `writeable=False`.
with pytest.raises(datastore.RecordNotFound):
with storage.describe(TestRecord, 'test_id') as test_record:
should_error = test_record.test
# Reading a non-existent field from a writeable record is an error
with pytest.raises(datastore.DatastoreTransactionError):
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
what_is_this = test_record.test
# Writing to a, previously nonexistent record, with a valid field works!
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
test_record.test = b'test data'
assert test_record.test == b'test data'
# Check that you can't reuse the record instance to write outside the context manager
with pytest.raises(TypeError):
test_record.test = b'should not write'
# Nor can you read outside the context manager
with pytest.raises(lmdb.Error):
should_error = test_record.test
# Records can also have ints as IDs
with storage.describe(TestRecord, 1337, writeable=True) as test_record:
test_record.test = b'test int ID'
assert test_record.test == b'test int ID'
# Writing to a non-existent field errors
with pytest.raises(datastore.DatastoreTransactionError):
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
test_record.nonexistent_field = b'this will error'
# Writing the wrong type to a field errors
with pytest.raises(datastore.DatastoreTransactionError):
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
test_record.test = 1234
# Check that nothing was written
with storage.describe(TestRecord, 'test_id') as test_record:
assert test_record.test != 1234
# Any unhandled errors in the context manager results in a transaction abort
with pytest.raises(datastore.DatastoreTransactionError):
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
# Valid write
test_record.test = b'this will not persist'
# Erroneous write causing an abort
test_record.nonexistent = b'causes an error and aborts the write'
# Check that nothing was written from the aborted transaction above.
with storage.describe(TestRecord, 'test_id') as test_record:
assert test_record.test == b'test data'
# However, a handled error will not cause an abort.
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
# Valid operation
test_record.test = b'this will persist'
try:
# Maybe we don't know that this field exists or not?
# Erroneous, but handled, operation -- doesn't cause an abort
should_error = test_record.bad_read
except TypeError:
pass
# Because we handled the `TypeError`, the write persists.
with storage.describe(TestRecord, 'test_id') as test_record:
assert test_record.test == b'this will persist'
# Be aware: if you don't handle _all the errors_, then the transaction will abort:
with pytest.raises(datastore.DatastoreTransactionError):
with storage.describe(TestRecord, 'test_id', writeable=True) as test_record:
# Valid operation
test_record.test = b'this will not persist'
try:
# We handle this one correctly
# Erroneous, but handled, operation -- doesn't cause an abort
this_will_not_abort = test_record.bad_read
except TypeError:
pass
# However, we don't handle this one correctly
# Erroneous UNHANDLED operation -- causes an abort
this_WILL_abort = test_record.bad_read
# The valid operation did not persist due to the unhandled error, despite
# the other one being handled.
with storage.describe(TestRecord, 'test_id') as test_record:
assert test_record.test != b'this will not persist'
# An applicable demonstration:
# Let's imagine we don't know if a `TestRecord` identified by `new_id` exists.
# If we want to conditionally modify it, we can do as follows:
with storage.describe(TestRecord, 'new_id', writeable=True) as new_test_record:
try:
# Assume the record exists
my_test_data = new_test_record.test
# Do something with my_test_data
except AttributeError:
# We handle the case that there's no record for `new_test_record.test`
# and write to it.
new_test_record.test = b'now it exists :)'
# And proof that it worked:
with storage.describe(TestRecord, 'new_id') as new_test_record:
assert new_test_record.test == b'now it exists :)'
def test_datastore_query_by(mock_or_real_datastore):
storage = mock_or_real_datastore
# Make two test record classes
class FooRecord(DatastoreRecord):
_foo = RecordField(bytes)
class BarRecord(DatastoreRecord):
_foo = RecordField(bytes)
_bar = RecordField(bytes)
# We won't add this one
class NoRecord(DatastoreRecord):
_nothing = RecordField(bytes)
# Create them
with storage.describe(FooRecord, 1, writeable=True) as rec:
rec.foo = b'one record'
with storage.describe(FooRecord, 'two', writeable=True) as rec:
rec.foo = b'another record'
with storage.describe(FooRecord, 'three', writeable=True) as rec:
rec.foo = b'another record'
with storage.describe(BarRecord, 1, writeable=True) as rec:
rec.bar = b'one record'
with storage.describe(BarRecord, 'two', writeable=True) as rec:
rec.foo = b'foo two record'
rec.bar = b'two record'
# Let's query!
with storage.query_by(FooRecord) as records:
assert len(records) == 3
assert type(records) == list
assert records[0]._DatastoreRecord__writeable is False
assert records[1]._DatastoreRecord__writeable is False
assert records[2]._DatastoreRecord__writeable is False
# Try with BarRecord
with storage.query_by(BarRecord) as records:
assert len(records) == 2
# Try to query for non-existent records
with pytest.raises(datastore.RecordNotFound):
with storage.query_by(NoRecord) as records:
assert len(records) == 'this never gets executed cause it raises'
# Queries without writeable are read only
with pytest.raises(datastore.DatastoreTransactionError):
with storage.query_by(FooRecord) as records:
records[0].foo = b'this should error'
# Let's query by specific record and field
with storage.query_by(BarRecord, filter_field='foo') as records:
assert len(records) == 1
# Query for a non-existent field in an existing record
with pytest.raises(datastore.RecordNotFound):
with storage.query_by(FooRecord, filter_field='bar') as records:
assert len(records) == 'this never gets executed cause it raises'
# Query for a non-existent field that is _similar to an existing field_
with pytest.raises(datastore.RecordNotFound):
with storage.query_by(FooRecord, filter_field='fo') as records:
assert len(records) == 'this never gets executed cause it raises'
# Query for a field with a filtering function
# When querying with a field _and_ a filtering function, the `filter_func`
# callable is given the field value you specified.
# We throw a `isinstance` in there to ensure that the type given is a field value and not a record
filter_func = lambda field_val: not isinstance(field_val, DatastoreRecord) and field_val == b'another record'
with storage.query_by(FooRecord, filter_field='foo', filter_func=filter_func) as records:
assert len(records) == 2
assert records[0].foo == b'another record'
assert records[1].foo == b'another record'
# Query with _only_ a filter func.
# This filter_func will receive a `DatastoreRecord` instance that is readonly
filter_func = lambda field_rec: isinstance(field_rec, DatastoreRecord) and field_rec.foo == b'one record'
with storage.query_by(FooRecord, filter_func=filter_func) as records:
assert len(records) == 1
assert records[0].foo == b'one record'
# This record isn't writeable
with pytest.raises(TypeError):
records[0].foo = b'this will error'
# Make a writeable query on BarRecord
with storage.query_by(BarRecord, writeable=True) as records:
records[0].bar = b'this writes'
records[1].bar = b'this writes'
assert records[0].bar == b'this writes'
assert records[1].bar == b'this writes'
# Writeable queries on non-existant records error
with pytest.raises(datastore.RecordNotFound):
with storage.query_by(NoRecord, writeable=True) as records:
assert len(records) == 'this never gets executed'
def test_datastore_record_read(mock_or_real_lmdb_env):
db_env = mock_or_real_lmdb_env
with db_env.begin() as db_tx:
# Check the default attrs.
test_rec = TestRecord(db_tx, 'testing', writeable=False)
assert test_rec._record_id == 'testing'
assert test_rec._DatastoreRecord__db_transaction == db_tx
assert test_rec._DatastoreRecord__writeable == False
assert test_rec._DatastoreRecord__storagekey == 'TestRecord:{record_field}:{record_id}'
# Reading an attr with no RecordField should error
with pytest.raises(TypeError):
should_error = test_rec.nonexistant_field
# Reading when no records exist errors
with pytest.raises(AttributeError):
should_error = test_rec.test
# The record is not writeable
with pytest.raises(TypeError):
test_rec.test = b'should error'
def test_datastore_record_write(mock_or_real_lmdb_env):
# Test writing
db_env = mock_or_real_lmdb_env
with db_env.begin(write=True) as db_tx:
test_rec = TestRecord(db_tx, 'testing', writeable=True)
assert test_rec._DatastoreRecord__writeable == True
# Write an invalid serialization of `test` and test retrieving it is
# a TypeError
db_tx.put(b'TestRecord:test:testing', msgpack.packb(1234))
with pytest.raises(TypeError):
should_error = test_rec.test
# Writing an invalid serialization of a field is a `TypeError`
with pytest.raises(TypeError):
test_rec.test = 1234
# Test deleting a field
test_rec.test = None
with pytest.raises(AttributeError):
should_error = test_rec.test
# Test writing a valid field and getting it.
test_rec.test = b'good write'
assert test_rec.test == b'good write'
assert msgpack.unpackb(db_tx.get(b'TestRecord:test:testing')) == b'good write'
# TODO: Mock a `DBWriteError`
# Test abort
# Transaction context manager attempts to commit the transaction at `__exit__()`,
# since there were no errors caught, but it's already been aborted,
# so `lmdb.Error` is raised.
with pytest.raises(lmdb.Error):
with db_env.begin(write=True) as db_tx:
test_rec = TestRecord(db_tx, 'testing', writeable=True)
test_rec.test = b'should not be set'
db_tx.abort()
# After abort, the value should still be the one before the previous `put`
with db_env.begin() as db_tx:
test_rec = TestRecord(db_tx, 'testing', writeable=False)
assert test_rec.test == b'good write'
def test_key_tuple():
partial_key = datastore.DatastoreKey.from_bytestring(b'TestRecord:test_field')
assert partial_key.record_type == 'TestRecord'
assert partial_key.record_field == 'test_field'
assert partial_key.record_id is None
full_key = datastore.DatastoreKey.from_bytestring(b'TestRecord:test_field:test_id')
assert full_key.record_type == 'TestRecord'
assert full_key.record_field == 'test_field'
assert full_key.record_id == 'test_id'
# Full keys can match partial key strings and other full key strings
assert full_key.compare_key(b'TestRecord:test_field:test_id') is True
assert full_key.compare_key(b'TestRecord:test_field') is True
assert full_key.compare_key(b'TestRecord:') is True
assert full_key.compare_key(b'BadRecord:') is False
assert full_key.compare_key(b'BadRecord:bad_field') is False
assert full_key.compare_key(b'BadRecord:bad_field:bad_id') is False
# Partial keys can't match key strings that are more complete than themselves
assert partial_key.compare_key(b'TestRecord:test_field:test_id') is False
assert partial_key.compare_key(b'TestRecord:test_field') is True
assert partial_key.compare_key(b'TestRecord') is True
assert partial_key.compare_key(b'BadRecord') is False
assert partial_key.compare_key(b'BadRecord:bad_field') is False
assert partial_key.compare_key(b'BadRecord:bad_field:bad_id') is False
# IDs as ints
int_id_key = datastore.DatastoreKey.from_bytestring(b'TestRecord:test_field:1')
assert int_id_key.record_id == 1
assert type(int_id_key.record_id) == int