Merge pull request #26 from kpn/fix/add-mypy

examples/consume_multiple_topics.py

@@ -1,9 +1,9 @@
-from kstreams import create_engine, Stream
-
 import asyncio
 import json
 
-topics = ["local--kstreams", "local--hello-world"]
+from kstreams import Stream, create_engine
+
+topics = ["local--kstreams-2", "local--hello-world"]
 
 stream_engine = create_engine(title="my-stream-engine")
 

examples/fastapi_example/__main__.py

@@ -1,4 +1,5 @@
 import logging
+
 import uvicorn
 
 

examples/fastapi_example/app.py

@@ -1,7 +1,8 @@
-from .streaming.streams import stream_engine
 from fastapi import FastAPI
 from starlette.responses import Response
-from starlette_prometheus import metrics, PrometheusMiddleware
+from starlette_prometheus import PrometheusMiddleware, metrics
+
+from .streaming.streams import stream_engine
 
 
 def create_app():

examples/fastapi_example/streaming/streams.py

@@ -1,6 +1,7 @@
-from .engine import stream_engine
 from kstreams import Stream
 
+from .engine import stream_engine
+
 
 @stream_engine.stream("local--kstream")
 async def stream(stream: Stream):

examples/json_serialization.py

@@ -1,10 +1,11 @@
-from kstreams import consts, create_engine, Stream
-from kstreams.custom_types import Headers
+import asyncio
+import json
 from typing import Any, Dict, Optional
 
 import aiokafka
-import asyncio
-import json
+
+from kstreams import Stream, consts, create_engine
+from kstreams.custom_types import Headers
 
 
 class JsonSerializer:

examples/simple.py

@@ -1,9 +1,10 @@
+import asyncio
+
 from aiokafka.structs import RecordMetadata
+
 from kstreams import create_engine
 from kstreams.streams import Stream
 
-import asyncio
-
 topic = "local--py-streams"
 
 stream_engine = create_engine(title="my-stream-engine")

kstreams/__init__.py

@@ -1,4 +1,4 @@
 from .clients import Consumer, ConsumerType, Producer, ProducerType  # noqa: F401
-from .create import create_engine, StreamEngine  # noqa: F401
+from .create import StreamEngine, create_engine  # noqa: F401
 from .prometheus.monitor import PrometheusMonitor, PrometheusMonitorType  # noqa: F401
 from .streams import KafkaConsumer, KafkaStream, Stream  # noqa: F401

kstreams/clients.py

@@ -1,9 +1,10 @@
-from . import conf, utils
-from pkgsettings import PrefixedSettings
+import logging
 from typing import Callable, Optional, TypeVar
 
 import aiokafka
-import logging
+from pkgsettings import PrefixedSettings
+
+from . import conf, utils
 
 logger = logging.getLogger(__name__)
 

kstreams/create.py

@@ -1,8 +1,9 @@
+from typing import Optional
+
 from .clients import Consumer, ConsumerType, Producer, ProducerType
 from .engine import StreamEngine
 from .prometheus.monitor import PrometheusMonitor, PrometheusMonitorType
 from .serializers import ValueDeserializer, ValueSerializer
-from typing import Optional
 
 
 def create_engine(

kstreams/engine.py

@@ -1,17 +1,18 @@
+import asyncio
+import inspect
+import logging
+from typing import Any, Coroutine, Dict, List, Optional, Type, Union
+
+from kstreams.clients import ConsumerType, ProducerType
+from kstreams.utils import encode_headers
+
 from .custom_types import DecoratedCallable, Headers
 from .exceptions import DuplicateStreamException
 from .prometheus.monitor import PrometheusMonitorType
 from .prometheus.tasks import metrics_task
 from .serializers import ValueDeserializer, ValueSerializer
 from .singlenton import Singleton
 from .streams import KafkaStream, Stream
-from kstreams.clients import ConsumerType, ProducerType
-from kstreams.utils import encode_headers
-from typing import Any, Coroutine, Dict, List, Optional, Type, Union
-
-import asyncio
-import inspect
-import logging
 
 logger = logging.getLogger(__name__)
 

kstreams/prometheus/monitor.py

@@ -1,7 +1,9 @@
-from kstreams.singlenton import Singleton
-from prometheus_client import Gauge
 from typing import Dict, TypeVar
 
+from prometheus_client import Gauge
+
+from kstreams.singlenton import Singleton
+
 PrometheusMonitorType = TypeVar("PrometheusMonitorType", bound="PrometheusMonitor")
 
 

kstreams/prometheus/tasks.py

@@ -1,9 +1,10 @@
-from .monitor import PrometheusMonitorType
+import asyncio
+from typing import Any, DefaultDict, List
+
 from kstreams.clients import Consumer
 from kstreams.streams import KafkaStream
-from typing import DefaultDict, List
 
-import asyncio
+from .monitor import PrometheusMonitorType
 
 
 async def metrics_task(streams: List[KafkaStream], monitor: PrometheusMonitorType):
@@ -43,7 +44,7 @@ async def generate_consumer_metrics(consumer: Consumer, monitor: PrometheusMonit
         }
     """
 
-    metrics = DefaultDict(str)
+    metrics: DefaultDict[Any, dict] = DefaultDict(dict)
 
     topic_partitions = consumer.assignment()
     for topic_partition in topic_partitions:

kstreams/serializers.py

@@ -1,8 +1,9 @@
-from .custom_types import Headers
 from typing import Any, Dict, Optional, Protocol
 
 import aiokafka
 
+from .custom_types import Headers
+
 
 class ValueDeserializer(Protocol):
     async def deserialize(

kstreams/singlenton.py

@@ -1,5 +1,8 @@
+from typing import Dict
+
+
 class Singleton(type):
-    _instances = {}
+    _instances: Dict = {}
 
     def __call__(cls, *args, **kwargs):
         if cls not in cls._instances:

kstreams/streams.py

@@ -1,6 +1,7 @@
-from .clients import Consumer, ConsumerType
-from .serializers import ValueDeserializer
-from aiokafka import errors, structs
+import asyncio
+import inspect
+import logging
+import uuid
 from typing import (
     Any,
     AsyncGenerator,
@@ -13,10 +14,10 @@
     Union,
 )
 
-import asyncio
-import inspect
-import logging
-import uuid
+from aiokafka import errors, structs
+
+from .clients import Consumer, ConsumerType
+from .serializers import ValueDeserializer
 
 logger = logging.getLogger(__name__)
 

kstreams/test_utils/structs.py

@@ -1,7 +1,8 @@
 from dataclasses import dataclass
-from kstreams.custom_types import KafkaHeaders
 from typing import Generic, NamedTuple, Optional, TypeVar
 
+from kstreams.custom_types import KafkaHeaders
+
 KT = TypeVar("KT")
 VT = TypeVar("VT")
 

kstreams/test_utils/test_clients.py

@@ -1,10 +1,12 @@
-from .structs import ConsumerRecord, RecordMetadata, TopicPartition
-from .topics import TopicManager
 from datetime import datetime
+from typing import Any, Coroutine, Dict, List, Optional, Tuple, Union
+
 from kstreams.clients import Consumer, Producer
 from kstreams.custom_types import Headers
 from kstreams.serializers import ValueSerializer
-from typing import Any, Coroutine, Dict, List, Optional, Tuple, Union
+
+from .structs import ConsumerRecord, RecordMetadata, TopicPartition
+from .topics import TopicManager
 
 
 class Base:
@@ -84,6 +86,8 @@ async def getone(
     ) -> Union[bytes, Dict]:  # The return type must be fixed later on
         for topic_partition in self._assigments:
             topic = TopicManager.get_topic(topic_partition.topic)
+            if topic is None:
+                raise AttributeError("There should be a topic")
 
             if not topic.consumed:
                 break

kstreams/test_utils/test_utils.py

@@ -1,14 +1,15 @@
-from .structs import RecordMetadata
-from .test_clients import TestConsumer, TestProducer
-from .topics import TopicManager
+import asyncio
+from types import TracebackType
+from typing import Any, Dict, List, Optional, Type
+
 from kstreams.create import create_engine
 from kstreams.custom_types import Headers
 from kstreams.serializers import ValueSerializer
 from kstreams.streams import Stream
-from types import TracebackType
-from typing import Any, Dict, List, Optional, Type
 
-import asyncio
+from .structs import RecordMetadata
+from .test_clients import TestConsumer, TestProducer
+from .topics import TopicManager
 
 
 class TestStreamClient:

kstreams/test_utils/topics.py

@@ -1,8 +1,8 @@
-from . import test_clients
+import asyncio
 from dataclasses import dataclass
 from typing import Any, ClassVar, Dict, Optional
 
-import asyncio
+from . import test_clients
 
 
 @dataclass

kstreams/utils.py

@@ -1,10 +1,10 @@
-from kstreams import custom_types
-from kstreams.conf import settings
-from tempfile import NamedTemporaryFile
-
 import contextlib
 import ssl
 import typing
+from tempfile import NamedTemporaryFile
+
+from kstreams import custom_types
+from kstreams.conf import settings
 
 
 class EmptySSLDataException(Exception):