Python AMQP Client Library
Project description
AMQP Client Python
Client with high level of abstraction for manipulation of messages in the event bus RabbitMQ.
Features:
- Automatic creation and management of queues, exchanges and channels;
- Connection persistence and auto reconnect;
- Support for direct, topic and fanout exchanges;
- Publish;
- Subscribe;
- Support for a Remote procedure call (RPC).
Examples:
you can use sync , async eventbus and sync wrapper of async eventbus
async usage
# basic configuration
from amqp_client_python import (
AsyncEventbusRabbitMQ,
Config, Options
)
from amqp_client_python.event import IntegrationEvent, IntegrationEventHandler
config = Config(Options("queue", "rpc_queue", "rpc_exchange"))
eventbus = AsyncEventbusRabbitMQ(config)
# publish
class ExampleEvent(IntegrationEvent):
EVENT_NAME: str = "ExampleEvent"
def __init__(self, event_type: str, message = []) -> None:
super().__init__(self.EVENT_NAME, event_type)
self.message = message
publish_event = ExampleEvent(rpc_exchange, ["message"])
eventbus.publish(publish_event, rpc_routing_key, "direct")
# subscribe
class ExampleEventHandler(IntegrationEventHandler):
def handle(self, body) -> None:
print(body) # handle messages
await eventbus.subscribe(subscribe_event, subscribe_event_handle, rpc_routing_key)
# rpc_publish
response = await eventbus.rpc_client(rpc_exchange, "user.find", ["content_message"])
# provider
async def handle2(*body) -> bytes:
print(f"body: {body}")
return b"content"
await eventbus.provide_resource("user.find", handle)
sync usage
from amqp_client_python import (
EventbusRabbitMQ,
Config, Options
)
from amqp_client_python.event import IntegrationEvent, IntegrationEventHandler
from examples.default import queue, rpc_queue, rpc_exchange, rpc_routing_key
class ExampleEvent(IntegrationEvent):
EVENT_NAME: str = "ExampleEvent"
ROUTING_KEY: str = rpc_routing_key
def __init__(self, event_type: str, message = []) -> None:
super().__init__(self.EVENT_NAME, event_type)
self.message = message
self.routing_key = self.ROUTING_KEY
class ExampleEventHandler(IntegrationEventHandler):
def handle(self, body) -> None:
print(body,"subscribe")
config = Config(Options(queue, rpc_queue, rpc_exchange))
eventbus = EventbusRabbitMQ(config=config)
class ExampleEvent(IntegrationEvent):
EVENT_NAME: str = "ExampleEvent"
def __init__(self, event_type: str, message = []) -> None:
super().__init__(self.EVENT_NAME, event_type)
self.message = message
from time import sleep
from random import randint
def handle(*body):
print(body[0], "rpc_provider")
return f"{body[0]}".encode("utf-8")
subscribe_event = ExampleEvent(rpc_exchange)
publish_event = ExampleEvent(rpc_exchange, ["message"])
subscribe_event_handle = ExampleEventHandler()
eventbus.subscribe(subscribe_event, subscribe_event_handle, rpc_routing_key)
eventbus.provide_resource(rpc_routing_key+"2", handle)
count = 0
running = True
from concurrent.futures import TimeoutError
while running:
try:
count += 1
if str(count) != eventbus.rpc_client(rpc_exchange, rpc_routing_key+"2", [f"{count}"]).decode("utf-8"):
running = False
#eventbus.publish(publish_event, rpc_routing_key, "direct")
#running = False
except TimeoutError as err:
print("timeout!!!: ", str(err))
except KeyboardInterrupt:
running=False
except BaseException as err:
print("Err:", err)
sync wrapper usage
from amqp_client_python import EventbusWrapperRabbitMQ, Config, Options
from amqp_client_python.event import IntegrationEvent, IntegrationEventHandler
config = Config(Options(queue, rpc_queue, rpc_exchange))
eventbus = EventbusWrapperRabbitMQ(config=config)
class ExampleEvent(IntegrationEvent):
EVENT_NAME: str = "ExampleEvent"
def __init__(self, event_type: str, message = []) -> None:
super().__init__(self.EVENT_NAME, event_type)
self.message = message
class ExampleEventHandler(IntegrationEventHandler):
async def handle(self, body) -> None:
print(body,"subscribe")
async def handle(*body):
print(body[0], "rpc_provider")
return f"{body[0]}".encode("utf-8")
subscribe_event = ExampleEvent(rpc_exchange)
publish_event = ExampleEvent(rpc_exchange, ["message"])
subscribe_event_handle = ExampleEventHandler()
# rpc_provider
eventbus.provide_resource(rpc_routing_key+"2", handle).result()
# subscribe
eventbus.subscribe(subscribe_event, subscribe_event_handle, rpc_routing_key).result()
count = 0
running = True
while running:
try:
count += 1
# rpc_client call
eventbus.rpc_client(rpc_exchange, rpc_routing_key+"2", [f"{count}"]).result().decode("utf-8")
# publish
eventbus.publish(publish_event, rpc_routing_key, "direct").result()
#running = False
except KeyboardInterrupt:
running=False
except BaseException as err:
print("Err:", err)
Know Limitations:
sync eventbus
Cannot use rpc call when inside of rpc provider and subscribe handlers
#/obs: works on sync eventbus wrapper
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
amqp_client_python-0.1.10.tar.gz
(24.6 kB
view hashes)
Built Distribution
Close
Hashes for amqp_client_python-0.1.10.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 180c4a0e187aa9e55af293731886e529bd373929cb0a24b951930a9a737de8ea |
|
MD5 | f197eb4f9d64b476bb78c98937fe237b |
|
BLAKE2b-256 | a8c55c2354914297256ae926d6a67fad241526816ddc3d34ab29446c388cac8c |
Close
Hashes for amqp_client_python-0.1.10-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4fc3dc233890756114396fa9ee25e5cdf44f50007bd0476465166bfc66127a26 |
|
MD5 | 3eff7c3bd835bdd6323caf98650ce937 |
|
BLAKE2b-256 | 6b78c1b53c3b9f3aadc093b06163a0df80b8226890e616b8d93be6fc705dff65 |