Add black and isort as formatting utility for the project. Start mypy…

… support. (#638)

* Refactored util.py and structs.py to pass mypy, format using blake and isort.
Having automatic code formatting helps with consistent styling across the project,
thus want to encorporate it part after part.

* Add config.cfg and stubs for Kafka library

* Flake should apply for the whole project, not only for formatted areas

* Fix tests

* Added dataclasses and async generator support for older python's

Makefile

@@ -5,14 +5,26 @@ SCALA_VERSION?=2.12
 KAFKA_VERSION?=2.2.2
 DOCKER_IMAGE=aiolibs/kafka:$(SCALA_VERSION)_$(KAFKA_VERSION)
 DIFF_BRANCH=origin/master
+FORMATTED_AREAS=aiokafka/util.py aiokafka/structs.py
 
 setup:
 	pip install -r requirements-dev.txt
 	pip install -Ue .
 
-flake:
-	extra=$$(python -c "import sys;sys.stdout.write('--exclude tests/test_pep492.py') if sys.version_info[:3] < (3, 5, 0) else sys.stdout.write('')"); \
-	flake8 aiokafka tests $$extra
+format:
+	isort -rc $(FORMATTED_AREAS) setup.py
+	black $(FORMATTED_AREAS) setup.py
+
+flake: lint
+lint:
+	black --check $(FORMATTED_AREAS) setup.py
+	@if ! isort -c -rc $(FORMATTED_AREAS) setup.py; then \
+            echo "Import sort errors, run 'make format' to fix them!!!"; \
+            isort --diff -rc $(FORMATTED_AREAS) setup.py; \
+            false; \
+        fi
+	flake8 aiokafka tests setup.py
+	mypy $(FORMATTED_AREAS)
 
 test: flake
 	py.test -s --show-capture=no --docker-image $(DOCKER_IMAGE) $(FLAGS) tests

aiokafka/structs.py

@@ -1,22 +1,52 @@
-import collections
+from dataclasses import dataclass
+from typing import Generic, NamedTuple, Optional, Sequence, Tuple, TypeVar
+
 from kafka.structs import (
-    OffsetAndMetadata, TopicPartition, BrokerMetadata, PartitionMetadata
+    BrokerMetadata,
+    OffsetAndMetadata,
+    PartitionMetadata,
+    TopicPartition,
 )
 
+
 __all__ = [
-    "OffsetAndMetadata", "TopicPartition", "RecordMetadata", "ConsumerRecord",
-    "BrokerMetadata", "PartitionMetadata"
+    "OffsetAndMetadata",
+    "TopicPartition",
+    "RecordMetadata",
+    "ConsumerRecord",
+    "BrokerMetadata",
+    "PartitionMetadata",
 ]
 
-RecordMetadata = collections.namedtuple(
-    'RecordMetadata', ['topic', 'partition', 'topic_partition', 'offset',
-                       'timestamp', 'timestamp_type'])
 
-ConsumerRecord = collections.namedtuple(
-    "ConsumerRecord", ["topic", "partition", "offset", "timestamp",
-                       "timestamp_type", "key", "value", "checksum",
-                       "serialized_key_size", "serialized_value_size",
-                       "headers"])
+class RecordMetadata(NamedTuple):
+    topic: str
+    partition: int
+    topic_partition: TopicPartition
+    offset: int
+    timestamp: Optional[int]  # Timestamp in millis, None for older Brokers
+    timestamp_type: int
+
+
+KT = TypeVar("KT")
+VT = TypeVar("VT")
+
+
+@dataclass
+class ConsumerRecord(Generic[KT, VT]):
+    topic: str
+    partition: int
+    offset: int
+    timestamp: int
+    timestamp_type: int
+    key: Optional[KT]
+    value: Optional[VT]
+    checksum: int
+    serialized_key_size: int
+    serialized_value_size: int
+    headers: Sequence[Tuple[str, bytes]]
+
 
-OffsetAndTimestamp = collections.namedtuple(
-    "OffsetAndTimestamp", ["offset", "timestamp"])
+class OffsetAndTimestamp(NamedTuple):
+    offset: int
+    timestamp: Optional[int]  # Only None if used with old broker version

aiokafka/util.py

@@ -1,9 +1,12 @@
 import asyncio
 import os
 import sys
+from asyncio import AbstractEventLoop
 from distutils.version import StrictVersion
+from typing import Dict, Tuple, TypeVar, Union
+
+from .structs import OffsetAndMetadata, TopicPartition
 
-from .structs import TopicPartition, OffsetAndMetadata
 
 __all__ = ["ensure_future", "create_future", "PY_35"]
 
@@ -13,22 +16,26 @@
 except ImportError:
     exec("from asyncio import async as ensure_future")
 
+T = TypeVar("T")
+
 
-def create_future(loop):
+def create_future(loop: AbstractEventLoop) -> "asyncio.Future[T]":
     try:
         return loop.create_future()
     except AttributeError:
         return asyncio.Future(loop=loop)
 
 
-def parse_kafka_version(api_version):
+def parse_kafka_version(api_version: str) -> Tuple[int, int, int]:
     version = StrictVersion(api_version).version
     if not (0, 9) <= version < (3, 0):
         raise ValueError(api_version)
     return version
 
 
-def commit_structure_validate(offsets):
+def commit_structure_validate(
+    offsets: Dict[TopicPartition, Union[int, Tuple[int, str], OffsetAndMetadata]]
+) -> Dict[TopicPartition, OffsetAndMetadata]:
     # validate `offsets` structure
     if not offsets or not isinstance(offsets, dict):
         raise ValueError(offsets)
@@ -66,7 +73,7 @@ def get_running_loop() -> asyncio.AbstractEventLoop:
 PY_35 = sys.version_info >= (3, 5)
 PY_352 = sys.version_info >= (3, 5, 2)
 PY_36 = sys.version_info >= (3, 6)
-NO_EXTENSIONS = bool(os.environ.get('AIOKAFKA_NO_EXTENSIONS'))
+NO_EXTENSIONS = bool(os.environ.get("AIOKAFKA_NO_EXTENSIONS"))
 
 INTEGER_MAX_VALUE = 2 ** 31 - 1
-INTEGER_MIN_VALUE = - 2 ** 31
+INTEGER_MIN_VALUE = -(2 ** 31)

requirements-ci.txt

@@ -1,11 +1,17 @@
 -r requirements-cython.txt
 flake8==3.8.3
+black==19.10b0
+mypy==0.782
+isort==4.3.21
 pytest==5.4.3
 pytest-cov==2.10.0
+pytest-asyncio==0.12.0
 docker==4.2.1
 lz4==3.1.0
 xxhash==1.4.3
 python-snappy==0.5.4
 docutils==0.16
 Pygments==2.6.1
 gssapi==1.6.2 # pyup: <= 1.6.2  # For some reason 1.6.5 does not install with py35
+dataclasses==0.5; python_version<"3.7"
+async_generator==1.10; python_version<"3.7"

requirements-win-test.txt

@@ -1,8 +1,14 @@
 -r requirements-cython.txt
 flake8==3.8.3
+black==19.10b0
+mypy==0.782
+isort==4.3.21
 pytest==5.4.3
 pytest-cov==2.10.0
+pytest-asyncio==0.12.0
 docker==4.2.1
 lz4==3.1.0
 xxhash==1.4.3
 python-snappy==0.5.4
+dataclasses==0.5; python_version<"3.7"
+async_generator==1.10; python_version<"3.7"

setup.cfg

@@ -0,0 +1,48 @@
+[flake8]
+max-line-length = 88
+exclude =
+    .git
+    venv
+    __pycache__
+    .tox
+
+[isort]
+line_length=88
+include_trailing_comma=True
+multi_line_output=3
+force_grid_wrap=0
+combine_as_imports=True
+lines_after_imports=2
+known_standard_library=dataclasses
+known_first_party=aiokafka,kafka
+known_third_party=pytest
+
+[mypy]
+check_untyped_defs = True
+disallow_any_generics = True
+disallow_untyped_defs = True
+follow_imports = silent
+strict_optional = True
+warn_redundant_casts = True
+warn_unused_ignores = True
+warn_unused_configs = True
+mypy_path=stubs
+
+[mypy-pytest]
+ignore_missing_imports = True
+
+[mypy-Cython]
+ignore_missing_imports = True
+
+[mypy-kafka.*]
+warn_no_return = False
+disallow_untyped_defs = False
+
+[mypy-kafka.vendor.*]
+ignore_missing_imports = True
+
+[mypy-aiokafka.*]
+ignore_missing_imports = True
+
+[mypy-aiokafka.util]
+ignore_missing_imports = False