# Path: microcosm_pubsub/batch.py
# class MessageBatchSchema(PubSubMessageSchema):
#     """
#     A message indicating that a batch of messages needs to be published.
# 
#     """
#     MEDIA_TYPE = created("batch_message")
# 
#     messages = fields.List(fields.Nested(BatchedMessageSchema), required=True)
# 
# Path: microcosm_pubsub/constants.py
# PUBLISHED_KEY = "X-Request-Published"
# 
# Path: microcosm_pubsub/conventions/naming.py
# def make_media_type(resource, lifecycle_change=None, organization=None, public=False):
#     """
#     Generate a canonical media type for a message that announces a lifecycle change to a resource.
# 
#     :param resource: the resource (name)
#     :param lifecycle_change: the change that happened to the resource
#     :param organization: the vendor organization to which this resource belongs
#     :param public: whether the resource is meant for public/external consumption
# 
#     """
#     if lifecycle_change is None:
#         lifecycle_change = LifecycleChange.Created
#     if organization is None:
#         organization = "globality"
# 
#     return "application/vnd.{}.{}.{}.{}".format(
#         # use a vendor specific media type
#         organization,
#         # differentiate messages sent over a pubsub channel from other messages
#         "pubsub" if public else "pubsub._",
#         # qualify the message with the kind of lifecycle change
#         str(lifecycle_change),
#         # specify the resource name
#         name_for(resource),
#     )
# 
# Path: microcosm_pubsub/errors.py
# class TopicNotDefinedError(Exception):
#     """
#     No topic was configured for message media type.
# 
#     """
#     pass
# 
# Path: microcosm_pubsub/tracing.py
# def add_trace_to_message(tracer, pubsub_message):
#     tag = tracer.outgoing_dynatrace_string_tag
#     if tag:
#         try:
#             message = json.loads(pubsub_message.message)
#             message["opaqueData"][OPAQUE_TAG_KEY] = tag
#             pubsub_message.message = json.dumps(message)
#         except Exception:
#             pass
# 
# @contextmanager
# def trace_outgoing_message(topic_arn: str):
#     if not oneagent:
#         yield StubOutgoingTracer()
#     else:
#         sdk = oneagent.get_sdk()
#         try:
#             topic_name = topic_arn.split(":")[-1]
#         except IndexError:
#             topic_name = topic_arn
# 
#         messaging_system = sdk.create_messaging_system_info(
#             AWS_SNS,
#             topic_name,
#             MessagingDestinationType.TOPIC,
#             Channel(ChannelType.TCP_IP, None),
#         )
#         with sdk.trace_outgoing_message(messaging_system) as tracer:
#             yield tracer


from collections import defaultdict
from dataclasses import dataclass
from distutils.util import strtobool
from functools import wraps
from logging import Logger
from time import time
from typing import Dict, List
from boto3.session import Session
from botocore.client import Config
from microcosm.api import defaults, typed
from microcosm.errors import NotBoundError
from microcosm_logging.decorators import logger
from microcosm_logging.timing import elapsed_time
from microcosm_pubsub.batch import MessageBatchSchema
from microcosm_pubsub.constants import PUBLISHED_KEY
from microcosm_pubsub.conventions.naming import make_media_type
from microcosm_pubsub.errors import TopicNotDefinedError
from microcosm_pubsub.tracing import add_trace_to_message, trace_outgoing_message
        from unittest.mock import MagicMock
        if isinstance(value, dict):
            for nested_key, nested_value in collapse_dict(value, key, separator):
                yield separator.join([prefix, nested_key]), nested_value
        else:
            yield separator.join([prefix, key]), value


def iter_topic_mappings(dct):
    for key, value in dct.items():
        if isinstance(value, str):
            yield key, value
        else:
            for nested_key, nested_value in collapse_dict(value, key):
                yield nested_key, nested_value


@defaults(
    default=None,
    mappings={},
)
def configure_sns_topic_arns(graph):
    """
    Configure a mapping from message types to topic ARNs.

    """
    if graph.config.sns_topic_arns.default is None:
        sns_topic_arns = dict()
    else:
        sns_topic_arns = defaultdict(lambda: graph.config.sns_topic_arns.default)
        # NB: Do not use the default for the batch schema

# Path: microcosm_pubsub/batch.py
# class MessageBatchSchema(PubSubMessageSchema):
#     """
#     A message indicating that a batch of messages needs to be published.
# 
#     """
#     MEDIA_TYPE = created("batch_message")
# 
#     messages = fields.List(fields.Nested(BatchedMessageSchema), required=True)
# 
# Path: microcosm_pubsub/constants.py
# PUBLISHED_KEY = "X-Request-Published"
# 
# Path: microcosm_pubsub/conventions/naming.py
# def make_media_type(resource, lifecycle_change=None, organization=None, public=False):
#     """
#     Generate a canonical media type for a message that announces a lifecycle change to a resource.
# 
#     :param resource: the resource (name)
#     :param lifecycle_change: the change that happened to the resource
#     :param organization: the vendor organization to which this resource belongs
#     :param public: whether the resource is meant for public/external consumption
# 
#     """
#     if lifecycle_change is None:
#         lifecycle_change = LifecycleChange.Created
#     if organization is None:
#         organization = "globality"
# 
#     return "application/vnd.{}.{}.{}.{}".format(
#         # use a vendor specific media type
#         organization,
#         # differentiate messages sent over a pubsub channel from other messages
#         "pubsub" if public else "pubsub._",
#         # qualify the message with the kind of lifecycle change
#         str(lifecycle_change),
#         # specify the resource name
#         name_for(resource),
#     )
# 
# Path: microcosm_pubsub/errors.py
# class TopicNotDefinedError(Exception):
#     """
#     No topic was configured for message media type.
# 
#     """
#     pass
# 
# Path: microcosm_pubsub/tracing.py
# def add_trace_to_message(tracer, pubsub_message):
#     tag = tracer.outgoing_dynatrace_string_tag
#     if tag:
#         try:
#             message = json.loads(pubsub_message.message)
#             message["opaqueData"][OPAQUE_TAG_KEY] = tag
#             pubsub_message.message = json.dumps(message)
#         except Exception:
#             pass
# 
# @contextmanager
# def trace_outgoing_message(topic_arn: str):
#     if not oneagent:
#         yield StubOutgoingTracer()
#     else:
#         sdk = oneagent.get_sdk()
#         try:
#             topic_name = topic_arn.split(":")[-1]
#         except IndexError:
#             topic_name = topic_arn
# 
#         messaging_system = sdk.create_messaging_system_info(
#             AWS_SNS,
#             topic_name,
#             MessagingDestinationType.TOPIC,
#             Channel(ChannelType.TCP_IP, None),
#         )
#         with sdk.trace_outgoing_message(messaging_system) as tracer:
#             yield tracer


# Path: microcosm_pubsub/producer.py
from collections import defaultdict
from dataclasses import dataclass
from distutils.util import strtobool
from functools import wraps
from logging import Logger
from time import time
from typing import Dict, List
from boto3.session import Session
from botocore.client import Config
from microcosm.api import defaults, typed
from microcosm.errors import NotBoundError
from microcosm_logging.decorators import logger
from microcosm_logging.timing import elapsed_time
from microcosm_pubsub.batch import MessageBatchSchema
from microcosm_pubsub.constants import PUBLISHED_KEY
from microcosm_pubsub.conventions.naming import make_media_type
from microcosm_pubsub.errors import TopicNotDefinedError
from microcosm_pubsub.tracing import add_trace_to_message, trace_outgoing_message
        from unittest.mock import MagicMock

        if isinstance(value, dict):
            for nested_key, nested_value in collapse_dict(value, key, separator):
                yield separator.join([prefix, nested_key]), nested_value
        else:
            yield separator.join([prefix, key]), value


def iter_topic_mappings(dct):
    for key, value in dct.items():
        if isinstance(value, str):
            yield key, value
        else:
            for nested_key, nested_value in collapse_dict(value, key):
                yield nested_key, nested_value


@defaults(
    default=None,
    mappings={},
)
def configure_sns_topic_arns(graph):
    """
    Configure a mapping from message types to topic ARNs.

    """
    if graph.config.sns_topic_arns.default is None:
        sns_topic_arns = dict()
    else:
        sns_topic_arns = defaultdict(lambda: graph.config.sns_topic_arns.default)
        # NB: Do not use the default for the batch schema

        sns_topic_arns[MessageBatchSchema.MEDIA_TYPE] = None