Producer Initialization¶

If your application produces events for other application to consume, you should set your INIESTA_INITIALIZATION_TYPE to include SNS_PRODUCER.

On start up of Insanic, a SNSClient instance is created and attached to the application. It is accessible with app.xavi. It is recommended to have only one SNSClient for each topic until the server is stopped. For other topics, you will need to create a separate SNSClient and manage its lifecycle.

Publishing Messages¶

With our client initialized, we can now create a SNSMessage to publish to our topic. There are several ways we can do this. We can to either publish messages explicitly or decorate view functions or methods to publish events on successful execution.

With `SNSClient`¶

The SNSClient provides a factory method to create a SNSMessage to prepare and send.

async def publish_somewhere(app):

    message: SNSMessage = app.xavi.create_message(
        event="SomeEvent",
        message={"extra": "extra stuff i want to send"}
    )

    await message.publish()

With `SNSMessage`¶

from iniesta.sns import SNSMessage

async def somewhere_in_my_code(app):

    message = SNSMessage.create_message(
        client=app.xavi,
        event="SomeEvent",
        message={"extra": "extra stuff I want to send"}
    )

    await message.publish()

With a decorator¶

The client also provides a separate decorator that can simplify event publishing.

from sanic.response import json
from insanic.views import InsanicView
from . import app # your Insanic app

class MyPublishingView(InsanicView):

    @app.xavi.publish_event(event="SaidHello")
    async def post(self, request, *args, **kwargs):

        return json({"hello": "hi"})

app.add_route(MyPublishingView.as_view(), "/hi")

Now if we run our server, and call the /hi endpoint, an event will be published with the event SaidHello, with the response body as the message.

Note

Publishing will only trigger if the response has a status_code of less than 300. Any exceptions raised or a response with a explicit status code of more than 300 will NOT publish a message.

Initializing a SNSClient¶

If you have more than 1 topic that you need to publish to, we can create a separate client to connect to.

To create your own SNSClient we need to use a separate async class method to initialize our client.

from iniesta.sns import SNSClient

async def initialize_client():
    client = await SNSClient.initialize(
        topic_arn = "your topic arn"
    )

This will confirm the topic exists and return an initialized SNSClient instance. It is not recommended creating a SNSClient each time your need to publish a message so, you will need to manage its lifecycle. A possible approach is to initialize on the before start listeners and attach it to the application.