SNS Message parser

Parsing SNS messages from AWS drop in class parser. Getting to the point of using a class object to perform the action as follows

with SNSParser() as p:
  results = p.extract_message(event=input_event)

The contents of the message above is dependent on your input structure which you can override in the ExpectedProperties class input to the parser. Default is command or a dictionary object that looks like this

Example Message

{
  "command": "testCommand"
  ...
}

Expectation is that you are sending messages of this format with boto3 or other compliant library

sns.publish(TopicArn=topic_arn, Message=json.dumps(message)

Parser class

import dataclasses
import json
import logging
from collections import deque
from typing import Any

from types import TracebackType


@dataclasses.dataclass(frozen=True, kw_only=True)
class ExpectedProperties:
    message: str = 'message'
    command: str = 'command'


class SNSParser:
    log = logging.getLogger(__name__)
    expected_properties: ExpectedProperties

    def __enter__(self) -> SNSParser:
        return self

    def __exit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None:
        if exc_type is not None:
            self.log.error(exc_type)

    def __init__(self, expected_props: ExpectedProperties | None = None) -> None:
        self.expected_properties = ExpectedProperties()
        if expected_props is not None:
            self.expected_properties = expected_props

    def extract_message(self, event:  dict[str, Any]) -> dict[str, Any]:
        self.log.debug(f'{event=}')
        if not isinstance(event, dict):
            event = json.loads(event, strict=False)
            self.log.debug(f'{event=}')
        input_event = event.get('event')
        if input_event is None or not isinstance(input_event, dict):
            self.log.info(f'Reassigning `input_event` to original `event`')
            input_event = event
        records = input_event.get('Records')
        if records is None:
            self.log.debug(f'No Records')
            return event
        if not isinstance(records, list):
            self.log.debug(f'`Records` is not a list')
            return event
        records = deque(records)
        if len(records) == 0:
            self.log.debug(f'No Records')
            return event
        sns_message = records.popleft().get('Sns')
        if sns_message is None:
            self.log.debug(f'No `Sns` property in first `Records`')
            return event
        if not isinstance(sns_message, dict):
            self.log.debug(f'`Sns` object is not a dictionary')
            return event
        message_contents = sns_message.get('Message')
        if message_contents is None:
            self.log.debug(f'No `Message` property in `Sns` object')
            return event
        if not isinstance(message_contents, dict):
            self.log.debug(f'Parsing `message_contents` as dict')
            self.log.debug(f'`message_contents` type is {type(message_contents)}')
            wrapped_message = json.loads(message_contents, strict=False)
        else:
            wrapped_message = message_contents.get('message')
        if not isinstance(wrapped_message, dict):
            self.log.debug(f'Parsing `wrapped_message` as dict')
            self.log.debug(f'`wrapped_message` type is {type(wrapped_message)}')
            return json.loads(wrapped_message, strict=False)
        inner_message = wrapped_message.get(self.expected_properties.message)
        if inner_message is None:
            self.log.info(f'Reassigning `inner_message` to `wrapped_message`')
            inner_message = wrapped_message
        if not isinstance(inner_message, dict):
            self.log.debug(f'Unable to parse `inner_message` as dict')
            return event
        if inner_message.get(self.expected_properties.command) is None:
            self.log.debug(f'No `{self.expected_properties.command}` property in `inner_message`')
            return event
        return inner_message