Source code for tinydb.utils

"""Utility functions."""fromcollectionsimportOrderedDict,abcfromtypingimportList,Iterator,TypeVar,Generic,Union,Optional,Type, \TYPE_CHECKINGK=TypeVar('K')V=TypeVar('V')D=TypeVar('D')T=TypeVar('T')__all__=('LRUCache','freeze','with_typehint')defwith_typehint(baseclass:Type[T]):"""    Add type hints from a specified class to a base class:    >>> class Foo(with_typehint(Bar)):    ...     pass    This would add type hints from class ``Bar`` to class ``Foo``.    Note that while PyCharm and Pyright (for VS Code) understand this pattern,    MyPy does not. For that reason TinyDB has a MyPy plugin in    ``mypy_plugin.py`` that adds support for this pattern.    """ifTYPE_CHECKING:# In the case of type checking: pretend that the target class inherits# from the specified base classreturnbaseclass# Otherwise: just inherit from `object` like a regular Python classreturnobject
[docs]classLRUCache(abc.MutableMapping,Generic[K,V]):""" A least-recently used (LRU) cache with a fixed cache size. This class acts as a dictionary but has a limited size. If the number of entries in the cache exceeds the cache size, the least-recently accessed entry will be discarded. This is implemented using an ``OrderedDict``. On every access the accessed entry is moved to the front by re-inserting it into the ``OrderedDict``. When adding an entry and the cache size is exceeded, the last entry will be discarded. """
[docs]def__init__(self,capacity=None)->None:self.capacity=capacityself.cache:OrderedDict[K,V]=OrderedDict()
@propertydeflru(self)->List[K]:returnlist(self.cache.keys())@propertydeflength(self)->int:returnlen(self.cache)
[docs]defclear(self)->None:self.cache.clear()
def__len__(self)->int:returnself.lengthdef__contains__(self,key:object)->bool:returnkeyinself.cachedef__setitem__(self,key:K,value:V)->None:self.set(key,value)def__delitem__(self,key:K)->None:delself.cache[key]def__getitem__(self,key)->V:value=self.get(key)ifvalueisNone:raiseKeyError(key)returnvaluedef__iter__(self)->Iterator[K]:returniter(self.cache)
[docs]defget(self,key:K,default:Optional[D]=None)->Optional[Union[V,D]]:value=self.cache.get(key)ifvalueisnotNone:self.cache.move_to_end(key,last=True)returnvaluereturndefault
defset(self,key:K,value:V):ifself.cache.get(key):self.cache.move_to_end(key,last=True)else:self.cache[key]=value# Check, if the cache is full and we have to remove old items# If the queue is of unlimited size, self.capacity is NaN and# x > NaN is always False in Python and the cache won't be cleared.ifself.capacityisnotNoneandself.length>self.capacity:self.cache.popitem(last=False)
classFrozenDict(dict):""" An immutable dictionary. This is used to generate stable hashes for queries that contain dicts. Usually, Python dicts are not hashable because they are mutable. This class removes the mutability and implements the ``__hash__`` method. """def__hash__(self):# Calculate the has by hashing a tuple of all dict itemsreturnhash(tuple(sorted(self.items())))def_immutable(self,*args,**kws):raiseTypeError('object is immutable')# Disable write access to the dict__setitem__=_immutable__delitem__=_immutableclear=_immutablesetdefault=_immutable# type: ignorepopitem=_immutabledefupdate(self,e=None,**f):raiseTypeError('object is immutable')defpop(self,k,d=None):raiseTypeError('object is immutable')deffreeze(obj):""" Freeze an object by making it immutable and thus hashable. """ifisinstance(obj,dict):# Transform dicts into ``FrozenDict``sreturnFrozenDict((k,freeze(v))fork,vinobj.items())elifisinstance(obj,list):# Transform lists into tuplesreturntuple(freeze(el)forelinobj)elifisinstance(obj,set):# Transform sets into ``frozenset``sreturnfrozenset(obj)else:# Don't handle all other objectsreturnobj