Filter Expressions
Learn about implementing filter expressions in your BDK project.
Overview
Filter expressions enable you to define filter criteria for your operations with the whose keyword. These expressions allow you to filter data based on conditions such as equality, comparisons, and more. For example:
get users from office365 whose email is "[email protected]"
Implementation
1. Include the filter_expression
parameter
filter_expression
parameterTo implement a filter expression, you need to provide the special filter_expression
parameter in your operation method definition. For example:
@procedure("to read some (*SMS* messages)")
def read_sms_messages(
self,
offset: Optional[int],
limit: Optional[int],
filter_expression: Optional[FilterExpression],
) -> List[SMSMessage]:
"""
Read some SMS messages using the Twilio API.
"""
2. Implement Visitors
The FilterExpressionVisitor
class is an abstract base class that defines methods for visiting different types of filter expressions. Each type of filter expression (binary, unary, value, noun phrase) is defined as a subclass of FilterExpression
. You will need to define a class that implements these methods to handle filter expressions. For example:
class MyFilterVisitor(FilterExpressionVisitor):
def visit_binary_expression(self, expression: FilterBinaryExpression) -> T:
# Implement logic for handling binary expressions
pass
def visit_unary_expression(self, expression: FilterUnaryExpression) -> T:
# Implement logic for handling unary expressions
pass
def visit_value(self, expression: ValueExpression) -> T:
# Implement logic for handling value expressions
pass
def visit_noun_phrases(self, expression: NounPhrasesExpression) -> T:
# Implement logic for handling noun phrases expressions
pass
Below is an example implementation of the FilterExpressionVisitor
class in the Twilio Book:
SENDER_NUMBER = NounPhrase.from_word_list(["sender", "number"])
RECIPIENT_NUMBER = NounPhrase.from_word_list(["recipient", "number"])
DATE_SENT = NounPhrase.from_word_list(["date", "sent"])
class SMSMessageFilter(FilterExpressionVisitor):
"""
Extract filtering information from the expression
"""
current_noun_phrase: Optional[NounPhrase] = None
current_value: Optional[Any] = None
recipient_number: Union[str, object] = values.unset
sender_number: Union[str, object] = values.unset
date_sent: Union[datetime, object] = values.unset
date_sent_before: Union[datetime, object] = values.unset
date_sent_after: Union[datetime, object] = values.unset
def visit_binary_expression(self, expression: FilterBinaryExpression):
expression.left.accept(self)
expression.right.accept(self)
if expression.operator == FilterBinaryOperator.EQUALS:
if self.current_noun_phrase == SENDER_NUMBER:
self.sender_number = str(self.current_value)
elif self.current_noun_phrase == RECIPIENT_NUMBER:
self.recipient_number = str(self.current_value)
elif self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent = self.current_value
elif expression.operator == FilterBinaryOperator.GREATER_THAN:
if self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent_after = self.current_value
elif expression.operator == FilterBinaryOperator.LESS_THAN:
if self.current_noun_phrase == DATE_SENT:
if not isinstance(self.current_value, datetime):
raise TypeMismatchError("date_sent", ConceptScalarType.DATETIME)
self.date_sent_before = self.current_value
elif expression.operator == FilterBinaryOperator.AND:
pass
else:
raise ValueError(f"unsupported filtering operator: {expression.operator}")
def visit_unary_expression(self, expression: FilterUnaryExpression):
pass
def visit_value(self, expression: ValueExpression):
self.current_value = expression.value
def visit_noun_phrases(self, expression: NounPhrasesExpression):
if len(expression.noun_phrases) != 1:
raise ValueError(
f"unsupported filtering noun phrase: {expression.noun_phrases}"
)
if expression.noun_phrases[0] not in [
SENDER_NUMBER,
RECIPIENT_NUMBER,
DATE_SENT,
]:
raise ValueError(
f"unsupported filtering noun phrase: {expression.noun_phrases}"
)
self.current_noun_phrase = expression.noun_phrases[0]
3. Processing Filter Expressions
Once you’ve defined your visitor class, you need to pass an an instance of it to the filter expression. This is done by calling accept onfilter_expression
. For example:
visitor = MyFilterVisitor()
filter_expression.accept(visitor)
Example
In this example, a filter expression is used in the read some SMS messages operation:
@procedure("to read some (*SMS* messages)")
def read_sms_messages(
self,
offset: Optional[int],
limit: Optional[int],
filter_expression: Optional[FilterExpression],
) -> List[SMSMessage]:
"""
Read some SMS messages using the Twilio API.
Returns:
A list of SMS messages that matches the specified filtering criteria
Example 1:
Retrieve SMS messages filtered by sender and recipient numbers
>>> read some sms messages whose sender number is "+18004445555" and whose recipient number is "+18004446666"
Example 2:
Retrieve SMS messages filtered by the date in which they were sent
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the message date
>>> read some sms messages whose date sent is the message date
Example 3:
Retrieve SMS messages that were sent in the specified time period
>>> convert "2022-03-01T15:00:00Z" to a datetime
>>> use the above as the start date
>>> convert "2022-03-03T15:00:00Z" to a datetime
>>> use the above as the end date
>>> read some sms messages whose date sent is after the start date and whose date sent is before the end date
"""
client = Client(self._account_sid, self._auth_token)
filter_visitor = SMSMessageFilter()
if filter_expression is not None:
filter_expression.accept(filter_visitor)
messages = client.messages.list(
to=filter_visitor.recipient_number,
from_=filter_visitor.sender_number,
date_sent=filter_visitor.date_sent,
date_sent_before=filter_visitor.date_sent_before,
date_sent_after=filter_visitor.date_sent_after,
)
if offset is not None:
messages = messages[offset:]
if limit is not None:
messages = messages[:limit]
return [SMSMessage.from_message_instance(message) for message in messages]
Updated about 1 month ago