enable pre-commit

.pre-commit-config.yaml

@@ -0,0 +1,29 @@
+repos:
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v2.21.2
+    hooks:
+      - id: pyupgrade
+        args: ["--py36-plus"]
+
+  - repo: https://gitlab.com/pycqa/flake8
+    rev: 3.9.2
+    hooks:
+      - id: flake8
+
+  - repo: https://github.com/asottile/yesqa
+    rev: v1.2.3
+    hooks:
+      - id: yesqa
+
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.0.1
+    hooks:
+      - id: check-merge-conflict
+      - id: check-toml
+      - id: check-yaml
+      - id: mixed-line-ending
+
+  - repo: https://github.com/pycqa/isort
+    rev: 5.9.2
+    hooks:
+      - id: isort

conftest.py

@@ -1,17 +1,29 @@
 import pytest
+
+
 def pytest_addoption(parser):
-    parser.addoption("-E", action="append", metavar="NAME",
-        help="only run tests matching the environment NAME.")
+    parser.addoption(
+        "-E",
+        action="append",
+        metavar="NAME",
+        help="only run tests matching the environment NAME.",
+    )
+
 
 def pytest_configure(config):
     # register an additional marker
-    config.addinivalue_line("markers",
-        "env(name): mark test to run only on named environment")
+    config.addinivalue_line(
+        "markers",
+        "env(name): mark test to run only on named environment",
+    )
+
 
 def pytest_runtest_setup(item):
     envnames = [mark.args[0] for mark in item.iter_markers(name='env')]
     if envnames:
-        if item.config.getoption("-E") is None or len(set(item.config.getoption("-E")) & set(envnames)) == 0:
+        if (
+            item.config.getoption("-E") is None
+            or len(set(item.config.getoption("-E")) & set(envnames)) == 0
+        ):
             # We skip test if does not mentioned by -E param
             pytest.skip("test requires env in %r" % envnames)
-

examples/complete_receive.py

@@ -6,7 +6,7 @@
 
 from pprint import pformat
 
-from kombu import Connection, Exchange, Queue, Consumer, eventloop
+from kombu import Connection, Consumer, Exchange, Queue, eventloop
 
 #: By default messages sent to exchanges are persistent (delivery_mode=2),
 #: and queues and exchanges are durable.
@@ -21,10 +21,11 @@ def pretty(obj):
 #: This is the callback applied when a message is received.
 def handle_message(body, message):
     print(f'Received message: {body!r}')
-    print('  properties:\n{}'.format(pretty(message.properties)))
-    print('  delivery_info:\n{}'.format(pretty(message.delivery_info)))
+    print(f'  properties:\n{pretty(message.properties)}')
+    print(f'  delivery_info:\n{pretty(message.delivery_info)}')
     message.ack()
 
+
 #: Create a connection and a channel.
 #: If hostname, userid, password and virtual_host is not specified
 #: the values below are the default, but listed here so it can

examples/complete_send.py

@@ -6,7 +6,7 @@
 
 """
 
-from kombu import Connection, Producer, Exchange, Queue
+from kombu import Connection, Exchange, Producer, Queue
 
 #: By default messages sent to exchanges are persistent (delivery_mode=2),
 #: and queues and exchanges are durable.
@@ -25,7 +25,10 @@
     #: Publish the message using the json serializer (which is the default),
     #: and zlib compression.  The kombu consumer will automatically detect
     #: encoding, serialization and compression used and decode accordingly.
-    producer.publish({'hello': 'world'},
-                     exchange=exchange,
-                     routing_key='kombu_demo',
-                     serializer='json', compression='zlib')
+    producer.publish(
+        {'hello': 'world'},
+        exchange=exchange,
+        routing_key='kombu_demo',
+        serializer='json',
+        compression='zlib',
+    )

examples/experimental/async_consume.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-from kombu import Connection, Exchange, Queue, Producer, Consumer
+from kombu import Connection, Consumer, Exchange, Producer, Queue
 from kombu.asynchronous import Hub
 
 hub = Hub()

examples/hello_consumer.py

@@ -1,5 +1,4 @@
-from kombu import Connection  # noqa
-
+from kombu import Connection
 
 with Connection('amqp://guest:guest@localhost:5672//') as conn:
     simple_queue = conn.SimpleQueue('simple_queue')

examples/hello_publisher.py

@@ -2,7 +2,6 @@
 
 from kombu import Connection
 
-
 with Connection('amqp://guest:guest@localhost:5672//') as conn:
     simple_queue = conn.SimpleQueue('simple_queue')
     message = f'helloworld, sent at {datetime.datetime.today()}'

examples/memory_transport.py

@@ -3,7 +3,7 @@
 """
 import time
 
-from kombu import Connection, Exchange, Queue, Consumer
+from kombu import Connection, Consumer, Exchange, Queue
 
 media_exchange = Exchange('media', 'direct')
 video_queue = Queue('video', exchange=media_exchange, routing_key='video')
@@ -19,6 +19,11 @@ def handle_message(body, message):
 consumer = Consumer(connection, task_queues, callbacks=[handle_message])
 
 producer = connection.Producer(serializer='json')
-producer.publish({"foo": "bar"}, exchange=media_exchange, routing_key='video', declare=task_queues)
+producer.publish(
+    {"foo": "bar"},
+    exchange=media_exchange,
+    routing_key='video',
+    declare=task_queues,
+)
 consumer.consume()
 connection.drain_events()

examples/rpc-tut6/rpc_client.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-from kombu import Connection, Producer, Consumer, Queue, uuid
+from kombu import Connection, Consumer, Producer, Queue, uuid
 
 
 class FibonacciRpcClient:

examples/simple_eventlet_receive.py

@@ -37,4 +37,5 @@ def wait_many(timeout=1):
                     message.ack()
                     print(message.payload)
 
+
 eventlet.spawn(wait_many).wait()

examples/simple_task_queue/client.py

@@ -21,8 +21,10 @@ def send_as_task(connection, fun, args=(), kwargs={}, priority='mid'):
                          declare=[task_exchange],
                          routing_key=routing_key)
 
+
 if __name__ == '__main__':
     from kombu import Connection
+
     from .tasks import hello_task
 
     connection = Connection('amqp://guest:guest@localhost:5672//')

examples/simple_task_queue/worker.py

@@ -1,5 +1,5 @@
-from kombu.mixins import ConsumerMixin
 from kombu.log import get_logger
+from kombu.mixins import ConsumerMixin
 from kombu.utils.functional import reprcall
 
 from .queues import task_queues
@@ -28,9 +28,11 @@ def process_task(self, body, message):
             logger.error('task raised exception: %r', exc)
         message.ack()
 
+
 if __name__ == '__main__':
     from kombu import Connection
     from kombu.utils.debug import setup_logging
+
     # setup root logger
     setup_logging(loglevel='INFO', loggers=[''])
 

kombu/__init__.py

@@ -3,7 +3,7 @@
 import os
 import re
 import sys
-from collections import namedtuple  # noqa
+from collections import namedtuple
 
 __version__ = '5.1.0'
 __author__ = 'Ask Solem'
@@ -32,17 +32,15 @@
     # This is never executed, but tricks static analyzers (PyDev, PyCharm,
     # pylint, etc.) into knowing the types of these symbols, and what
     # they contain.
-    from kombu.connection import Connection, BrokerConnection   # noqa
-    from kombu.entity import Exchange, Queue, binding           # noqa
-    from kombu.message import Message                           # noqa
-    from kombu.messaging import Consumer, Producer              # noqa
-    from kombu.pools import connections, producers              # noqa
-    from kombu.utils.url import parse_url                       # noqa
-    from kombu.common import eventloop, uuid                    # noqa
-    from kombu.serialization import (                           # noqa
-        enable_insecure_serializers,
-        disable_insecure_serializers,
-    )
+    from kombu.common import eventloop, uuid  # noqa
+    from kombu.connection import BrokerConnection, Connection  # noqa
+    from kombu.entity import Exchange, Queue, binding  # noqa
+    from kombu.message import Message  # noqa
+    from kombu.messaging import Consumer, Producer  # noqa
+    from kombu.pools import connections, producers  # noqa
+    from kombu.serialization import disable_insecure_serializers  # noqa
+    from kombu.serialization import enable_insecure_serializers
+    from kombu.utils.url import parse_url  # noqa
 
 # Lazy loading.
 # - See werkzeug/__init__.py for the rationale behind this.

kombu/asynchronous/__init__.py

@@ -1,7 +1,7 @@
 """Event loop."""
 
-from .hub import Hub, get_event_loop, set_event_loop
+from kombu.utils.eventio import ERR, READ, WRITE
 
-from kombu.utils.eventio import READ, WRITE, ERR
+from .hub import Hub, get_event_loop, set_event_loop
 
 __all__ = ('READ', 'WRITE', 'ERR', 'Hub', 'get_event_loop', 'set_event_loop')

kombu/asynchronous/aws/connection.py

@@ -1,15 +1,15 @@
 """Amazon AWS Connection."""
 
+from email import message_from_bytes
+from email.mime.message import MIMEMessage
+
 from vine import promise, transform
 
 from kombu.asynchronous.aws.ext import AWSRequest, get_response
-
 from kombu.asynchronous.http import Headers, Request, get_client
 
-from email import message_from_bytes
-from email.mime.message import MIMEMessage
 
-def message_from_headers(hdr):  # noqa
+def message_from_headers(hdr):
     bs = "\r\n".join("{}: {}".format(*h) for h in hdr)
     return message_from_bytes(bs.encode())
 
@@ -129,7 +129,7 @@ def __repr__(self):
 class AsyncConnection:
     """Async AWS Connection."""
 
-    def __init__(self, sqs_connection, http_client=None, **kwargs):  # noqa
+    def __init__(self, sqs_connection, http_client=None, **kwargs):
         self.sqs_connection = sqs_connection
         self._httpclient = http_client or get_client()
 
@@ -184,7 +184,7 @@ def make_request(self, operation, params_, path, verb, callback=None):  # noqa
         params = params_.copy()
         if operation:
             params['Action'] = operation
-        signer = self.sqs_connection._request_signer  # noqa
+        signer = self.sqs_connection._request_signer
 
         # defaults for non-get
         signing_type = 'standard'
@@ -225,7 +225,7 @@ def get_status(self, operation, params, path='/', parent=None, verb='GET', callb
             ),
         )
 
-    def _on_list_ready(self, parent, markers, operation, response):  # noqa
+    def _on_list_ready(self, parent, markers, operation, response):
         service_model = self.sqs_connection.meta.service_model
         if response.status == self.STATUS_CODE_OK:
             _, parsed = get_response(
@@ -243,7 +243,7 @@ def _on_list_ready(self, parent, markers, operation, response):  # noqa
         else:
             raise self._for_status(response, response.read())
 
-    def _on_obj_ready(self, parent, operation, response):  # noqa
+    def _on_obj_ready(self, parent, operation, response):
         service_model = self.sqs_connection.meta.service_model
         if response.status == self.STATUS_CODE_OK:
             _, parsed = get_response(
@@ -253,7 +253,7 @@ def _on_obj_ready(self, parent, operation, response):  # noqa
         else:
             raise self._for_status(response, response.read())
 
-    def _on_status_ready(self, parent, operation, response):  # noqa
+    def _on_status_ready(self, parent, operation, response):
         service_model = self.sqs_connection.meta.service_model
         if response.status == self.STATUS_CODE_OK:
             httpres, _ = get_response(

kombu/asynchronous/aws/sqs/connection.py

@@ -8,7 +8,6 @@
 from .message import AsyncMessage
 from .queue import AsyncQueue
 
-
 __all__ = ('AsyncSQSConnection',)
 
 
@@ -84,7 +83,7 @@ def delete_message(self, queue, receipt_handle, callback=None):
     def delete_message_batch(self, queue, messages, callback=None):
         params = {}
         for i, m in enumerate(messages):
-            prefix = 'DeleteMessageBatchRequestEntry.{}'.format(i + 1)
+            prefix = f'DeleteMessageBatchRequestEntry.{i + 1}'
             params.update({
                 f'{prefix}.Id': m.id,
                 f'{prefix}.ReceiptHandle': m.receipt_handle,
@@ -114,7 +113,7 @@ def send_message(self, queue, message_content,
     def send_message_batch(self, queue, messages, callback=None):
         params = {}
         for i, msg in enumerate(messages):
-            prefix = 'SendMessageBatchRequestEntry.{}'.format(i + 1)
+            prefix = f'SendMessageBatchRequestEntry.{i + 1}'
             params.update({
                 f'{prefix}.Id': msg[0],
                 f'{prefix}.MessageBody': msg[1],
@@ -137,7 +136,7 @@ def change_message_visibility(self, queue, receipt_handle,
     def change_message_visibility_batch(self, queue, messages, callback=None):
         params = {}
         for i, t in enumerate(messages):
-            pre = 'ChangeMessageVisibilityBatchRequestEntry.{}'.format(i + 1)
+            pre = f'ChangeMessageVisibilityBatchRequestEntry.{i + 1}'
             params.update({
                 f'{pre}.Id': t[0].id,
                 f'{pre}.ReceiptHandle': t[0].receipt_handle,

kombu/asynchronous/debug.py

@@ -1,6 +1,6 @@
 """Event-loop debugging tools."""
 
-from kombu.utils.eventio import READ, WRITE, ERR
+from kombu.utils.eventio import ERR, READ, WRITE
 from kombu.utils.functional import reprcall
 
 
@@ -40,13 +40,13 @@ def repr_events(h, events):
 
 def repr_readers(h):
     """Return description of pending readers."""
-    return ['({}){}->{}'.format(fd, _rcb(cb), repr_flag(READ | ERR))
+    return [f'({fd}){_rcb(cb)}->{repr_flag(READ | ERR)}'
             for fd, cb in h.readers.items()]
 
 
 def repr_writers(h):
     """Return description of pending writers."""
-    return ['({}){}->{}'.format(fd, _rcb(cb), repr_flag(WRITE))
+    return [f'({fd}){_rcb(cb)}->{repr_flag(WRITE)}'
             for fd, cb in h.writers.items()]
 
 

kombu/asynchronous/http/__init__.py

@@ -1,6 +1,6 @@
 from kombu.asynchronous import get_event_loop
 
-from .base import Request, Headers, Response
+from .base import Headers, Request, Response
 
 __all__ = ('Client', 'Headers', 'Response', 'Request')
 

kombu/asynchronous/http/base.py

@@ -3,14 +3,13 @@
 import sys
 from http.client import responses
 
-from vine import Thenable, promise, maybe_promise
+from vine import Thenable, maybe_promise, promise
 
 from kombu.exceptions import HttpError
 from kombu.utils.compat import coro
 from kombu.utils.encoding import bytes_to_str
 from kombu.utils.functional import maybe_list, memoize
 
-
 __all__ = ('Headers', 'Response', 'Request')
 
 PYPY = hasattr(sys, 'pypy_version_info')