Python script to query and save all SNS topics and their attributes in a given region.

review_sns_topics.py

#!/usr/bin/env python3
import json
import re
import sys
import os

import boto3

# -------------------------------------------------------------------- main ---


def review_topics(config_opts):
    config = compile_config(config_opts)
    sns_client = config["aws.session"].client("sns")
    topic_list = list_sns_topics(sns_client)
    topics = process_topics(topic_list, config["log.topics"])
    report = create_topics_report(topics)
    if config["dump.enabled"]:
        dump_json(report, config["dump.path"])
    return report


# --------------------------------------------------------------------- fns ---


def create_topics_report(topics):
    no_subs = list(filter(lambda x: x.get("SubscriptionsConfirmed", 0) == 0, topics))
    metrics = {"total": len(topics), "no-subs": len(no_subs)}
    report = {"topics": topics, "metrics": metrics}
    return report


def process_topics(topics, log_enable):
    processed_topics = []
    for topic in topics:
        processed_topics.append(topic)
        if log_enable == False:
            continue
        count = len(processed_topics)
        print(f'{count:<12}{topic["SubscriptionsConfirmed"]:<10}{topic["TopicArn"]}')
        if count == 10:
            break
    return processed_topics


def dump_json(data, path):
    with open(path, "w") as file:
        json.dump(data, file)


def load_json(path):
    with open(path, "r") as file:
        data = json.load(file)
    return data


# --------------------------------------------------------------------- aws ---


def list_sns_topics(sns_client):
    paginator = sns_client.get_paginator("list_topics")
    pages = paginator.paginate()
    for page in pages:
        topics = page["Topics"]
        for topic in topics:
            topic_attrs = get_sns_topic_attributes(topic["TopicArn"], sns_client)
            yield topic_attrs


def get_sns_topic_attributes(topic_arn, sns_client):
    response = sns_client.get_topic_attributes(TopicArn=topic_arn)
    return response.get("Attributes", {})


# ------------------------------------------------------------------ config ---


def compile_config(config):
    config["aws.session"] = boto3.Session(
        region_name=config["aws.region"], profile_name=config["aws.profile"]
    )
    return config


def get_default_config():
    config = {
        "aws.region": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
        "aws.profile": os.environ.get("AWS_DEFAULT_PROFILE"),
        "log.topics": int(os.environ.get("SCRIPT_LOG_TOPICS", "1")) == 1,
        "report.enabled": int(os.environ.get("SCRIPT_REPORT_ENABLED", "1")) == 1,
        "dump.enabled": int(os.environ.get("SCRIPT_DUMP_ENABLED", "1")) == 1,
        "dump.path": os.environ.get("SCRIPT_DUMP_PATH", "./report.json"),
    }
    return config


# ---------------------------------------------------------------- handlers ---


def script_handler(args):
    config = get_default_config()
    # script.py <aws_profile>
    if len(args) == 2:
        config["aws.profile"] = args[1]
    # script.py <aws_profile> <aws_region>
    elif len(args) == 3:
        config["aws.profile"] = args[1]
        config["aws.region"] = args[2]
    result = review_topics(config)
    print(f"total topics w/o subs: {result['metrics']['no-subs']}")
    print(f"total topics: {result['metrics']['total']}")


def lambda_handler(event, context):
    raise Exception("Not implemented yet!")


if __name__ == "__main__":
    script_handler(sys.argv)

Script to query and save the SNS topic attributes for all SNS topics in a given region. Requires boto3 to be installed.