Migrate amqp library (#194)

* migrate amqp library * few tests/modifications * formatting * removing uamqp library * fixing issue on data encoding * fixing tests/ isort and black checker * removing useless files * fixing examples * reverting utils.py * fixing doc
rabbitmq-community · May 28, 2024 · 3e24496 · 3e24496
1 parent 65b7a22
commit 3e24496
Show file tree

Hide file tree

Showing 27 changed files with 2,810 additions and 85 deletions.
diff --git a/README.md b/README.md
@@ -56,7 +56,7 @@ The client supports two codecs to store the messages to the server:
 By default you should use `AMQP 1.0` codec:
 ```python
    amqp_message = AMQPMessage(
-    body="hello: {}".format(i),
+    body=bytes("hello: {}".format(i), "utf-8"),
   )
 ```
 
@@ -287,7 +287,7 @@ This one:
 ```python
  for i in range(1_000_000):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
             )
             # send is asynchronous
             await producer.send(stream=STREAM, message=amqp_message)

diff --git a/docs/examples/basic_producers/producer_send.py b/docs/examples/basic_producers/producer_send.py
@@ -17,7 +17,7 @@ async def publish():
 
         for i in range(MESSAGES):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
             )
             # send is asynchronous
             await producer.send(stream=STREAM, message=amqp_message)

diff --git a/docs/examples/basic_producers/producer_send_batch.py b/docs/examples/basic_producers/producer_send_batch.py
@@ -20,7 +20,7 @@ async def publish():
             messages = []
             for i in range(BATCH):
                 amqp_message = AMQPMessage(
-                    body="hello: {}".format(i),
+                    body=bytes("hello: {}".format(i), "utf-8"),
                 )
                 messages.append(amqp_message)
             # send_batch is synchronous. will wait till termination

diff --git a/docs/examples/basic_producers/producer_send_wait.py b/docs/examples/basic_producers/producer_send_wait.py
@@ -14,7 +14,7 @@ async def publish():
         # sending a thousand of messages in AMQP format
         for i in range(MESSAGES):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
             )
             # send is synchronous. It will also wait synchronously for the confirmation to arrive from the server
             # it is really very slow and send() + callback for asynchronous confirmation should be used instead.

diff --git a/docs/examples/filtering/producer_filtering.py b/docs/examples/filtering/producer_filtering.py
@@ -26,7 +26,7 @@ async def publish():
                 "region": "New York",
             }
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties=application_properties,
             )
             # send is asynchronous
@@ -43,7 +43,7 @@ async def publish():
                 "region": "California",
             }
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties=application_properties,
             )
             # send is asynchronous

diff --git a/docs/examples/filtering/super_stream_producer_filtering.py b/docs/examples/filtering/super_stream_producer_filtering.py
@@ -36,7 +36,7 @@ async def publish():
         for i in range(MESSAGES):
             application_properties = {"region": "New York", "id": "{}".format(i)}
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties=application_properties,
             )
             # send is asynchronous
@@ -51,7 +51,7 @@ async def publish():
         for i in range(MESSAGES):
             application_properties = {"region": "California", "id": "{}".format(i)}
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties=application_properties,
             )
             # send is asynchronous

diff --git a/docs/examples/producers_with_confirmations/send_batch_with_confirmation.py b/docs/examples/producers_with_confirmations/send_batch_with_confirmation.py
@@ -34,7 +34,7 @@ async def publish():
             messages = []
             for i in range(BATCH):
                 amqp_message = AMQPMessage(
-                    body="hello: {}".format(i),
+                    body=bytes("hello: {}".format(i), "utf-8"),
                 )
                 messages.append(amqp_message)
             # send_batch is synchronous. will wait till termination

diff --git a/docs/examples/producers_with_confirmations/send_with_confirmation.py b/docs/examples/producers_with_confirmations/send_with_confirmation.py
@@ -32,7 +32,7 @@ async def publish():
         # sending a million of messages in AMQP format
         for i in range(MESSAGES):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
             )
             # send is asynchronous - also confirmation is taken asynchronously by _on_publish_confirm_client callback
             # you can specify different callbacks for different messages.

diff --git a/docs/examples/reliable_client/BestPracticesClient.py b/docs/examples/reliable_client/BestPracticesClient.py
@@ -217,7 +217,7 @@ async def publish(rabbitmq_configuration: dict):
             return
 
         amqp_message = AMQPMessage(
-            body="hello: {}".format(i),
+            body=bytes("hello: {}".format(i), "utf-8"),
             application_properties={"id": "{}".format(i)},
         )
         # send is asynchronous

diff --git a/docs/examples/single_active_consumer/sac_super_stream_producer.py b/docs/examples/single_active_consumer/sac_super_stream_producer.py
@@ -38,7 +38,7 @@ async def publish():
         # run slowly several messages in order to test with sac
         for i in range(1000000):
             amqp_message = AMQPMessage(
-                body="message_:{}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 properties=uamqp.message.MessageProperties(message_id=i),
             )
             await producer.send(message=amqp_message)

diff --git a/docs/examples/sub_entry_batch/producer_sub_entry_batch.py b/docs/examples/sub_entry_batch/producer_sub_entry_batch.py
@@ -19,7 +19,7 @@ async def publish():
             messages = []
             for i in range(BATCH):
                 amqp_message = AMQPMessage(
-                    body="a:{}".format(i),
+                    body=bytes("hello: {}".format(i), "utf-8"),
                 )
                 messages.append(amqp_message)
 

diff --git a/docs/examples/super_stream/super_stream_producer.py b/docs/examples/super_stream/super_stream_producer.py
@@ -36,7 +36,7 @@ async def publish():
         start_time = time.perf_counter()
         for i in range(MESSAGES):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties={"id": "{}".format(i)},
             )
             await super_stream_producer.send(amqp_message)

diff --git a/docs/examples/super_stream/super_stream_producer_key.py b/docs/examples/super_stream/super_stream_producer_key.py
@@ -36,7 +36,7 @@ async def publish():
         # Sending a million messages
         for i in range(MESSAGES):
             amqp_message = AMQPMessage(
-                body="hello: {}".format(i),
+                body=bytes("hello: {}".format(i), "utf-8"),
                 application_properties={"id": "{}".format(i)},
             )
             await super_stream_producer.send(amqp_message)

diff --git a/docs/examples/tls/producer.py b/docs/examples/tls/producer.py
@@ -26,7 +26,7 @@ async def publish():
             messages = []
             for i in range(BATCH):
                 amqp_message = AMQPMessage(
-                    body="hello: {}".format(i),
+                    body=bytes("hello: {}".format(i), "utf-8"),
                 )
                 messages.append(amqp_message)
             # send_batch is synchronous. will wait till termination

diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "rstream"
-version = "0.19.1"
+version = "0.20.0"
 description = "A python client for RabbitMQ Streams"
 authors = ["George Fortunatov <[email protected]>", "Daniele Palaia <[email protected]>"]
 readme = "README.md"
@@ -10,7 +10,6 @@ license = "MIT"
 
 [tool.poetry.dependencies]
 python = "^3.9"
-uamqp = "^1.6.3"
 requests = "^2.31.0"
 mmh3 = "^4.0.0"
 
@@ -24,11 +23,13 @@ pytest-asyncio = "^0.15.1"
 black = "^23.12.1"
 requests = "^2.31.0"
 mmh3 = "^4.0.0"
+typing_extensions ="^4.11.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.0"
 requests = "^2.31.0"
 types-requests = "^2.31.0.20240406"
+typing_extensions ="^4.11.0"
 
 [tool.black]
 line-length = 110

diff --git a/rstream/__init__.py b/rstream/__init__.py
@@ -19,6 +19,8 @@
 del metadata
 
 from .amqp import AMQPMessage, amqp_decoder  # noqa: E402
+from ._pyamqp.message import Properties  # noqa: E402
+from ._pyamqp.message import Header  # noqa: E402
 from .compression import CompressionType  # noqa: E402
 from .constants import (  # noqa: E402
     ConsumerOffsetSpecification,
@@ -56,6 +58,8 @@
     "Consumer",
     "RawMessage",
     "Producer",
+    "Properties",
+    "Header",
     "OffsetType",
     "ConsumerOffsetSpecification",
     "ConfirmationStatus",