Make build flake8 clean and add flake8 precheck to .travis.yml

Author: ewencp

.travis.yml

@@ -16,5 +16,8 @@ before_install:
   - pip install --upgrade pip
   - pip install pytest-timeout
 install:
+  - pip install flake8
   - pip install -v --global-option=build_ext --global-option="-Itmp-build/include/" --global-option="-Ltmp-build/lib" . .[avro]
-script: py.test -v --timeout 20 --ignore=tmp-build --import-mode append
+script:
+  - flake8
+  - py.test -v --timeout 20 --ignore=tmp-build --import-mode append

confluent_kafka/__init__.py

@@ -1,5 +1,5 @@
 __all__ = ['cimpl', 'avro', 'kafkatest']
-from .cimpl import (Consumer,
+from .cimpl import (Consumer,  # noqa
                     KafkaError,
                     KafkaException,
                     Message,

confluent_kafka/avro/__init__.py

@@ -2,58 +2,12 @@
     Avro schema registry module: Deals with encoding and decoding of messages with avro schemas
 
 """
-import sys
 
 from confluent_kafka import Producer, Consumer
-
-VALID_LEVELS = ['NONE', 'FULL', 'FORWARD', 'BACKWARD']
-
-
-def loads(schema_str):
-    """ Parse a schema given a schema string """
-    if sys.version_info[0] < 3:
-        return schema.parse(schema_str)
-    else:
-        return schema.Parse(schema_str)
-
-
-def load(fp):
-    """ Parse a schema from a file path """
-    with open(fp) as f:
-        return loads(f.read())
-
-
-# avro.schema.RecordSchema and avro.schema.PrimitiveSchema classes are not hashable. Hence defining them explicitely as a quick fix
-def _hash_func(self):
-    return hash(str(self))
-
-
-try:
-    from avro import schema
-
-    schema.RecordSchema.__hash__ = _hash_func
-    schema.PrimitiveSchema.__hash__ = _hash_func
-except ImportError:
-    pass
-
-
-class ClientError(Exception):
-    """ Error thrown by Schema Registry clients """
-
-    def __init__(self, message, http_code=None):
-        self.message = message
-        self.http_code = http_code
-        super(ClientError, self).__init__(self.__str__())
-
-    def __repr__(self):
-        return "ClientError(error={error})".format(error=self.message)
-
-    def __str__(self):
-        return self.message
-
-
+from confluent_kafka.avro.error import ClientError
+from confluent_kafka.avro.load import load, loads  # noqa
 from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
-from confluent_kafka.avro.serializer import (SerializerError,
+from confluent_kafka.avro.serializer import (SerializerError,  # noqa
                                              KeySerializerError,
                                              ValueSerializerError)
 from confluent_kafka.avro.serializer.message_serializer import MessageSerializer
@@ -117,7 +71,6 @@ def produce(self, **kwargs):
             else:
                 raise KeySerializerError("Avro schema required for key")
 
-
         super(AvroProducer, self).produce(topic, value, key, **kwargs)
 
 

confluent_kafka/avro/cached_schema_registry_client.py

@@ -25,9 +25,11 @@
 
 import requests
 
-from . import ClientError, VALID_LEVELS
+from .error import ClientError
 from . import loads
 
+VALID_LEVELS = ['NONE', 'FULL', 'FORWARD', 'BACKWARD']
+
 # Common accept header sent
 ACCEPT_HDR = "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json"
 log = logging.getLogger(__name__)
@@ -119,7 +121,7 @@ def register(self, subject, avro_schema):
 
         schemas_to_id = self.subject_to_schema_ids[subject]
         schema_id = schemas_to_id.get(avro_schema, None)
-        if schema_id != None:
+        if schema_id is not None:
             return schema_id
         # send it up
         url = '/'.join([self.url, 'subjects', subject, 'versions'])
@@ -222,7 +224,7 @@ def get_version(self, subject, avro_schema):
         """
         schemas_to_version = self.subject_to_schema_versions[subject]
         version = schemas_to_version.get(avro_schema, None)
-        if version != None:
+        if version is not None:
             return version
 
         url = '/'.join([self.url, 'subjects', subject])

confluent_kafka/avro/error.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python
+#
+# Copyright 2017 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+class ClientError(Exception):
+    """ Error thrown by Schema Registry clients """
+
+    def __init__(self, message, http_code=None):
+        self.message = message
+        self.http_code = http_code
+        super(ClientError, self).__init__(self.__str__())
+
+    def __repr__(self):
+        return "ClientError(error={error})".format(error=self.message)
+
+    def __str__(self):
+        return self.message

confluent_kafka/avro/load.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+#
+# Copyright 2017 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import sys
+
+
+def loads(schema_str):
+    """ Parse a schema given a schema string """
+    if sys.version_info[0] < 3:
+        return schema.parse(schema_str)
+    else:
+        return schema.Parse(schema_str)
+
+
+def load(fp):
+    """ Parse a schema from a file path """
+    with open(fp) as f:
+        return loads(f.read())
+
+
+# avro.schema.RecordSchema and avro.schema.PrimitiveSchema classes are not hashable. Hence defining them explicitly as
+# a quick fix
+def _hash_func(self):
+    return hash(str(self))
+
+
+try:
+    from avro import schema
+
+    schema.RecordSchema.__hash__ = _hash_func
+    schema.PrimitiveSchema.__hash__ = _hash_func
+except ImportError:
+    schema = None

confluent_kafka/avro/serializer/message_serializer.py

@@ -125,10 +125,9 @@ def encode_record_with_schema_id(self, schema_id, record, is_key=False):
                 if not schema:
                     raise serialize_err("Schema does not exist")
                 self.id_to_writers[schema_id] = avro.io.DatumWriter(schema)
-            except ClientError as e:
+            except ClientError:
                 exc_type, exc_value, exc_traceback = sys.exc_info()
-                raise serialize_err( + repr(
-                    traceback.format_exception(exc_type, exc_value, exc_traceback)))
+                raise serialize_err(repr(traceback.format_exception(exc_type, exc_value, exc_traceback)))
 
         # get the writer
         writer = self.id_to_writers[schema_id]
@@ -171,14 +170,15 @@ def _get_decoder_func(self, schema_id, payload):
             # try to use fast avro
             try:
                 schema_dict = schema.to_json()
-                obj = read_data(payload, schema_dict)
-                # here means we passed so this is something fastavro can do
-                # seek back since it will be called again for the
-                # same payload - one time hit
+                read_data(payload, schema_dict)
 
+                # If we reach this point, this means we have fastavro and it can
+                # do this deserialization. Rewind since this method just determines
+                # the reader function and we need to deserialize again along the
+                # normal path.
                 payload.seek(curr_pos)
-                decoder_func = lambda p: read_data(p, schema_dict)
-                self.id_to_decoder_func[schema_id] = decoder_func
+
+                self.id_to_decoder_func[schema_id] = lambda p: read_data(p, schema_dict)
                 return self.id_to_decoder_func[schema_id]
             except:
                 pass

confluent_kafka/kafkatest/verifiable_client.py

@@ -13,15 +13,22 @@
 # limitations under the License.
 
 
-import signal, socket, os, sys, time, json, re, datetime
+import datetime
+import json
+import os
+import re
+import signal
+import socket
+import sys
+import time
 
 
 class VerifiableClient(object):
     """
     Generic base class for a kafkatest verifiable client.
     Implements the common kafkatest protocol and semantics.
     """
-    def __init__ (self, conf):
+    def __init__(self, conf):
         """
         """
         super(VerifiableClient, self).__init__()
@@ -31,37 +38,36 @@ def __init__ (self, conf):
         signal.signal(signal.SIGTERM, self.sig_term)
         self.dbg('Pid is %d' % os.getpid())
 
-    def sig_term (self, sig, frame):
+    def sig_term(self, sig, frame):
         self.dbg('SIGTERM')
         self.run = False
 
     @staticmethod
-    def _timestamp ():
+    def _timestamp():
         return time.strftime('%H:%M:%S', time.localtime())
 
-    def dbg (self, s):
+    def dbg(self, s):
         """ Debugging printout """
         sys.stderr.write('%% %s DEBUG: %s\n' % (self._timestamp(), s))
 
-    def err (self, s, term=False):
+    def err(self, s, term=False):
         """ Error printout, if term=True the process will terminate immediately. """
         sys.stderr.write('%% %s ERROR: %s\n' % (self._timestamp(), s))
         if term:
             sys.stderr.write('%% FATAL ERROR ^\n')
             sys.exit(1)
 
-    def send (self, d):
+    def send(self, d):
         """ Send dict as JSON to stdout for consumtion by kafkatest handler """
         d['_time'] = str(datetime.datetime.now())
         self.dbg('SEND: %s' % json.dumps(d))
         sys.stdout.write('%s\n' % json.dumps(d))
         sys.stdout.flush()
 
-
     @staticmethod
-    def set_config (conf, args):
+    def set_config(conf, args):
         """ Set client config properties using args dict. """
-        for n,v in args.iteritems():
+        for n, v in args.iteritems():
             if v is None:
                 continue
             # Things to ignore