"""
arango with a simple (dict-like or list-like) interface
"""
import re
from dol.base import Persister
from pyArango.connection import Connection
from pyArango.theExceptions import DocumentNotFoundError
# ArangoDB's document-key alphabet, minus ``%``. pyArango places keys in request
# URLs without encoding them and the server percent-decodes the path, so allowing
# ``%`` would let encoded ``/`` or ``..`` address other collections or databases.
# Do not add ``%`` back to match ArangoDB's documented alphabet.
_VALID_KEY_CHARS = re.compile(r"[A-Za-z0-9_\-:.@()+,=;$!*']+")
_MAX_KEY_BYTES = 254
_DOT_SEGMENTS = {".", ".."}
[docs]
def validate_document_key(key):
"""Return ``key`` if it is a safe ArangoDB document key, else raise ``KeyError``.
Keys end up in request URLs, so anything outside ArangoDB's key alphabet (such
as ``/``) or a dot segment could address a different collection or database.
>>> validate_document_key('Robot::0a1b')
'Robot::0a1b'
>>> validate_document_key('a/b')
Traceback (most recent call last):
...
KeyError: "Invalid ArangoDB document key: 'a/b'"
>>> validate_document_key('..')
Traceback (most recent call last):
...
KeyError: "Invalid ArangoDB document key: '..'"
"""
if (
not isinstance(key, str)
or key in _DOT_SEGMENTS
or len(key.encode("utf-8")) > _MAX_KEY_BYTES
or not _VALID_KEY_CHARS.fullmatch(key)
):
raise KeyError(f"Invalid ArangoDB document key: {key!r}")
return key
[docs]
class ArangoDbPersister(Persister):
"""
A basic ArangoDB persister.
>>> from py2store.persisters.arangodb_w_pyarango import ArangoDbPersister
>>> s = ArangoDbPersister()
>>> k = {'key': '777'} # Each collection will happily accept user-defined _key values.
>>> v = {'val': 'bar'}
>>> for _key in s:
... del s[_key]
...
>>> k in s
False
>>> len(s)
0
>>> s[k] = v
>>> len(s)
1
>>> s[k]
{'val': 'bar'}
>>> s.get(k)
{'val': 'bar'}
>>> s.get({'not': 'a key'}, {'default': 'val'}) # testing s.get with default
{'default': 'val'}
>>> list(s.values())
[{'val': 'bar'}]
>>> k in s # testing __contains__ again
True
>>> del s[k]
>>> len(s)
0
>>> s = ArangoDbPersister(db_name='py2store', key_fields=('name',))
>>> for _key in s:
... del s[_key]
...
>>> s[{'name': 'guido'}] = {'yob': 1956, 'proj': 'python', 'bdfl': False}
>>> s[{'name': 'guido'}]
{'yob': 1956, 'proj': 'python', 'bdfl': False}
>>> s[{'name': 'vitalik'}] = {'yob': 1994, 'proj': 'ethereum', 'bdfl': True}
>>> s[{'name': 'vitalik'}]
{'yob': 1994, 'proj': 'ethereum', 'bdfl': True}
>>> for key, val in s.items():
... print(f"{key}: {val}")
{'name': 'guido'}: {'yob': 1956, 'proj': 'python', 'bdfl': False}
{'name': 'vitalik'}: {'yob': 1994, 'proj': 'ethereum', 'bdfl': True}
"""
# reserved by the database fields
_reserved = {"_key", "_id", "_rev"}
def __init__(
self,
user="root",
password="root",
url="http://127.0.0.1:8529",
db_name="py2store",
collection_name="test",
key_fields=("key",), # _id, _key and _rev are reserved by db
key_fields_separator="::",
):
self._connection = Connection(
arangoURL=url,
username=user,
password=password,
)
self._db_name = db_name
self._collection_name = collection_name
# If DB not created:
if not self._connection.hasDatabase(self._db_name):
self._connection.createDatabase(self._db_name)
self._adb = self._connection[self._db_name]
# If collection not created:
if not self._adb.hasCollection(self._collection_name):
self._collection = self._adb.createCollection(name=self._collection_name)
self._collection = self._adb[self._collection_name]
if isinstance(key_fields, str):
key_fields = (key_fields,)
self._key_fields = key_fields
self._key_fields_separator = key_fields_separator
def _make_key(self, keys_dict):
"""
Convert a dict of keys into a real key-string by joining dict values in a predefined order.
DB requirements for the key:
The key must be a string value.
Keys are case-sensitive.
Numeric keys are not allowed.
The key must be from 1 byte to 254 bytes long.
It must consist of:
- letters a-z (lower or upper case),
- digits 0-9
- any of the following characters: _ - : . @ ( ) + , = ; $ ! * ' %
Any other characters cannot be used inside key values.
"""
key_values = [keys_dict[key_label] for key_label in self._key_fields]
key_str = self._key_fields_separator.join(key_values)
return validate_document_key(key_str)
def _split_key(self, joined_key_str):
"""
Convert a key-string used by DB internally
into a user-friendly dict of key labels and values.
"""
key_values = joined_key_str.split(self._key_fields_separator)
keys_dict = dict(zip(self._key_fields, key_values))
return keys_dict
def __fetchitem__(self, keys_dict):
key = self._make_key(keys_dict)
try:
return self._collection[key]
except DocumentNotFoundError:
raise KeyError(f"No document found for query: {keys_dict}")
def __getitem__(self, keys_dict):
item = self.__fetchitem__(keys_dict)
doc = item.getStore()
# todo (Mike): maybe move this cleanup to a base Arango Store?
# exclude reserved keys and corresponded values
data = {
key: doc[key]
for key in doc
if key not in self._reserved and key not in self._key_fields
}
return data
def __setitem__(self, keys_dict, values_dict):
key = self._make_key(keys_dict) # validate before touching the backend
try:
doc = self.__fetchitem__(keys_dict)
except KeyError:
doc = self._collection.createDocument()
doc._key = key
for k, v in values_dict.items():
doc[k] = v
doc.save()
def __delitem__(self, keys_dict):
doc = self.__fetchitem__(keys_dict)
doc.delete()
def __iter__(self):
docs = self._collection.fetchAll()
yield from (
{key_name: doc[key_name] for key_name in self._key_fields} for doc in docs
)
def __len__(self):
return self._collection.count()
#######################################################################################################################
# Stores
from functools import wraps
from dol.base import Store
from dol.util import lazyprop
[docs]
class ArangoDbStore(Store):
@wraps(ArangoDbPersister.__init__)
def __init__(self, *args, **kwargs):
persister = ArangoDbPersister(*args, **kwargs)
super().__init__(persister)
[docs]
class ArangoDbTupleKeyStore(ArangoDbStore):
"""
ArangoDbStore using tuple keys.
>>> from py2store.stores.arangodb_store import ArangoDbTupleKeyStore
>>> s = ArangoDbTupleKeyStore(collection_name='test', key_fields=('key', 'user'))
>>> k = (1234, 'user')
>>> v = {'name': 'bob', 'age': 42}
>>> if k in s: # deleting all docs in tmp
... del s[k]
>>> assert (k in s) == False # see that key is not in store (and testing __contains__)
>>> orig_length = len(s)
>>> s[k] = v
>>> assert len(s) == orig_length + 1
>>> assert k in list(s)
>>> assert s[k] == v
>>> assert s.get(k) == v
>>> assert v in list(s.values())
>>> assert (k in s) == True # testing __contains__ again
>>> del s[k]
>>> assert len(s) == orig_length
"""
@lazyprop
def _key_fields(self):
return self.store._key_fields
def _id_of_key(self, k):
return {field: field_val for field, field_val in zip(self._key_fields, k)}
def _key_of_id(self, _id):
return tuple(_id[x] for x in self._key_fields)