pyupgrade the code (#752)

* Run pyupgrade with `--py36-plus` option

Since the **aiokafka** package is not compatible
with Python versions lower than 3.6, the codebase can be upgraded.

* Replace formats with f-strings.

* Further replacements of formats with f-strings

* Fix lint errors.

Author: pawelrubin

aiokafka/client.py

@@ -246,7 +246,7 @@ async def bootstrap(self):
             break
         else:
             raise KafkaConnectionError(
-                'Unable to bootstrap from {}'.format(self.hosts))
+                f'Unable to bootstrap from {self.hosts}')
 
         # detect api version if need
         if self._api_version == 'auto':
@@ -544,12 +544,12 @@ async def check_version(self, node_id=None):
         conn = await self._get_conn(node_id, no_hint=True)
         if conn is None:
             raise KafkaConnectionError(
-                "No connection to node with id {}".format(node_id))
+                f"No connection to node with id {node_id}")
         for version, request in test_cases:
             try:
                 if not conn.connected():
                     await conn.connect()
-                assert conn, 'no connection to node with id {}'.format(node_id)
+                assert conn, f'no connection to node with id {node_id}'
                 # request can be ignored by Kafka broker,
                 # so we send metadata request and wait response
                 task = create_task(conn.send(request))
@@ -591,10 +591,10 @@ def _check_api_version_response(self, response):
 
         error_type = Errors.for_code(response.error_code)
         assert error_type is Errors.NoError, "API version check failed"
-        max_versions = dict([
-            (api_key, max_version)
+        max_versions = {
+            api_key: max_version
             for api_key, _, max_version in response.api_versions
-        ])
+        }
         # Get the best match of test cases
         for broker_version, api_key, version in test_cases:
             if max_versions.get(api_key, -1) >= version:

aiokafka/conn.py

@@ -63,8 +63,7 @@ def pick_best(self, request_versions):
                         req_klass.API_VERSION <= supported_versions[1]:
                     return req_klass
         raise Errors.KafkaError(
-            "Could not pick a version for API_KEY={} from {}. ".format(
-                api_key, supported_versions)
+            f"Could not pick a version for API_KEY={api_key} from {supported_versions}."
         )
 
 
@@ -184,7 +183,7 @@ def __init__(self, host, port, *, client_id='aiokafka',
     # that
     def __del__(self, _warnings=warnings):
         if self.connected():
-            _warnings.warn("Unclosed AIOKafkaConnection {!r}".format(self),
+            _warnings.warn(f"Unclosed AIOKafkaConnection {self!r}",
                            ResourceWarning,
                            source=self)
             if self._loop.is_closed():
@@ -242,8 +241,8 @@ async def _do_version_lookup(self):
         versions = {}
         for api_key, min_version, max_version in response.api_versions:
             assert min_version <= max_version, (
-                "{} should be less than or equal to {} for {}".format(
-                    min_version, max_version, api_key)
+                f"{min_version} should be less than"
+                f" or equal to {max_version} for {api_key}"
             )
             versions[api_key] = (min_version, max_version)
         self._version_info = VersionInfo(versions)
@@ -270,9 +269,9 @@ async def _do_sasl_handshake(self):
 
             if self._sasl_mechanism not in response.enabled_mechanisms:
                 exc = Errors.UnsupportedSaslMechanismError(
-                    'Kafka broker does not support %s sasl mechanism. '
-                    'Enabled mechanisms are: %s'
-                    % (self._sasl_mechanism, response.enabled_mechanisms))
+                    f"Kafka broker does not support {self._sasl_mechanism} sasl "
+                    f"mechanism. Enabled mechanisms are: {response.enabled_mechanisms}"
+                )
                 self.close(reason=CloseReason.AUTH_FAILURE, exc=exc)
                 raise exc
 
@@ -364,7 +363,7 @@ def sasl_principal(self):
         service = self._sasl_kerberos_service_name
         domain = self._sasl_kerberos_domain_name or self.host
 
-        return "{service}@{domain}".format(service=service, domain=domain)
+        return f"{service}@{domain}"
 
     @staticmethod
     def _on_read_task_error(self_ref, read_task):
@@ -401,7 +400,7 @@ def _idle_check(self_ref):
                 wake_up_in, self._idle_check, self_ref)
 
     def __repr__(self):
-        return "<AIOKafkaConnection host={0.host} port={0.port}>".format(self)
+        return f"<AIOKafkaConnection host={self.host} port={self.port}>"
 
     @property
     def host(self):
@@ -414,8 +413,8 @@ def port(self):
     def send(self, request, expect_response=True):
         if self._writer is None:
             raise Errors.KafkaConnectionError(
-                "No connection to broker at {0}:{1}"
-                .format(self._host, self._port))
+                f"No connection to broker at {self._host}:{self._port}"
+            )
 
         correlation_id = self._next_correlation_id()
         header = RequestHeader(request,
@@ -428,8 +427,8 @@ def send(self, request, expect_response=True):
         except OSError as err:
             self.close(reason=CloseReason.CONNECTION_BROKEN)
             raise Errors.KafkaConnectionError(
-                "Connection at {0}:{1} broken: {2}".format(
-                    self._host, self._port, err))
+                f"Connection at {self._host}:{self._port} broken: {err}"
+            )
 
         self.log.debug(
             '%s Request %d: %s', self, correlation_id, request)
@@ -443,17 +442,16 @@ def send(self, request, expect_response=True):
     def _send_sasl_token(self, payload, expect_response=True):
         if self._writer is None:
             raise Errors.KafkaConnectionError(
-                "No connection to broker at {0}:{1}"
-                .format(self._host, self._port))
-
+                f"No connection to broker at {self._host}:{self._port}"
+            )
         size = struct.pack(">i", len(payload))
         try:
             self._writer.write(size + payload)
         except OSError as err:
             self.close(reason=CloseReason.CONNECTION_BROKEN)
             raise Errors.KafkaConnectionError(
-                "Connection at {0}:{1} broken: {2}".format(
-                    self._host, self._port, err))
+                f"Connection at {self._host}:{self._port} broken: {err}"
+            )
 
         if not expect_response:
             return self._writer.drain()
@@ -476,8 +474,8 @@ def close(self, reason=None, exc=None):
             for _, _, fut in self._requests:
                 if not fut.done():
                     error = Errors.KafkaConnectionError(
-                        "Connection at {0}:{1} closed".format(
-                            self._host, self._port))
+                        f"Connection at {self._host}:{self._port} closed"
+                    )
                     if exc is not None:
                         error.__cause__ = exc
                         error.__context__ = exc
@@ -536,8 +534,9 @@ def _handle_frame(self, resp):
 
             elif correlation_id != recv_correlation_id:
                 error = Errors.CorrelationIdError(
-                    'Correlation ids do not match: sent {}, recv {}'
-                    .format(correlation_id, recv_correlation_id))
+                    f"Correlation ids do not match: sent {correlation_id},"
+                    f" recv {recv_correlation_id}"
+                )
                 if not fut.done():
                     fut.set_exception(error)
                 self.close(reason=CloseReason.OUT_OF_SYNC)
@@ -660,8 +659,7 @@ def __init__(self, *, loop, sasl_plain_password,
         self._authenticator = self.authenticator_scram()
 
     def first_message(self):
-        client_first_bare = 'n={},r={}'.format(
-            self._sasl_plain_username, self._nonce)
+        client_first_bare = f"n={self._sasl_plain_username},r={self._nonce}"
         self._auth_message += client_first_bare
         return 'n,,' + client_first_bare
 
@@ -689,8 +687,8 @@ def process_server_first_message(self, server_first):
             self._server_key, self._auth_message.encode('utf-8'))
 
     def final_message(self):
-        return 'c=biws,r={},p={}'.format(
-            self._nonce, base64.b64encode(self._client_proof).decode('utf-8'))
+        client_proof = base64.b64encode(self._client_proof).decode('utf-8')
+        return f"c=biws,r={self._nonce},p={client_proof}"
 
     def process_server_final_message(self, server_final):
         params = dict(pair.split('=', 1) for pair in server_final.split(','))
@@ -734,9 +732,7 @@ async def step(self, payload):
             .encode("utf-8"), True
 
     def _build_oauth_client_request(self, token, token_extensions):
-        return "n,,\x01auth=Bearer {}{}\x01\x01".format(
-            token, token_extensions
-            )
+        return f"n,,\x01auth=Bearer {token}{token_extensions}\x01\x01"
 
     def _token_extensions(self):
         """
@@ -751,7 +747,7 @@ def _token_extensions(self):
             extensions = self._sasl_oauth_token_provider.extensions()
             if len(extensions) > 0:
                 msg = "\x01".join(
-                    ["{}={}".format(k, v) for k, v in extensions.items()])
+                    [f"{k}={v}" for k, v in extensions.items()])
                 return "\x01" + msg
 
         return ""

aiokafka/consumer/consumer.py

@@ -27,7 +27,7 @@
 log = logging.getLogger(__name__)
 
 
-class AIOKafkaConsumer(object):
+class AIOKafkaConsumer:
     """
     A client that consumes records from a Kafka cluster.
 
@@ -322,7 +322,7 @@ def __init__(self, *topics, loop=None,
 
     def __del__(self, _warnings=warnings):
         if self._closed is False:
-            _warnings.warn("Unclosed AIOKafkaConsumer {!r}".format(self),
+            _warnings.warn(f"Unclosed AIOKafkaConsumer {self!r}",
                            ResourceWarning,
                            source=self)
             context = {'consumer': self,
@@ -346,8 +346,9 @@ async def start(self):
         await self._wait_topics()
 
         if self._client.api_version < (0, 9):
-            raise ValueError("Unsupported Kafka version: {}".format(
-                self._client.api_version))
+            raise ValueError(
+                f"Unsupported Kafka version: {self._client.api_version}"
+            )
 
         if self._isolation_level == "read_committed" and \
                 self._client.api_version < (0, 11):
@@ -549,7 +550,7 @@ async def commit(self, offsets=None):
             for tp in offsets:
                 if tp not in assignment.tps:
                     raise IllegalStateError(
-                        "Partition {} is not assigned".format(tp))
+                        f"Partition {tp} is not assigned")
 
         await self._coordinator.commit_offsets(assignment, offsets)
 
@@ -630,7 +631,7 @@ async def position(self, partition):
         while True:
             if not self._subscription.is_assigned(partition):
                 raise IllegalStateError(
-                    'Partition {} is not assigned'.format(partition))
+                    f'Partition {partition} is not assigned')
 
             assignment = self._subscription.subscription.assignment
             tp_state = assignment.state_value(partition)
@@ -645,7 +646,7 @@ async def position(self, partition):
                 if not tp_state.has_valid_position:
                     if self._subscription.subscription is None:
                         raise IllegalStateError(
-                            'Partition {} is not assigned'.format(partition))
+                            f'Partition {partition} is not assigned')
                     if self._subscription.subscription.assignment is None:
                         self._coordinator.check_errors()
                         await self._subscription.wait_for_assignment()
@@ -772,7 +773,7 @@ async def seek_to_beginning(self, *partitions):
             )
             if not_assigned:
                 raise IllegalStateError(
-                    "Partitions {} are not assigned".format(not_assigned))
+                    f"Partitions {not_assigned} are not assigned")
 
         for tp in partitions:
             log.debug("Seeking to beginning of partition %s", tp)
@@ -814,7 +815,7 @@ async def seek_to_end(self, *partitions):
             )
             if not_assigned:
                 raise IllegalStateError(
-                    "Partitions {} are not assigned".format(not_assigned))
+                    f"Partitions {not_assigned} are not assigned")
 
         for tp in partitions:
             log.debug("Seeking to end of partition %s", tp)
@@ -861,7 +862,7 @@ async def seek_to_committed(self, *partitions):
             )
             if not_assigned:
                 raise IllegalStateError(
-                    "Partitions {} are not assigned".format(not_assigned))
+                    f"Partitions {not_assigned} are not assigned")
 
         committed_offsets = {}
         for tp in partitions:
@@ -909,14 +910,16 @@ async def offsets_for_times(self, timestamps):
         """
         if self._client.api_version <= (0, 10, 0):
             raise UnsupportedVersionError(
-                "offsets_for_times API not supported for cluster version {}"
-                .format(self._client.api_version))
+                "offsets_for_times API not supported"
+                f" for cluster version {self._client.api_version}"
+            )
         for tp, ts in timestamps.items():
             timestamps[tp] = int(ts)
             if ts < 0:
                 raise ValueError(
-                    "The target time for partition {} is {}. The target time "
-                    "cannot be negative.".format(tp, ts))
+                    f"The target time for partition {tp} is {ts}."
+                    " The target time cannot be negative."
+                )
         offsets = await self._fetcher.get_offsets_by_times(
             timestamps, self._request_timeout_ms)
         return offsets
@@ -948,8 +951,9 @@ async def beginning_offsets(self, partitions):
         """
         if self._client.api_version <= (0, 10, 0):
             raise UnsupportedVersionError(
-                "offsets_for_times API not supported for cluster version {}"
-                .format(self._client.api_version))
+                "offsets_for_times API not supported"
+                f" for cluster version {self._client.api_version}"
+            )
         offsets = await self._fetcher.beginning_offsets(
             partitions, self._request_timeout_ms)
         return offsets
@@ -983,8 +987,9 @@ async def end_offsets(self, partitions):
         """
         if self._client.api_version <= (0, 10, 0):
             raise UnsupportedVersionError(
-                "offsets_for_times API not supported for cluster version {}"
-                .format(self._client.api_version))
+                "offsets_for_times API not supported"
+                f" for cluster version {self._client.api_version}"
+            )
         offsets = await self._fetcher.end_offsets(
             partitions, self._request_timeout_ms)
         return offsets
@@ -1044,7 +1049,7 @@ def subscribe(self, topics=(), pattern=None, listener=None):
                 pattern = re.compile(pattern)
             except re.error as err:
                 raise ValueError(
-                    "{!r} is not a valid pattern: {}".format(pattern, err))
+                    f"{pattern!r} is not a valid pattern: {err}")
             self._subscription.subscribe_pattern(
                 pattern=pattern, listener=listener)
             # NOTE: set_topics will trigger a rebalance, so the coordinator

aiokafka/consumer/fetcher.py

@@ -53,7 +53,7 @@ def to_str(cls, value):
         if value == cls.NONE:
             return "none"
         else:
-            return "timestamp({})".format(value)
+            return f"timestamp({value})"
 
 
 class FetchResult:
@@ -145,8 +145,7 @@ def has_more(self):
         return self._partition_records is not None
 
     def __repr__(self):
-        return "<FetchResult position={!r}>".format(
-            self._partition_records.next_fetch_offset)
+        return f"<FetchResult position={self._partition_records.next_fetch_offset!r}>"
 
 
 class FetchError:
@@ -166,7 +165,7 @@ def check_raise(self):
         raise self._error
 
     def __repr__(self):
-        return "<FetchError error={!r}>".format(self._error)
+        return f"<FetchError error={self._error!r}>"
 
 
 class PartitionRecords:
@@ -213,7 +212,7 @@ def _unpack_records(self):
                 # This iterator will be closed after the exception, so we don't
                 # try to drain other batches here. They will be refetched.
                 raise Errors.CorruptRecordException(
-                    "Invalid CRC - {tp}".format(tp=tp))
+                    f"Invalid CRC - {tp}")
 
             if self._isolation_level == READ_COMMITTED and \
                     next_batch.producer_id is not None:
@@ -389,7 +388,7 @@ def __init__(
             self._isolation_level = READ_COMMITTED
         else:
             raise ValueError(
-                "Incorrect isolation level {}".format(isolation_level))
+                f"Incorrect isolation level {isolation_level}")
 
         self._records = collections.OrderedDict()
         self._in_flight = set()