Source code for langchain_community.vectorstores.redis.filters
fromenumimportEnumfromfunctoolsimportwrapsfromtypingimportAny,Callable,Dict,List,Optional,Set,Tuple,Unionfromlangchain_community.utilities.redisimportTokenEscaper# disable mypy error for dunder method overrides# mypy: disable-error-code="override"
[docs]classRedisFilterOperator(Enum):"""RedisFilterOperator enumerator is used to create RedisFilterExpressions."""EQ=1NE=2LT=3GT=4LE=5GE=6OR=7AND=8LIKE=9IN=10
[docs]classRedisFilter:"""Collection of RedisFilterFields."""
def_set_value(self,val:Any,val_type:Tuple[Any],operator:RedisFilterOperator)->None:# check that the operator is supported by this classifoperatornotinself.OPERATORS:raiseValueError(f"Operator {operator} not supported by {self.__class__.__name__}. "+f"Supported operators are {self.OPERATORS.values()}.")ifnotisinstance(val,val_type):raiseTypeError(f"Right side argument passed to operator {self.OPERATORS[operator]} "f"with left side "f"argument {self.__class__.__name__} must be of type {val_type}, "f"received value {val}")self._value=valself._operator=operator
[docs]defcheck_operator_misuse(func:Callable)->Callable:"""Decorator to check for misuse of equality operators."""@wraps(func)defwrapper(instance:Any,*args:Any,**kwargs:Any)->Any:# Extracting 'other' from positional arguments or keyword argumentsother=kwargs.get("other")if"other"inkwargselseNoneifnotother:forarginargs:ifisinstance(arg,type(instance)):other=argbreakifisinstance(other,type(instance)):raiseValueError("Equality operators are overridden for FilterExpression creation. Use "".equals() for equality checks")returnfunc(instance,*args,**kwargs)returnwrapper
[docs]classRedisTag(RedisFilterField):"""RedisFilterField representing a tag in a Redis index."""OPERATORS:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:"==",RedisFilterOperator.NE:"!=",RedisFilterOperator.IN:"==",}OPERATOR_MAP:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:"@%s:{%s}",RedisFilterOperator.NE:"(-@%s:{%s})",RedisFilterOperator.IN:"@%s:{%s}",}SUPPORTED_VAL_TYPES=(list,set,tuple,str,type(None))
[docs]def__init__(self,field:str):"""Create a RedisTag FilterField. Args: field (str): The name of the RedisTag field in the index to be queried against. """super().__init__(field)
def_set_tag_value(self,other:Union[List[str],Set[str],Tuple[str],str],operator:RedisFilterOperator,)->None:ifisinstance(other,(list,set,tuple)):try:# "if val" clause removes non-truthy values from listother=[str(val)forvalinotherifval]exceptValueError:raiseValueError("All tags within collection must be strings")# above to catch the "" caseelifnotother:other=[]elifisinstance(other,str):other=[other]self._set_value(other,self.SUPPORTED_VAL_TYPES,operator)# type: ignore@check_operator_misusedef__eq__(self,other:Union[List[str],Set[str],Tuple[str],str])->"RedisFilterExpression":"""Create a RedisTag equality filter expression. Args: other (Union[List[str], Set[str], Tuple[str], str]): The tag(s) to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisTag >>> filter = RedisTag("brand") == "nike" """self._set_tag_value(other,RedisFilterOperator.EQ)returnRedisFilterExpression(str(self))@check_operator_misusedef__ne__(self,other:Union[List[str],Set[str],Tuple[str],str])->"RedisFilterExpression":"""Create a RedisTag inequality filter expression. Args: other (Union[List[str], Set[str], Tuple[str], str]): The tag(s) to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisTag >>> filter = RedisTag("brand") != "nike" """self._set_tag_value(other,RedisFilterOperator.NE)returnRedisFilterExpression(str(self))@propertydef_formatted_tag_value(self)->str:return"|".join([self.escaper.escape(tag)fortaginself._value])def__str__(self)->str:"""Return the query syntax for a RedisTag filter expression."""ifnotself._value:return"*"returnself.OPERATOR_MAP[self._operator]%(self._field,self._formatted_tag_value,)
[docs]classRedisNum(RedisFilterField):"""RedisFilterField representing a numeric field in a Redis index."""OPERATORS:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:"==",RedisFilterOperator.NE:"!=",RedisFilterOperator.LT:"<",RedisFilterOperator.GT:">",RedisFilterOperator.LE:"<=",RedisFilterOperator.GE:">=",}OPERATOR_MAP:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:"@%s:[%s%s]",RedisFilterOperator.NE:"(-@%s:[%s%s])",RedisFilterOperator.GT:"@%s:[(%s +inf]",RedisFilterOperator.LT:"@%s:[-inf (%s]",RedisFilterOperator.GE:"@%s:[%s +inf]",RedisFilterOperator.LE:"@%s:[-inf %s]",}SUPPORTED_VAL_TYPES=(int,float,type(None))def__str__(self)->str:"""Return the query syntax for a RedisNum filter expression."""ifself._valueisNone:return"*"if(self._operator==RedisFilterOperator.EQorself._operator==RedisFilterOperator.NE):returnself.OPERATOR_MAP[self._operator]%(self._field,self._value,self._value,)else:returnself.OPERATOR_MAP[self._operator]%(self._field,self._value)@check_operator_misusedef__eq__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric equality filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("zipcode") == 90210 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.EQ)# type: ignorereturnRedisFilterExpression(str(self))@check_operator_misusedef__ne__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric inequality filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("zipcode") != 90210 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.NE)# type: ignorereturnRedisFilterExpression(str(self))def__gt__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric greater than filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("age") > 18 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.GT)# type: ignorereturnRedisFilterExpression(str(self))def__lt__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric less than filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("age") < 18 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.LT)# type: ignorereturnRedisFilterExpression(str(self))def__ge__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric greater than or equal to filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("age") >= 18 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.GE)# type: ignorereturnRedisFilterExpression(str(self))def__le__(self,other:Union[int,float])->"RedisFilterExpression":"""Create a Numeric less than or equal to filter expression. Args: other (Union[int, float]): The value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisNum >>> filter = RedisNum("age") <= 18 """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.LE)# type: ignorereturnRedisFilterExpression(str(self))
[docs]classRedisText(RedisFilterField):"""RedisFilterField representing a text field in a Redis index."""OPERATORS:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:"==",RedisFilterOperator.NE:"!=",RedisFilterOperator.LIKE:"%",}OPERATOR_MAP:Dict[RedisFilterOperator,str]={RedisFilterOperator.EQ:'@%s:("%s")',RedisFilterOperator.NE:'(-@%s:"%s")',RedisFilterOperator.LIKE:"@%s:(%s)",}SUPPORTED_VAL_TYPES=(str,type(None))@check_operator_misusedef__eq__(self,other:str)->"RedisFilterExpression":"""Create a RedisText equality (exact match) filter expression. Args: other (str): The text value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisText >>> filter = RedisText("job") == "engineer" """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.EQ)# type: ignorereturnRedisFilterExpression(str(self))@check_operator_misusedef__ne__(self,other:str)->"RedisFilterExpression":"""Create a RedisText inequality filter expression. Args: other (str): The text value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisText >>> filter = RedisText("job") != "engineer" """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.NE)# type: ignorereturnRedisFilterExpression(str(self))def__mod__(self,other:str)->"RedisFilterExpression":"""Create a RedisText "LIKE" filter expression. Args: other (str): The text value to filter on. Example: >>> from langchain_community.vectorstores.redis import RedisText >>> filter = RedisText("job") % "engine*" # suffix wild card match >>> filter = RedisText("job") % "%%engine%%" # fuzzy match w/ LD >>> filter = RedisText("job") % "engineer|doctor" # contains either term >>> filter = RedisText("job") % "engineer doctor" # contains both terms """self._set_value(other,self.SUPPORTED_VAL_TYPES,RedisFilterOperator.LIKE)# type: ignorereturnRedisFilterExpression(str(self))def__str__(self)->str:"""Return the query syntax for a RedisText filter expression."""ifnotself._value:return"*"returnself.OPERATOR_MAP[self._operator]%(self._field,self._value,)
[docs]classRedisFilterExpression:"""Logical expression of RedisFilterFields. RedisFilterExpressions can be combined using the & and | operators to create complex logical expressions that evaluate to the Redis Query language. This presents an interface by which users can create complex queries without having to know the Redis Query language. Filter expressions are not initialized directly. Instead they are built by combining RedisFilterFields using the & and | operators. Examples: >>> from langchain_community.vectorstores.redis import RedisTag, RedisNum >>> brand_is_nike = RedisTag("brand") == "nike" >>> price_is_under_100 = RedisNum("price") < 100 >>> filter = brand_is_nike & price_is_under_100 >>> print(str(filter)) (@brand:{nike} @price:[-inf (100)]) """
def__str__(self)->str:# top level check that allows recursive calls to __str__ifnotself._filterandnotself._operator:raiseValueError("Improperly initialized RedisFilterExpression")# if there's an operator, combine expressions accordinglyifself._operator:ifnotisinstance(self._left,RedisFilterExpression)ornotisinstance(self._right,RedisFilterExpression):raiseTypeError("Improper combination of filters.""Both left and right should be type FilterExpression")operator_str=" | "ifself._operator==RedisFilterOperator.ORelse" "returnself.format_expression(self._left,self._right,operator_str)# check that base case, the filter is setifnotself._filter:raiseValueError("Improperly initialized RedisFilterExpression")returnself._filter