Skip to content

👑 aka. Reis Mags ✉️ Async communications using AWS SNS + SQS for Python ✨

License

Notifications You must be signed in to change notification settings

jairovadillo/razemax

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

36 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Razemax

Build Status

✉️ 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: int, target: str):
        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 must be subscribed to SNS topic before running the consumer, also the queue needs to filter only the events of interest. If the provided SNS topic pushes events which are not included in the mapper, then razemax will be raising exceptions.

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

pip install razemax

Running the tests

To run end to end tests do:

make unit-tests
make integration-tests

License

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

About

👑 aka. Reis Mags ✉️ Async communications using AWS SNS + SQS for Python ✨

Topics

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Contributors 4

  •  
  •  
  •  
  •