Writing Cloud-Agnostic Queue Integration Code with Kiro

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Iterator

@dataclass
class Message:
body: bytes
headers: dict[str, str] = field(default_factory=dict)

class QueueInterface(ABC):
@abstractmethod
def publish(self, queue_name: str, message: Message) -\> None:
"""Publish a message to the named queue."""

@abstractmethod
def subscribe(self, queue_name: str) -\> Iterator[Message]:
"""Yield messages from the named queue in FIFO order."""

from typing import Iterator

import pika

from .interface import Message, QueueInterface

class RabbitMqAdapter(QueueInterface):
def __init__(self, url: str) -\> None:
if not url:
raise ValueError("url must be a non-empty AMQP URL")
self._params = pika.URLParameters(url)

def publish(self, queue_name: str, message: Message) -\> None:
conn = pika.BlockingConnection(self._params)
try:
ch = conn.channel()
ch.queue_declare(queue=queue_name, durable=True)
ch.basic_publish(
exchange="",
routing_key=queue_name,
body=message.body,
properties=pika.BasicProperties(headers=dict(message.headers)),
)
finally:
conn.close()

def subscribe(self, queue_name: str) -\> Iterator[Message]:
conn = pika.BlockingConnection(self._params)
ch = conn.channel()
ch.queue_declare(queue=queue_name, durable=True)
try:
for method, props, body in ch.consume(queue_name, inactivity_timeout=5.0):
if method is None: # no message within the timeout
return
yield Message(body=body, headers=dict((props.headers or {}) if props else {}))
ch.basic_ack(method.delivery_tag)
finally:
try:
ch.cancel()
finally:
conn.close()

import json

from my_queues import Message, QueueInterface

class OrderProducer:
def __init__(self, queue: QueueInterface) -\> None:
self._queue = queue

def order_placed(self, order_id: str, total: float) -\> None:
body = json.dumps({"order_id": order_id, "total": total}).encode()
self._queue.publish("orders", Message(body=body, headers={"type": "order.placed"}))

class OrderConsumer:
def __init__(self, queue: QueueInterface) -\> None:
self._queue = queue

def process(self) -\> list[str]:
processed = []
for msg in self._queue.subscribe("orders"):
if msg.headers.get("type") == "order.placed":
processed.append(json.loads(msg.body)["order_id"])
return processed

import copy
import threading
from collections import defaultdict, deque
from typing import Iterator

from .interface import Message, QueueInterface

class InMemoryAdapter(QueueInterface):
def __init__(self) -\> None:
self._queues: dict[str, deque[Message]] = defaultdict(deque)
self._lock = threading.Lock()

def publish(self, queue_name: str, message: Message) -\> None:
with self._lock:
self._queues[queue_name].append(copy.deepcopy(message))

def subscribe(self, queue_name: str) -\> Iterator[Message]:
while True:
with self._lock:
if not self._queues[queue_name]:
return
yield self._queues[queue_name].popleft()

FORBIDDEN_PREFIXES = ("azure.servicebus", "google.cloud.pubsub", "pika")

violations = []
for py_path in PACKAGE_ROOT.rglob("*.py"):
if py_path.name in EXEMPT_FILENAMES: # the adapters and the factory
continue
tree = ast.parse(py_path.read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Import):
names = [a.name for a in node.names]
elif isinstance(node, ast.ImportFrom) and node.module:
names = [node.module]
else:
continue
violations += [f"{py_path}: {n}" for n in names if n.startswith(FORBIDDEN_PREFIXES)]

assert not violations, "Use create_queue() instead of importing a broker SDK."