GitLocker: The Coding Marketplace

Description:

razemax 0.0.8

Razemax

✉️ Async communications using AWS SNS + SQS for Python services ✨
Documentation
In-Memory event manager
Show me the code
from razemax.event_manager import EventManager

class NorthKoreaThreatCreatedEvent:
def __init__(self, id, target):
self.id = id
self.target = target

def trump_subscriber(event: NorthKoreaThreatCreatedEvent):
print(f"North korea will attack us or {event.target}!")

EventManager.subscribe(trump_subscriber, NorthKoreaThreatCreatedEvent)
EventManager.trigger(NorthKoreaThreatCreatedEvent(0, "Mexico"))

Result:
North korea will attack us or Mexico!

Trigger subscribers from SQS
Preconditions
SQS queue has to be subscribed to SNS topic before running the consumer
Code
from razemax.consumers import MessageConsumer
from razemax.drivers import SQSDriver
from razemax.event_manager import EventManager
from razemax.publisher import SNSMessagePublisher

def kp_message_to_event(message):
# Highly recommended to use Marshmallow to validate
return NorthKoreaThreatCreatedEvent(message.body['id'], message.body['target_name'])

mapper = {
'KPThreatCreated': kp_message_to_event
}

aws_settings = {
'region_name': "",
'aws_access_key_id': "",
'aws_secret_access_key': "",
'endpoint_url': ""
}

queue_driver = SQSDriver.build("korea-threats-queue", aws_settings)
MessageConsumer(mapper, EventManager, queue_driver).process_message()

publisher = SNSMessagePublisher.build(aws_settings, 'korea-topic')
publisher.publish('KPThreatCreated', {'id': 21, 'target_name': 'Portugal'})

Result:
North korea will attack us or Portugal!

Installing (TODO)
pip install git@github.com/transporter
Running the tests
To run end to end tests do:
make unit-tests
make integration-tests

Authors

Jairo Vadillo (@jairovadillo)

License
This project is licensed under the MIT License - see the LICENSE.md file for details