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__()))
def save_aggregate_reports_to_kafka ( self , aggregate_reports , aggregate_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
Saves aggregate DMARC reports to Kafka