parsedmarc/kafkaclient.py
def save_aggregate_reports_to_kafka(self, aggregate_reports,
                                        aggregate_topic):
        """
        Saves aggregate DMARC reports to Kafka

        Args:
            aggregate_reports (list):  A list of aggregate report dictionaries
            to save to Kafka
            aggregate_topic (str): The name of the Kafka topic

        """
        if (type(aggregate_reports) == dict or
           type(aggregate_reports) == OrderedDict):
            aggregate_reports = [aggregate_reports]

        if len(aggregate_reports) < 1:
            return

        for report in aggregate_reports:
            report['date_range'] = self.generate_daterange(report)
            report = self.strip_metadata(report)

            for slice in report['records']:
                slice['date_range'] = report['date_range']
                slice['org_name'] = report['org_name']
                slice['org_email'] = report['org_email']
                slice['policy_published'] = report['policy_published']
                slice['report_id'] = report['report_id']
                logger.debug("Sending slice.")
                try:
                    logger.debug("Saving aggregate report to Kafka")
                    self.producer.send(aggregate_topic, slice)
                except UnknownTopicOrPartitionError:
                    raise KafkaError(
                        "Kafka error: Unknown topic or partition on broker")
                except Exception as e:
                    raise KafkaError(
                        "Kafka error: {0}".format(e.__str__()))
                try:
                    self.producer.flush()
                except Exception as e:
                    raise KafkaError(
                        "Kafka error: {0}".format(e.__str__()))
def save_aggregate_reports_to_kafka(self, aggregate_reports,
                                        aggregate_topic):
        """
        Saves aggregate DMARC reports to Kafka

        Args:
            aggregate_reports (list):  A list of aggregate report dictionaries
            to save to Kafka
            aggregate_topic (str): The name of the Kafka topic

        """
        if (type(aggregate_reports) == dict or
           type(aggregate_reports) == OrderedDict):
            aggregate_reports = [aggregate_reports]

        if len(aggregate_reports) < 1:
            return

        for report in aggregate_reports:
            report['date_range'] = self.generate_daterange(report)
            report = self.strip_metadata(report)

            for slice in report['records']:
                slice['date_range'] = report['date_range']
                slice['org_name'] = report['org_name']
                slice['org_email'] = report['org_email']
                slice['policy_published'] = report['policy_published']
                slice['report_id'] = report['report_id']
                logger.debug("Sending slice.")
                try:
                    logger.debug("Saving aggregate report to Kafka")
                    self.producer.send(aggregate_topic, slice)
                except UnknownTopicOrPartitionError:
                    raise KafkaError(
                        "Kafka error: Unknown topic or partition on broker")
                except Exception as e:
                    raise KafkaError(
                        "Kafka error: {0}".format(e.__str__()))
                try:
                    self.producer.flush()
                except Exception as e:
                    raise KafkaError(
                        "Kafka error: {0}".format(e.__str__()))
Saves aggregate DMARC reports to Kafka

        Args:
            aggregate_reports (list):  A list of aggregate report dictionaries
            to save to Kafka
            aggregate_topic (str): The name of the Kafka topic