implement SQL Filter (#80)

Gsantomaggio · Copilot · web-flow · commit 42048b037990 · 2025-09-15T09:51:27.000+02:00
This PR implements SQL filter functionality for RabbitMQ AMQP streams, allowing users to filter messages using SQL-like expressions. The implementation adds support for SQL filters in the stream consumer options and includes comprehensive test coverage and documentation. Added SQL filter support to StreamConsumerOptions and StreamFilterOptions classes Created comprehensive test cases to verify SQL filter functionality Updated documentation with a new example demonstrating SQL filter usage closes: #73 --------- Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
diff --git a/.ci/ubuntu/gha-setup.sh b/.ci/ubuntu/gha-setup.sh
@@ -7,7 +7,7 @@ set -o xtrace
 script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
-readonly rabbitmq_image=rabbitmq:4.1.0-management
+readonly rabbitmq_image=rabbitmq:4.2-rc-management
 
 
 readonly docker_name_prefix='rabbitmq-amqp-python-client'
diff --git a/examples/streams_with_filters/example_streams_with_filters.py b/examples/streams_with_filters/example_streams_with_filters.py
@@ -72,7 +72,7 @@ def main() -> None:
     See: https://www.rabbitmq.com/docs/next/stream-filtering#stage-2-amqp-filter-expressions
     """
 
-    queue_name = "stream-example-with_filtering-queue"
+    queue_name = "stream-example-with-message-properties-filter-queue"
     logger.info("Creating connection")
     environment = Environment("amqp://guest:guest@localhost:5672/")
     connection = create_connection(environment)
diff --git a/examples/streams_with_sql_filters/example_streams_with_sql_filters.py b/examples/streams_with_sql_filters/example_streams_with_sql_filters.py
@@ -0,0 +1,144 @@
+# type: ignore
+import logging
+
+from rabbitmq_amqp_python_client import (
+    AddressHelper,
+    AMQPMessagingHandler,
+    Connection,
+    ConnectionClosed,
+    Converter,
+    Environment,
+    Event,
+    Message,
+    OffsetSpecification,
+    StreamConsumerOptions,
+    StreamFilterOptions,
+    StreamSpecification,
+)
+
+MESSAGES_TO_PUBLISH = 100
+
+
+class MyMessageHandler(AMQPMessagingHandler):
+
+    def __init__(self):
+        super().__init__()
+        self._count = 0
+
+    def on_amqp_message(self, event: Event):
+        # only messages with banana filters and with subject yellow
+        # and application property from = italy get received
+        self._count = self._count + 1
+        logger.info(
+            "Received message: {}, subject {} application properties {} .[Total Consumed: {}]".format(
+                Converter.bytes_to_string(event.message.body),
+                event.message.subject,
+                event.message.application_properties,
+                self._count,
+            )
+        )
+        self.delivery_context.accept(event)
+
+    def on_connection_closed(self, event: Event):
+        # if you want you can add cleanup operations here
+        print("connection closed")
+
+    def on_link_closed(self, event: Event) -> None:
+        # if you want you can add cleanup operations here
+        print("link closed")
+
+
+def create_connection(environment: Environment) -> Connection:
+    connection = environment.connection()
+    connection.dial()
+
+    return connection
+
+
+logging.basicConfig()
+logger = logging.getLogger("[streams_with_filters]")
+logger.setLevel(logging.INFO)
+
+
+def main() -> None:
+    """
+    In this example we create a stream queue and a consumer with SQL filter
+
+    See: https://www.rabbitmq.com/docs/next/stream-filtering#stage-2-amqp-filter-expressions
+    """
+
+    queue_name = "stream-example-with-sql-filter-queue"
+    logger.info("Creating connection")
+    environment = Environment("amqp://guest:guest@localhost:5672/")
+    connection = create_connection(environment)
+    management = connection.management()
+    # delete the queue if it exists
+    management.delete_queue(queue_name)
+    # create a stream queue
+    management.declare_queue(StreamSpecification(name=queue_name))
+
+    addr_queue = AddressHelper.queue_address(queue_name)
+
+    consumer_connection = create_connection(environment)
+    sql = (
+        "properties.subject LIKE '%in_the_filter%' "
+        "AND a_in_the_filter_key = 'a_in_the_filter_value'"
+    )
+
+    consumer = consumer_connection.consumer(
+        addr_queue,
+        message_handler=MyMessageHandler(),
+        stream_consumer_options=StreamConsumerOptions(
+            offset_specification=OffsetSpecification.first,
+            filter_options=StreamFilterOptions(sql=sql),
+        ),
+    )
+    print(
+        "create a consumer and consume the test message - press control + c to terminate to consume"
+    )
+
+    # print("create a publisher and publish a test message")
+    publisher = connection.publisher(addr_queue)
+
+    # publish messages won't match the filter
+    for i in range(MESSAGES_TO_PUBLISH):
+        publisher.publish(Message(Converter.string_to_bytes(body="apple: " + str(i))))
+
+    # publish messages that will match the filter
+    for i in range(MESSAGES_TO_PUBLISH):
+        msqMatch = Message(
+            body=Converter.string_to_bytes("the_right_one_sql"),
+            # will match due of %
+            subject="something_in_the_filter_{}".format(i),
+            application_properties={"a_in_the_filter_key": "a_in_the_filter_value"},
+        )
+        publisher.publish(msqMatch)
+
+    publisher.close()
+
+    while True:
+        try:
+            consumer.run()
+        except KeyboardInterrupt:
+            pass
+        except ConnectionClosed:
+            print("connection closed")
+            continue
+        except Exception as e:
+            print("consumer exited for exception " + str(e))
+
+        break
+
+    #
+    logger.info("consumer exited, deleting queue")
+    management.delete_queue(queue_name)
+
+    print("closing connections")
+    management.close()
+    print("after management closing")
+    environment.close()
+    print("after connection closing")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/rabbitmq_amqp_python_client/entities.py b/rabbitmq_amqp_python_client/entities.py
@@ -7,6 +7,8 @@
 from .exceptions import ValidationCodeException
 from .qpid.proton._data import Described, symbol
 
+SQL_FILTER = "sql-filter"
+AMQP_SQL_FILTER = "amqp:sql-filter"
 STREAM_FILTER_SPEC = "rabbitmq:stream-filter"
 STREAM_OFFSET_SPEC = "rabbitmq:stream-offset-spec"
 STREAM_FILTER_MATCH_UNFILTERED = "rabbitmq:stream-match-unfiltered"
@@ -159,7 +161,6 @@ class MessageProperties:
     Attributes:
         message_id: Uniquely identifies a message within the system (int, UUID, bytes, or str).
         user_id: Identity of the user responsible for producing the message.
-        to: Intended destination node of the message.
         subject: Summary information about the message content and purpose.
         reply_to: Address of the node to send replies to.
         correlation_id: Client-specific id for marking or identifying messages (int, UUID, bytes, or str).
@@ -174,7 +175,6 @@ class MessageProperties:
 
     message_id: Optional[Union[int, str, bytes]] = None
     user_id: Optional[bytes] = None
-    to: Optional[str] = None
     subject: Optional[str] = None
     reply_to: Optional[str] = None
     correlation_id: Optional[Union[int, str, bytes]] = None
@@ -245,20 +245,24 @@ def __init__(
         if offset_specification is not None:
             self._offset(offset_specification)
 
-        if filter_options is not None and filter_options.values is not None:
+        if filter_options is None:
+            return
+
+        if filter_options.values is not None:
             self._filter_values(filter_options.values)
 
-        if filter_options is not None and filter_options.match_unfiltered:
+        if filter_options.match_unfiltered:
             self._filter_match_unfiltered(filter_options.match_unfiltered)
 
-        if filter_options is not None and filter_options.message_properties is not None:
+        if filter_options.message_properties is not None:
             self._filter_message_properties(filter_options.message_properties)
-        if (
-            filter_options is not None
-            and filter_options.application_properties is not None
-        ):
+
+        if filter_options.application_properties is not None:
             self._filter_application_properties(filter_options.application_properties)
 
+        if filter_options.sql is not None and filter_options.sql != "":
+            self._filter_sql(filter_options.sql)
+
     def _offset(self, offset_specification: Union[OffsetSpecification, int]) -> None:
         """
         Set the offset specification for the stream.
@@ -334,6 +338,15 @@ def _filter_application_properties(
                     Described(symbol(AMQP_APPLICATION_PROPERTIES_FILTER), app_prop)
                 )
 
+    def _filter_sql(self, sql: str) -> None:
+        """
+        Set SQL filter for the stream.
+
+        Args:
+            sql: SQL string to apply as a filter
+        """
+        self._filter_set[symbol(SQL_FILTER)] = Described(symbol(AMQP_SQL_FILTER), sql)
+
     def filter_set(self) -> Dict[symbol, Described]:
         """
         Get the current filter set configuration.
diff --git a/tests/test_streams.py b/tests/test_streams.py
