fix: mypy updated from version 0.9 to 1.8.0. Some typing fixed. Closes …

…#159 (#165)

kstreams/__init__.py

@@ -1,6 +1,6 @@
 from aiokafka.structs import ConsumerRecord, RecordMetadata, TopicPartition
 
-from .clients import Consumer, ConsumerType, Producer, ProducerType
+from .clients import Consumer, Producer
 from .create import StreamEngine, create_engine
 from .prometheus.monitor import PrometheusMonitor, PrometheusMonitorType
 from .rebalance_listener import (
@@ -14,9 +14,7 @@
 
 __all__ = [
     "Consumer",
-    "ConsumerType",
     "Producer",
-    "ProducerType",
     "StreamEngine",
     "create_engine",
     "PrometheusMonitor",

kstreams/clients.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Callable, Optional, TypeVar
+from typing import Callable, Optional
 
 import aiokafka
 
@@ -40,7 +40,3 @@ def key_serializer(key):
                     return key.encode("utf-8")
 
         super().__init__(*args, key_serializer=key_serializer, **self.config)
-
-
-ConsumerType = TypeVar("ConsumerType", bound=Consumer)
-ProducerType = TypeVar("ProducerType", bound=Producer)

kstreams/create.py

@@ -1,7 +1,7 @@
 from typing import Optional, Type
 
 from .backends.kafka import Kafka
-from .clients import Consumer, ConsumerType, Producer, ProducerType
+from .clients import Consumer, Producer
 from .engine import StreamEngine
 from .prometheus.monitor import PrometheusMonitor
 from .serializers import Deserializer, Serializer
@@ -10,8 +10,8 @@
 def create_engine(
     title: Optional[str] = None,
     backend: Optional[Kafka] = None,
-    consumer_class: Type[ConsumerType] = Consumer,
-    producer_class: Type[ProducerType] = Producer,
+    consumer_class: Type[Consumer] = Consumer,
+    producer_class: Type[Producer] = Producer,
     serializer: Optional[Serializer] = None,
     deserializer: Optional[Deserializer] = None,
     monitor: Optional[PrometheusMonitor] = None,

kstreams/engine.py

@@ -7,7 +7,7 @@
 from kstreams.structs import TopicPartitionOffset
 
 from .backends.kafka import Kafka
-from .clients import ConsumerType, ProducerType
+from .clients import Consumer, Producer
 from .exceptions import DuplicateStreamException, EngineNotStartedException
 from .middleware import ExceptionMiddleware, Middleware
 from .prometheus.monitor import PrometheusMonitor
@@ -61,8 +61,8 @@ def __init__(
         self,
         *,
         backend: Kafka,
-        consumer_class: typing.Type[ConsumerType],
-        producer_class: typing.Type[ProducerType],
+        consumer_class: typing.Type[Consumer],
+        producer_class: typing.Type[Producer],
         monitor: PrometheusMonitor,
         title: typing.Optional[str] = None,
         deserializer: typing.Optional[Deserializer] = None,
@@ -75,7 +75,7 @@ def __init__(
         self.deserializer = deserializer
         self.serializer = serializer
         self.monitor = monitor
-        self._producer: typing.Optional[typing.Type[ProducerType]] = None
+        self._producer: typing.Optional[typing.Type[Producer]] = None
         self._streams: typing.List[Stream] = []
 
     async def send(

kstreams/prometheus/monitor.py

@@ -5,7 +5,7 @@
 from prometheus_client import Gauge
 
 from kstreams import TopicPartition
-from kstreams.clients import ConsumerType
+from kstreams.clients import Consumer
 from kstreams.streams import Stream
 
 logger = logging.getLogger(__name__)
@@ -164,7 +164,7 @@ def add_producer(self, producer):
     def add_streams(self, streams):
         self._streams = streams
 
-    async def generate_consumer_metrics(self, consumer: ConsumerType):
+    async def generate_consumer_metrics(self, consumer: Consumer):
         """
         Generate Consumer Metrics for Prometheus
 

kstreams/streams.py

@@ -13,7 +13,7 @@
 from kstreams.structs import TopicPartitionOffset
 
 from .backends.kafka import Kafka
-from .clients import Consumer, ConsumerType
+from .clients import Consumer
 from .middleware import Middleware
 from .rebalance_listener import RebalanceListener
 from .serializers import Deserializer
@@ -86,7 +86,7 @@ def __init__(
         *,
         func: StreamFunc,
         backend: typing.Optional[Kafka] = None,
-        consumer_class: typing.Type[ConsumerType] = Consumer,
+        consumer_class: typing.Type[Consumer] = Consumer,
         name: typing.Optional[str] = None,
         config: typing.Optional[typing.Dict] = None,
         deserializer: typing.Optional[Deserializer] = None,
@@ -97,7 +97,7 @@ def __init__(
         self.func = func
         self.backend = backend
         self.consumer_class = consumer_class
-        self.consumer: typing.Optional[ConsumerType] = None
+        self.consumer: typing.Optional[Consumer] = None
         self.config = config or {}
         self._consumer_task: typing.Optional[asyncio.Task] = None
         self.name = name or str(uuid.uuid4())
@@ -114,7 +114,7 @@ def __init__(
         # so we always create a list and then we expand it with *topics
         self.topics = [topics] if isinstance(topics, str) else topics
 
-    def _create_consumer(self) -> ConsumerType:
+    def _create_consumer(self) -> Consumer:
         if self.backend is None:
             raise BackendNotSet("A backend has not been set for this stream")
         config = {**self.backend.model_dump(), **self.config}
@@ -240,8 +240,8 @@ async def func_wrapper_with_typing(self) -> None:
             cr = await self.getone()
             await self.func(cr)
 
-    def seek_to_initial_offsets(self):
-        if not self.seeked_initial_offsets:
+    def seek_to_initial_offsets(self) -> None:
+        if not self.seeked_initial_offsets and self.consumer is not None:
             assignments: typing.Set[TopicPartition] = self.consumer.assignment()
             if self.initial_offsets is not None:
                 topicPartitionOffset: TopicPartitionOffset

kstreams/utils.py

@@ -1,7 +1,7 @@
 import contextlib
 import ssl
 from tempfile import NamedTemporaryFile
-from typing import Optional, Union
+from typing import Any, Optional, Union
 
 from aiokafka.helpers import create_ssl_context as aiokafka_create_ssl_context
 
@@ -53,13 +53,13 @@ def create_ssl_context_from_mem(
 
 def create_ssl_context(
     *,
-    cafile: str = None,
-    capath: str = None,
-    cadata: Union[str, bytes] = None,
-    certfile: str = None,
-    keyfile: str = None,
-    password: str = None,
-    crlfile=None,
+    cafile: Optional[str] = None,
+    capath: Optional[str] = None,
+    cadata: Union[str, bytes, None] = None,
+    certfile: Optional[str] = None,
+    keyfile: Optional[str] = None,
+    password: Optional[str] = None,
+    crlfile: Any = None,
 ):
     """Wrapper of [aiokafka.helpers.create_ssl_context](
         https://aiokafka.readthedocs.io/en/stable/api.html#helpers

pyproject.toml

@@ -28,7 +28,7 @@ pydantic = ">=2.0.0,<3.0.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.1"
-mypy = "^0.971"
+mypy = "^1.8.0"
 pytest-httpserver = "<1.0"
 pytest-cov = "^2.11.1"
 jedi = "0.17.2"