"""Contains the querying interface.Starting with :class:`~tinydb.queries.Query` you can construct complexqueries:>>> ((where('f1') == 5) & (where('f2') != 2)) | where('s').matches(r'^\\w+$')(('f1' == 5) and ('f2' != 2)) or ('s' ~= ^\\w+$ )Queries are executed by using the ``__call__``:>>> q = where('val') == 5>>> q({'val': 5})True>>> q({'val': 1})False"""importreimportsysfromtypingimportMapping,Tuple,Callable,Any,Union,List,Optionalfrom.utilsimportfreezeifsys.version_info>=(3,8):fromtypingimportProtocolelse:fromtyping_extensionsimportProtocol__all__=('Query','QueryLike','where')defis_sequence(obj):returnhasattr(obj,'__iter__')classQueryLike(Protocol):"""    A typing protocol that acts like a query.    Something that we use as a query must have two properties:    1. It must be callable, accepting a `Mapping` object and returning a       boolean that indicates whether the value matches the query, and    2. it must have a stable hash that will be used for query caching.    In addition, to mark a query as non-cacheable (e.g. if it involves    some remote lookup) it needs to have a method called ``is_cacheable``    that returns ``False``.    This query protocol is used to make MyPy correctly support the query    pattern that TinyDB uses.    See also https://mypy.readthedocs.io/en/stable/protocols.html#simple-user-defined-protocols    """def__call__(self,value:Mapping)->bool:...def__hash__(self)->int:...
[docs]classQueryInstance:"""    A query instance.    This is the object on which the actual query operations are performed. The    :class:`~tinydb.queries.Query` class acts like a query builder and    generates :class:`~tinydb.queries.QueryInstance` objects which will    evaluate their query against a given document when called.    Query instances can be combined using logical OR and AND and inverted using    logical NOT.    In order to be usable in a query cache, a query needs to have a stable hash    value with the same query always returning the same hash. That way a query    instance can be used as a key in a dictionary.    """
[docs]def__init__(self,test:Callable[[Mapping],bool],hashval:Optional[Tuple]):self._test=testself._hash=hashval
defis_cacheable(self)->bool:returnself._hashisnotNone
[docs]def__call__(self,value:Mapping)->bool:"""        Evaluate the query to check if it matches a specified value.        :param value: The value to check.        :return: Whether the value matches this query.        """returnself._test(value)
[docs]def__hash__(self)->int:# We calculate the query hash by using the ``hashval`` object which# describes this query uniquely, so we can calculate a stable hash# value by simply hashing itreturnhash(self._hash)
[docs]def__repr__(self):return'QueryImpl{}'.format(self._hash)
[docs]def__eq__(self,other:object):ifisinstance(other,QueryInstance):returnself._hash==other._hashreturnFalse
# --- Query modifiers -----------------------------------------------------def__and__(self,other:'QueryInstance')->'QueryInstance':# We use a frozenset for the hash as the AND operation is commutative# (a & b == b & a) and the frozenset does not consider the order of# elementsifself.is_cacheable()andother.is_cacheable():hashval=('and',frozenset([self._hash,other._hash]))else:hashval=NonereturnQueryInstance(lambdavalue:self(value)andother(value),hashval)def__or__(self,other:'QueryInstance')->'QueryInstance':# We use a frozenset for the hash as the OR operation is commutative# (a | b == b | a) and the frozenset does not consider the order of# elementsifself.is_cacheable()andother.is_cacheable():hashval=('or',frozenset([self._hash,other._hash]))else:hashval=NonereturnQueryInstance(lambdavalue:self(value)orother(value),hashval)def__invert__(self)->'QueryInstance':hashval=('not',self._hash)ifself.is_cacheable()elseNonereturnQueryInstance(lambdavalue:notself(value),hashval)
[docs]classQuery(QueryInstance):"""    TinyDB Queries.    Allows building queries for TinyDB databases. There are two main ways of    using queries:    1) ORM-like usage:    >>> User = Query()    >>> db.search(User.name == 'John Doe')    >>> db.search(User['logged-in'] == True)    2) Classical usage:    >>> db.search(where('value') == True)    Note that ``where(...)`` is a shorthand for ``Query(...)`` allowing for    a more fluent syntax.    Besides the methods documented here you can combine queries using the    binary AND and OR operators:    >>> # Binary AND:    >>> db.search((where('field1').exists()) & (where('field2') == 5))    >>> # Binary OR:    >>> db.search((where('field1').exists()) | (where('field2') == 5))    Queries are executed by calling the resulting object. They expect to get    the document to test as the first argument and return ``True`` or    ``False`` depending on whether the documents match the query or not.    """
[docs]def__init__(self)->None:# The current path of fields to access when evaluating the objectself._path:Tuple[Union[str,Callable],...]=()# Prevent empty queries to be evaluateddefnotest(_):raiseRuntimeError('Empty query was evaluated')super().__init__(test=notest,hashval=(None,))
[docs]def__repr__(self):return'{}()'.format(type(self).__name__)
[docs]def__hash__(self):returnsuper().__hash__()
def__getattr__(self,item:str):# Generate a new query object with the new query path# We use type(self) to get the class of the current query in case# someone uses a subclass of ``Query``query=type(self)()# Now we add the accessed item to the query path ...query._path=self._path+(item,)# ... and update the query hashquery._hash=('path',query._path)ifself.is_cacheable()elseNonereturnquerydef__getitem__(self,item:str):# A different syntax for ``__getattr__``# We cannot call ``getattr(item)`` here as it would try to resolve# the name as a method name first, only then call our ``__getattr__``# method. By calling ``__getattr__`` directly, we make sure that# calling e.g. ``Query()['test']`` will always generate a query for a# document's ``test`` field instead of returning a reference to the# ``Query.test`` methodreturnself.__getattr__(item)def_generate_test(self,test:Callable[[Any],bool],hashval:Tuple,allow_empty_path:bool=False)->QueryInstance:"""        Generate a query based on a test function that first resolves the query        path.        :param test: The test the query executes.        :param hashval: The hash of the query.        :return: A :class:`~tinydb.queries.QueryInstance` object        """ifnotself._pathandnotallow_empty_path:raiseValueError('Query has no path')defrunner(value):try:# Resolve the pathforpartinself._path:ifisinstance(part,str):value=value[part]else:value=part(value)except(KeyError,TypeError):returnFalseelse:# Perform the specified testreturntest(value)returnQueryInstance(lambdavalue:runner(value),(hashvalifself.is_cacheable()elseNone))
[docs]def__eq__(self,rhs:Any):"""        Test a dict value for equality.        >>> Query().f1 == 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value==rhs,('==',self._path,freeze(rhs)))
[docs]def__ne__(self,rhs:Any):"""        Test a dict value for inequality.        >>> Query().f1 != 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value!=rhs,('!=',self._path,freeze(rhs)))
[docs]def__lt__(self,rhs:Any)->QueryInstance:"""        Test a dict value for being lower than another value.        >>> Query().f1 < 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value<rhs,('<',self._path,rhs))
[docs]def__le__(self,rhs:Any)->QueryInstance:"""        Test a dict value for being lower than or equal to another value.        >>> where('f1') <= 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value<=rhs,('<=',self._path,rhs))
[docs]def__gt__(self,rhs:Any)->QueryInstance:"""        Test a dict value for being greater than another value.        >>> Query().f1 > 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value>rhs,('>',self._path,rhs))
[docs]def__ge__(self,rhs:Any)->QueryInstance:"""        Test a dict value for being greater than or equal to another value.        >>> Query().f1 >= 42        :param rhs: The value to compare against        """returnself._generate_test(lambdavalue:value>=rhs,('>=',self._path,rhs))
[docs]defexists(self)->QueryInstance:"""        Test for a dict where a provided key exists.        >>> Query().f1.exists()        """returnself._generate_test(lambda_:True,('exists',self._path))
[docs]defmatches(self,regex:str,flags:int=0)->QueryInstance:"""        Run a regex test against a dict value (whole string has to match).        >>> Query().f1.matches(r'^\\w+$')        :param regex: The regular expression to use for matching        :param flags: regex flags to pass to ``re.match``        """deftest(value):ifnotisinstance(value,str):returnFalsereturnre.match(regex,value,flags)isnotNonereturnself._generate_test(test,('matches',self._path,regex))
[docs]defsearch(self,regex:str,flags:int=0)->QueryInstance:"""        Run a regex test against a dict value (only substring string has to        match).        >>> Query().f1.search(r'^\\w+$')        :param regex: The regular expression to use for matching        :param flags: regex flags to pass to ``re.match``        """deftest(value):ifnotisinstance(value,str):returnFalsereturnre.search(regex,value,flags)isnotNonereturnself._generate_test(test,('search',self._path,regex))
[docs]deftest(self,func:Callable[[Mapping],bool],*args)->QueryInstance:"""        Run a user-defined test function against a dict value.        >>> def test_func(val):        ...     return val == 42        ...        >>> Query().f1.test(test_func)        .. warning::            The test function provided needs to be deterministic (returning the            same value when provided with the same arguments), otherwise this            may mess up the query cache that :class:`~tinydb.table.Table`            implements.        :param func: The function to call, passing the dict as the first                     argument        :param args: Additional arguments to pass to the test function        """returnself._generate_test(lambdavalue:func(value,*args),('test',self._path,func,args))
[docs]defany(self,cond:Union[QueryInstance,List[Any]])->QueryInstance:"""        Check if a condition is met by any document in a list,        where a condition can also be a sequence (e.g. list).        >>> Query().f1.any(Query().f2 == 1)        Matches::            {'f1': [{'f2': 1}, {'f2': 0}]}        >>> Query().f1.any([1, 2, 3])        Matches::            {'f1': [1, 2]}            {'f1': [3, 4, 5]}        :param cond: Either a query that at least one document has to match or                     a list of which at least one document has to be contained                     in the tested document.        """ifcallable(cond):deftest(value):returnis_sequence(value)andany(cond(e)foreinvalue)else:deftest(value):returnis_sequence(value)andany(eincondforeinvalue)returnself._generate_test(lambdavalue:test(value),('any',self._path,freeze(cond)))
[docs]defall(self,cond:Union['QueryInstance',List[Any]])->QueryInstance:"""        Check if a condition is met by all documents in a list,        where a condition can also be a sequence (e.g. list).        >>> Query().f1.all(Query().f2 == 1)        Matches::            {'f1': [{'f2': 1}, {'f2': 1}]}        >>> Query().f1.all([1, 2, 3])        Matches::            {'f1': [1, 2, 3, 4, 5]}        :param cond: Either a query that all documents have to match or a list                     which has to be contained in the tested document.        """ifcallable(cond):deftest(value):returnis_sequence(value)andall(cond(e)foreinvalue)else:deftest(value):returnis_sequence(value)andall(einvalueforeincond)returnself._generate_test(lambdavalue:test(value),('all',self._path,freeze(cond)))
[docs]defone_of(self,items:List[Any])->QueryInstance:"""        Check if the value is contained in a list or generator.        >>> Query().f1.one_of(['value 1', 'value 2'])        :param items: The list of items to check with        """returnself._generate_test(lambdavalue:valueinitems,('one_of',self._path,freeze(items)))
deffragment(self,document:Mapping)->QueryInstance:deftest(value):forkeyindocument:ifkeynotinvalueorvalue[key]!=document[key]:returnFalsereturnTruereturnself._generate_test(lambdavalue:test(value),('fragment',freeze(document)),allow_empty_path=True)
[docs]defnoop(self)->QueryInstance:"""        Always evaluate to ``True``.        Useful for having a base value when composing queries dynamically.        """returnQueryInstance(lambdavalue:True,())
[docs]defmap(self,fn:Callable[[Any],Any])->'Query':"""        Add a function to the query path. Similar to __getattr__ but for        arbitrary functions.        """query=type(self)()# Now we add the callable to the query path ...query._path=self._path+(fn,)# ... and kill the hash - callable objects can be mutable, so it's# harmful to cache their results.query._hash=Nonereturnquery
defwhere(key:str)->Query:"""    A shorthand for ``Query()[key]``    """returnQuery()[key]