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