Update ruff to 0.3.2, apply more linting rules (#986)

* Update ruff to 0.3.0
* Apply pyupgrade rules
* Apply flake8-2020, flake8-async
* Partially apply flake8-bandit, flake8-blind-except, flake8-bugbear
* Apply flake8-builtins, flake8-comprehensions
* Apply flake8-future-annotations and other flake rules
* Apply flake8-logging-format and others
* Apply flake8-no-pep420, flake8-pie
* Apply flake8-print
* Ignores for the rest of flake8-bandit
* Ignore for rest BLE; remove pylint ignores
* Fix problems found by flake8-bugbear
* Apply flake8-return
* Apply flake8-simplify
* Apply pygrep-hooks
* Apply Pylint rules
* Apply tryceratops
* Apply Perflint
* Apply flake8-logging, Ruff-specific
* Update ruff to 0.3.2
* Fix error found by github-advanced-security
* Return back to runtime-valid annotations
* Modern way to use metaclass
* Fix bootstrap_servers annotation
* Remove broken __hash__ method

Author: ods

aiokafka/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.10.0"  # noqa
+__version__ = "0.10.0"
 
 from .abc import ConsumerRebalanceListener
 from .client import AIOKafkaClient
@@ -16,6 +16,7 @@
     # Clients API
     "AIOKafkaProducer",
     "AIOKafkaConsumer",
+    "AIOKafkaClient",
     # ABC's
     "ConsumerRebalanceListener",
     # Errors
@@ -27,5 +28,3 @@
     "OffsetAndTimestamp",
     "OffsetAndMetadata",
 ]
-
-AIOKafkaClient

aiokafka/abc.py

@@ -63,7 +63,6 @@ def on_partitions_revoked(self, revoked):
             revoked (list(TopicPartition)): the partitions that were assigned
                 to the consumer on the last rebalance
         """
-        pass
 
     @abc.abstractmethod
     def on_partitions_assigned(self, assigned):
@@ -83,7 +82,6 @@ def on_partitions_assigned(self, assigned):
             assigned (list(TopicPartition)): the partitions assigned to the
                 consumer (may include partitions that were previously assigned)
         """
-        pass
 
 
 class AbstractTokenProvider(abc.ABC):
@@ -104,14 +102,11 @@ class AbstractTokenProvider(abc.ABC):
         https://docs.confluent.io/platform/current/kafka/authentication_sasl/authentication_sasl_oauth.html
     """
 
-    def __init__(self, **config):
-        pass
-
     @abc.abstractmethod
     async def token(self):
         """
         An async callback returning a :class:`str` ID/Access Token to be sent to
-        the Kafka client. In case where a synchoronous callback is needed,
+        the Kafka client. In case where a synchronous callback is needed,
         implementations like following can be used:
 
         .. code-block:: python
@@ -124,9 +119,8 @@ async def token(self):
                         None, self._token)
 
                 def _token(self):
-                    # The actual synchoronous token callback.
+                    # The actual synchronous token callback.
         """
-        pass
 
     def extensions(self):
         """

aiokafka/admin/client.py

@@ -2,7 +2,7 @@
 import logging
 from collections import defaultdict
 from ssl import SSLContext
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from aiokafka import __version__
 from aiokafka.client import AIOKafkaClient
@@ -85,7 +85,7 @@ def __init__(
         self,
         *,
         loop=None,
-        bootstrap_servers: str = "localhost",
+        bootstrap_servers: Union[str, List[str]] = "localhost",
         client_id: str = "aiokafka-" + __version__,
         request_timeout_ms: int = 40000,
         connections_max_idle_ms: int = 540000,
@@ -171,16 +171,15 @@ def _matching_api_version(self, operation: List[Request]) -> int:
         api_key = operation[0].API_KEY
         if not self._version_info or api_key not in self._version_info:
             raise IncompatibleBrokerVersion(
-                "Kafka broker does not support the '{}' Kafka protocol.".format(
-                    operation[0].__name__
-                )
+                f"Kafka broker does not support the '{operation[0].__name__}' "
+                "Kafka protocol."
             )
         min_version, max_version = self._version_info[api_key]
         version = min(len(operation) - 1, max_version)
         if version < min_version:
             raise IncompatibleBrokerVersion(
-                "No version of the '{}' Kafka protocol is supported by "
-                "both the client and broker.".format(operation[0].__name__)
+                f"No version of the '{operation[0].__name__}' Kafka protocol "
+                "is supported by both the client and broker."
             )
         return version
 
@@ -190,14 +189,8 @@ def _convert_new_topic_request(new_topic):
             new_topic.name,
             new_topic.num_partitions,
             new_topic.replication_factor,
-            [
-                (partition_id, replicas)
-                for partition_id, replicas in new_topic.replica_assignments.items()
-            ],
-            [
-                (config_key, config_value)
-                for config_key, config_value in new_topic.topic_configs.items()
-            ],
+            list(new_topic.replica_assignments.items()),
+            list(new_topic.topic_configs.items()),
         )
 
     async def create_topics(
@@ -223,9 +216,7 @@ async def create_topics(
             if validate_only:
                 raise IncompatibleBrokerVersion(
                     "validate_only requires CreateTopicsRequest >= v1, "
-                    "which is not supported by Kafka {}.".format(
-                        self._client.api_version
-                    )
+                    f"which is not supported by Kafka {self._client.api_version}."
                 )
             request = CreateTopicsRequest[version](
                 create_topic_requests=topics,
@@ -239,8 +230,8 @@ async def create_topics(
             )
         else:
             raise NotImplementedError(
-                "Support for CreateTopics v{} has not yet been added "
-                "to AIOKafkaAdminClient.".format(version)
+                f"Support for CreateTopics v{version} has not yet been added "
+                "to AIOKafkaAdminClient."
             )
         response = await self._client.send(self._client.get_random_node(), request)
         return response
@@ -317,7 +308,7 @@ async def describe_configs(
         if version == 0 and include_synonyms:
             raise IncompatibleBrokerVersion(
                 "include_synonyms requires DescribeConfigsRequest >= v1,"
-                " which is not supported by Kafka {}.".format(self._client.api_version)
+                f" which is not supported by Kafka {self._client.api_version}."
             )
         broker_res, topic_res = self._convert_config_resources(
             config_resources,
@@ -462,7 +453,7 @@ async def describe_consumer_groups(
             raise IncompatibleBrokerVersion(
                 "include_authorized_operations requests "
                 "DescribeGroupsRequest >= v3, which is not "
-                "supported by Kafka {}".format(version)
+                f"supported by Kafka {version}"
             )
         req_class = DescribeGroupsRequest[version]
         futures = []

aiokafka/client.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextlib
 import logging
 import random
 import time
@@ -188,16 +189,14 @@ def hosts(self):
     async def close(self):
         if self._sync_task:
             self._sync_task.cancel()
-            try:
+            with contextlib.suppress(asyncio.CancelledError):
                 await self._sync_task
-            except asyncio.CancelledError:
-                pass
             self._sync_task = None
         # Be careful to wait for graceful closure of all connections, so we
         # process all pending buffers.
-        futs = []
-        for conn in self._conns.values():
-            futs.append(conn.close(reason=CloseReason.SHUTDOWN))
+        futs = [
+            conn.close(reason=CloseReason.SHUTDOWN) for conn in self._conns.values()
+        ]
         if futs:
             await asyncio.gather(*futs)
 
@@ -231,7 +230,7 @@ async def bootstrap(self):
                     sasl_mechanism=self._sasl_mechanism,
                     sasl_plain_username=self._sasl_plain_username,
                     sasl_plain_password=self._sasl_plain_password,
-                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,  # noqa: E501
+                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,
                     sasl_kerberos_domain_name=self._sasl_kerberos_domain_name,
                     sasl_oauth_token_provider=self._sasl_oauth_token_provider,
                     version_hint=version_hint,
@@ -370,7 +369,7 @@ def force_metadata_update(self):
             if not self._md_update_waiter.done():
                 self._md_update_waiter.set_result(None)
             self._md_update_fut = self._loop.create_future()
-        # Metadata will be updated in the background by syncronizer
+        # Metadata will be updated in the background by synchronizer
         return asyncio.shield(self._md_update_fut)
 
     async def fetch_all_metadata(self):
@@ -413,10 +412,7 @@ def _on_connection_closed(self, conn, reason):
         """Callback called when connection is closed"""
         # Connection failures imply that our metadata is stale, so let's
         # refresh
-        if (
-            reason == CloseReason.CONNECTION_BROKEN
-            or reason == CloseReason.CONNECTION_TIMEOUT
-        ):
+        if reason in [CloseReason.CONNECTION_BROKEN, CloseReason.CONNECTION_TIMEOUT]:
             self.force_metadata_update()
 
     async def _get_conn(self, node_id, *, group=ConnectionGroup.DEFAULT, no_hint=False):
@@ -471,7 +467,7 @@ async def _get_conn(self, node_id, *, group=ConnectionGroup.DEFAULT, no_hint=Fal
                     sasl_mechanism=self._sasl_mechanism,
                     sasl_plain_username=self._sasl_plain_username,
                     sasl_plain_password=self._sasl_plain_password,
-                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,  # noqa: E501
+                    sasl_kerberos_service_name=self._sasl_kerberos_service_name,
                     sasl_kerberos_domain_name=self._sasl_kerberos_domain_name,
                     sasl_oauth_token_provider=self._sasl_oauth_token_provider,
                     version_hint=version_hint,
@@ -511,7 +507,7 @@ async def send(self, node_id, request, *, group=ConnectionGroup.DEFAULT):
         if not (await self.ready(node_id, group=group)):
             raise NodeNotReadyError(
                 "Attempt to send a request to node"
-                " which is not ready (node id {}).".format(node_id)
+                f" which is not ready (node id {node_id})."
             )
 
         # Every request gets a response, except one special case:
@@ -524,10 +520,10 @@ async def send(self, node_id, request, *, group=ConnectionGroup.DEFAULT):
         )
         try:
             result = await future
-        except asyncio.TimeoutError:
+        except asyncio.TimeoutError as exc:
             # close connection so it is renewed in next request
             self._conns[(node_id, group)].close(reason=CloseReason.CONNECTION_TIMEOUT)
-            raise RequestTimedOutError()
+            raise RequestTimedOutError() from exc
         else:
             return result
 
@@ -536,14 +532,14 @@ async def check_version(self, node_id=None):
         if node_id is None:
             default_group_conns = [
                 n_id
-                for (n_id, group) in self._conns.keys()
+                for (n_id, group) in self._conns
                 if group == ConnectionGroup.DEFAULT
             ]
             if default_group_conns:
                 node_id = default_group_conns[0]
             else:
                 assert self.cluster.brokers(), "no brokers in metadata"
-                node_id = list(self.cluster.brokers())[0].nodeId
+                node_id = next(iter(self.cluster.brokers())).nodeId
 
         from aiokafka.protocol.admin import ApiVersionRequest_v0, ListGroupsRequest_v0
         from aiokafka.protocol.commit import (
@@ -577,14 +573,12 @@ async def check_version(self, node_id=None):
                 # so we send metadata request and wait response
                 task = create_task(conn.send(request))
                 await asyncio.wait([task], timeout=0.1)
-                try:
+                # metadata request can be cancelled in case
+                # of invalid correlationIds order
+                with contextlib.suppress(KafkaError):
                     await conn.send(MetadataRequest_v0([]))
-                except KafkaError:
-                    # metadata request can be cancelled in case
-                    # of invalid correlationIds order
-                    pass
                 response = await task
-            except KafkaError:
+            except KafkaError:  # noqa: PERF203
                 continue
             else:
                 # To avoid having a connection in undefined state
@@ -593,7 +587,7 @@ async def check_version(self, node_id=None):
                 if isinstance(request, ApiVersionRequest_v0):
                     # Starting from 0.10 kafka broker we determine version
                     # by looking at ApiVersionResponse
-                    version = self._check_api_version_response(response)
+                    return self._check_api_version_response(response)
                 return version
 
         raise UnrecognizedBrokerVersion()

aiokafka/cluster.py

@@ -128,21 +128,20 @@ def available_partitions_for_topic(self, topic):
         """
         if topic not in self._partitions:
             return None
-        return set(
-            [
-                partition
-                for partition, metadata in self._partitions[topic].items()
-                if metadata.leader != -1
-            ]
-        )
+        return {
+            partition
+            for partition, metadata in self._partitions[topic].items()
+            if metadata.leader != -1
+        }
 
     def leader_for_partition(self, partition):
         """Return node_id of leader, -1 unavailable, None if unknown."""
         if partition.topic not in self._partitions:
             return None
-        elif partition.partition not in self._partitions[partition.topic]:
+        partitions = self._partitions[partition.topic]
+        if partition.partition not in partitions:
             return None
-        return self._partitions[partition.topic][partition.partition].leader
+        return partitions[partition.partition].leader
 
     def partitions_for_broker(self, broker_id):
         """Return TopicPartitions for which the broker is a leader.
@@ -222,7 +221,8 @@ def update_metadata(self, metadata):
         """
         if not metadata.brokers:
             log.warning("No broker metadata found in MetadataResponse -- ignoring.")
-            return self.failed_update(Errors.MetadataEmptyBrokerList(metadata))
+            self.failed_update(Errors.MetadataEmptyBrokerList(metadata))
+            return
 
         _new_brokers = {}
         for broker in metadata.brokers:
@@ -342,11 +342,11 @@ def add_group_coordinator(self, group, response):
         if error_type is not Errors.NoError:
             log.error("GroupCoordinatorResponse error: %s", error_type)
             self._groups[group] = -1
-            return
+            return None
 
         # Use a coordinator-specific node id so that group requests
         # get a dedicated connection
-        node_id = "coordinator-{}".format(response.coordinator_id)
+        node_id = f"coordinator-{response.coordinator_id}"
         coordinator = BrokerMetadata(node_id, response.host, response.port, None)
 
         log.info("Group coordinator for %s is %s", group, coordinator)

aiokafka/codec.py

@@ -1,8 +1,7 @@
-from __future__ import annotations
-
 import gzip
 import io
 import struct
+from typing import Optional
 
 from typing_extensions import Buffer
 
@@ -32,7 +31,7 @@ def has_lz4() -> bool:
     return cramjam is not None
 
 
-def gzip_encode(payload: Buffer, compresslevel: int | None = None) -> bytes:
+def gzip_encode(payload: Buffer, compresslevel: Optional[int] = None) -> bytes:
     if not compresslevel:
         compresslevel = 9
 
@@ -188,7 +187,7 @@ def lz4_decode(payload: Buffer) -> bytes:
     return bytes(cramjam.lz4.decompress(payload))
 
 
-def zstd_encode(payload: Buffer, level: int | None = None) -> bytes:
+def zstd_encode(payload: Buffer, level: Optional[int] = None) -> bytes:
     if not has_zstd():
         raise NotImplementedError("Zstd codec is not available")