def _get_complaint_data()

in emails/views.py [0:0]

• 43 lines of code
• 12 McCabe index (conditional complexity)

def _get_complaint_data(message_json: AWS_SNSMessageJSON) -> RawComplaintData:
    """
    Extract complaint data from an AWS SES Complaint Notification.

    This extracts only the data used by _handle_complaint(). It also works on
    complaint events, which have a similar structure and the same data needed
    by _handle_complaint.

    For more information on the complaint notification, see:
    https://docs.aws.amazon.com/ses/latest/dg/notification-contents.html#complaint-object
    """
    complaint = message_json["complaint"]

    T = TypeVar("T")

    def get_or_log(
        key: str, source: dict[str, T], data_type: type[T]
    ) -> tuple[T, bool]:
        """Get a value from a dictionary, or log if not found"""
        if key in source:
            return source[key], True
        logger.error(
            "_get_complaint_data: Unexpected message format",
            extra={"missing_key": key, "found_keys": ",".join(sorted(source.keys()))},
        )
        return data_type(), False

    raw_recipients, has_cr = get_or_log("complainedRecipients", complaint, list)
    complained_recipients = []
    no_entries = True
    for entry in raw_recipients:
        no_entries = False
        raw_email_address, has_email = get_or_log("emailAddress", entry, str)
        if has_email:
            email_address = parseaddr(raw_email_address)[1]
            extra = {
                key: value for key, value in entry.items() if key != "emailAddress"
            }
            complained_recipients.append((email_address, extra))
    if has_cr and no_entries:
        logger.error("_get_complaint_data: Empty complainedRecipients")

    mail, has_mail = get_or_log("mail", message_json, dict)
    if has_mail:
        commonHeaders, has_ch = get_or_log("commonHeaders", mail, dict)
    else:
        commonHeaders, has_ch = {}, False
    if has_ch:
        raw_from_addresses, _ = get_or_log("from", commonHeaders, list)
    else:
        raw_from_addresses = []
    from_addresses = [parseaddr(addr)[1] for addr in raw_from_addresses]

    # Only present when set
    feedback_type = complaint.get("complaintFeedbackType", "")
    # Only present when destination is on account suppression list
    subtype = complaint.get("complaintSubType", "")
    # Only present for feedback reports
    user_agent = complaint.get("userAgent", "")

    return RawComplaintData(
        complained_recipients, from_addresses, subtype, user_agent, feedback_type
    )