ReaderAbstract.__init__() - Code Metrics - Inspection of "Feature/confluent kafka consumer" - mre/kafka-influxdb - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#43)

by Matthias

created 2016-11-29 12:37 UTC

ReaderAbstract.init() A

↳ Parent: ReaderAbstract

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	1
c	2
b	0
f	0
dl	0
loc	12
rs	9.4285

import logging
from kafka_influxdb.encoder.errors import EncoderError


class ReaderAbstract(object):
    """
    A Kafka consumer based on kafka-python
    See: https://github.com/dpkp/kafka-python
    """

    # We need to treat legacy Kafka Versions (< 0.9.0) a little different.
    # First, they don't work without Zookeeper and confluent-kafka has no support for Zookeeper.
    # Second, they don't support API discovery.
    # The specific workarounds required are documented below.
    KAFKA_VERSION_ZOOKEEPER_OPTIONAL = "0.9.0"

    def __init__(self, host, port, group, topic, broker_version=KAFKA_VERSION_ZOOKEEPER_OPTIONAL):
        """
        Initialize Kafka reader
        """
        self.host = host
        self.port = str(port)
        self.group = group
        self.topic = topic
        self.broker_version = broker_version

        # Initialized on read
        self.consumer = None

    def read(self):
        """
        Read from Kafka. Reconnect on error.
        """
        try:
            self._connect()
            for msg in self._handle_read():
                yield msg
        finally:
            logging.info("Performing cleanup before stopping.")
            self._shutdown()

    def _connect(self):
        """
        Overwrite in child classes
        """
        raise NotImplementedError

    def _shutdown(self):
        """
        Cleanup tasks (e.g. closing the Kafka connection).
        Can be overwritten by specific readers if required.
        """
        if self.consumer:
            self.consumer.close()

    def _handle_read(self):
        """
        Read messages from Kafka.
        Library-specific internal message handling.
        Needs to be implemented by every reader
        """
        raise NotImplementedError


1			import logging
2			from kafka_influxdb.encoder.errors import EncoderError
3
4
5			class ReaderAbstract(object):
6			"""
7			A Kafka consumer based on kafka-python
8			See: https://github.com/dpkp/kafka-python
9			"""
10
11			# We need to treat legacy Kafka Versions (< 0.9.0) a little different.
12			# First, they don't work without Zookeeper and confluent-kafka has no support for Zookeeper.
13			# Second, they don't support API discovery.
14			# The specific workarounds required are documented below.
15			KAFKA_VERSION_ZOOKEEPER_OPTIONAL = "0.9.0"
16
17			def __init__(self, host, port, group, topic, broker_version=KAFKA_VERSION_ZOOKEEPER_OPTIONAL):
18			"""
19			Initialize Kafka reader
20			"""
21			self.host = host
22			self.port = str(port)
23			self.group = group
24			self.topic = topic
25			self.broker_version = broker_version
26
27			# Initialized on read
28			self.consumer = None
29
30			def read(self):
31			"""
32			Read from Kafka. Reconnect on error.
33			"""
34			try:
35			self._connect()
36			for msg in self._handle_read():
37			yield msg
38			finally:
39			logging.info("Performing cleanup before stopping.")
40			self._shutdown()
41
42			def _connect(self):
43			"""
44			Overwrite in child classes
45			"""
46			raise NotImplementedError
47
48			def _shutdown(self):
49			"""
50			Cleanup tasks (e.g. closing the Kafka connection).
51			Can be overwritten by specific readers if required.
52			"""
53			if self.consumer:
54			self.consumer.close()
55
56			def _handle_read(self):
57			"""
58			Read messages from Kafka.
59			Library-specific internal message handling.
60			Needs to be implemented by every reader
61			"""
62			raise NotImplementedError
63

mre / kafka-influxdb

Pull Request — master (#43)

ReaderAbstract.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ReaderAbstract.init() A