fixing mypy issues

docs/examples/reliable_client/BestPracticesClient.py

@@ -61,7 +61,7 @@ async def make_producer(rabbitmq_data: dict) -> Producer | SuperStreamProducer:
 
     else:
 
-        producer = SuperStreamProducer(
+        producer = SuperStreamProducer(  # type: ignore
             host=host,
             username=username,
             password=password,
@@ -131,7 +131,7 @@ async def make_consumer(rabbitmq_data: dict) -> Consumer | SuperStreamConsumer:
 
     else:
 
-        consumer = SuperStreamConsumer(
+        consumer = SuperStreamConsumer(  # type: ignore
             host=host,
             username=username,
             password=password,
@@ -182,7 +182,7 @@ async def publish(rabbitmq_configuration: dict):
     producers = int(rabbitmq_configuration["Producers"])
     delay_sending_msg = int(rabbitmq_configuration["DelayDuringSendMs"])
 
-    producer = await make_producer(rabbitmq_configuration)
+    producer = await make_producer(rabbitmq_configuration)  # type: ignore
     await producer.start()
 
     # create a stream if it doesn't already exist
@@ -217,7 +217,7 @@ async def publish(rabbitmq_configuration: dict):
 
         else:
             try:
-                await producer.send(message=amqp_message, on_publish_confirm=_on_publish_confirm_client)
+                await producer.send(message=amqp_message, on_publish_confirm=_on_publish_confirm_client)  # type: ignore
             except Exception as ex:
                 print("exception while sending " + str(ex))
 
@@ -242,7 +242,7 @@ async def consume(rabbitmq_configuration: dict):
     consumers = int(rabbitmq_configuration["Consumers"])
     stream_name = rabbitmq_configuration["StreamName"]
 
-    consumer = await make_consumer(rabbitmq_configuration)
+    consumer = await make_consumer(rabbitmq_configuration)  # type: ignore
 
     # create a stream if it doesn't already exist
     if not is_super_stream_scenario:
@@ -260,7 +260,7 @@ async def consume(rabbitmq_configuration: dict):
                 offset_specification=offset_spec,
             )
     else:
-        await consumer.subscribe(callback=on_message, decoder=amqp_decoder, offset_specification=offset_spec)
+        await consumer.subscribe(callback=on_message, decoder=amqp_decoder, offset_specification=offset_spec)  # type: ignore
 
     await consumer.run()
 

pyproject.toml

@@ -27,6 +27,8 @@ mmh3 = "^4.0.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.0"
+requests = "^2.31.0"
+types-requests = "^2.31.0.20240406"
 
 [tool.black]
 line-length = 110

rstream/consumer.py

@@ -454,10 +454,12 @@ async def reconnect_stream(self, stream: str, offset: Optional[int] = None) -> N
             if stream == self._subscribers[subscriber_id].stream:
                 curr_subscriber = self._subscribers[subscriber_id]
                 curr_subscriber_id = subscriber_id
-        del self._subscribers[curr_subscriber_id]
+        if curr_subscriber_id is not None:
+            del self._subscribers[curr_subscriber_id]
 
         if stream in self._clients:
-            await self._clients[stream].free_available_id(curr_subscriber.subscription_id)
+            if curr_subscriber is not None:
+                await self._clients[stream].free_available_id(curr_subscriber.subscription_id)
             await self._clients[stream].close()
             del self._clients[stream]
 
@@ -519,7 +521,8 @@ async def _maybe_clean_up_during_lost_connection(self, stream: str):
 
         if stream in self._clients:
             await self._clients[stream].remove_stream(stream)
-            await self._clients[stream].free_available_id(curr_subscriber.subscription_id)
+            if curr_subscriber is not None:
+                await self._clients[stream].free_available_id(curr_subscriber.subscription_id)
             if await self._clients[stream].get_stream_count() == 0:
                 await self._clients[stream].close()
             del self._clients[stream]