pgmq-python: explicitly cast types in send_batch

tembo-pgmq-python/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "tembo-pgmq-python"
-version = "0.9.1"
+version = "0.9.2"
 description = "Python client for the PGMQ Postgres extension."
 authors = ["Adam Hendel <[email protected]>"]
 license = "PostgreSQL"

tembo-pgmq-python/tembo_pgmq_python/async_queue.py

@@ -186,11 +186,11 @@ async def send_batch(self, queue: str, messages: List[dict], delay: int = 0, con
         else:
             return await self._send_batch_internal(queue, messages, delay, conn)
 
-    async def _send_batch_internal(self, queue, messages, delay, conn):
+    async def _send_batch_internal(self, queue: str, messages: List[dict], delay: int, conn):
         self.logger.debug(f"Sending batch of messages to queue '{queue}' with delay={delay}")
         jsonb_array = [dumps(message).decode("utf-8") for message in messages]
         result = await conn.fetch(
-            "SELECT * FROM pgmq.send_batch($1, $2::jsonb[], $3);",
+            "SELECT * FROM pgmq.send_batch($1::string, $2::jsonb[], $3::integer);",
             queue,
             jsonb_array,
             delay,

tembo-pgmq-python/tembo_pgmq_python/queue.py

@@ -124,7 +124,7 @@ def send(self, queue: str, message: dict, delay: int = 0, conn=None) -> int:
     def send_batch(self, queue: str, messages: List[dict], delay: int = 0, conn=None) -> List[int]:
         """Send a batch of messages to a queue."""
         self.logger.debug(f"send_batch called with conn: {conn}")
-        query = "select * from pgmq.send_batch(%s, %s, %s);"
+        query = "select * from pgmq.send_batch(%s::text, %s::jsonb[], %s::integer);"
         params = [queue, [Jsonb(message) for message in messages], delay]
         result = self._execute_query_with_result(query, params, conn=conn)
         return [message[0] for message in result]