Repo is updated for release.

Signed-off-by: Pavel Kirilin <[email protected]>

Author: s3rius

.flake8

@@ -83,6 +83,12 @@ per-file-ignores =
   test_*.py,tests.py,tests_*.py,*/tests/*:
   ; Use of assert detected
   S101,
+  ; Found outer scope names shadowing.
+  WPS442,
+  ; Found direct magic attribute usage.
+  WPS609,
+  ; Found protected attribute usage
+  WPS437,
 
 exclude =
   ./.git,

.github/workflows/release.yml

@@ -0,0 +1,24 @@
+name: Release python package
+
+on:
+  push:
+    tags:
+      - "*"
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.9"
+      - name: Install deps
+        uses: knowsuchagency/poetry-install@v1
+        env:
+          POETRY_VIRTUALENVS_CREATE: false
+      - name: Release package
+        env:
+          POETRY_PYPI_TOKEN_PYPI: ${{ secrets.PYPI_TOKEN }}
+        run: poetry publish --build

.github/workflows/test.yml

@@ -0,0 +1,83 @@
+name: Testing taskiq-aio-pika
+
+on: push
+
+jobs:
+  black:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.9"
+      - name: Install deps
+        uses: knowsuchagency/poetry-install@v1
+        env:
+          POETRY_VIRTUALENVS_CREATE: false
+      - name: Run black check
+        run: poetry run black --check .
+  flake8:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.9"
+      - name: Install deps
+        uses: knowsuchagency/poetry-install@v1
+        env:
+          POETRY_VIRTUALENVS_CREATE: false
+      - name: Run flake8 check
+        run: poetry run flake8 --count .
+  mypy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "3.9"
+      - name: Install deps
+        uses: knowsuchagency/poetry-install@v1
+        env:
+          POETRY_VIRTUALENVS_CREATE: false
+      - name: Run mypy check
+        run: poetry run mypy .
+  pytest:
+    services:
+      rabbit:
+        image: rabbitmq:3.9.16-alpine
+        env:
+          RABBITMQ_DEFAULT_USER: "guest"
+          RABBITMQ_DEFAULT_PASS: "guest"
+          RABBITMQ_DEFAULT_VHOST: "/"
+        options: >-
+          --health-cmd="rabbitmq-diagnostics check_running -q"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=8
+        ports:
+          - 5672:5672
+    strategy:
+      matrix:
+        py_version: ["3.7", "3.8", "3.9", "3.10"]
+        os: [ubuntu-latest, windows-latest]
+    runs-on: "${{ matrix.os }}"
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: "${{ matrix.py_version }}"
+      - name: Update pip
+        run: python -m pip install -U pip
+      - name: Install poetry
+        run: python -m pip install poetry
+      - name: Install deps
+        run: poetry install
+        env:
+          POETRY_VIRTUALENVS_CREATE: false
+      - name: Run pytest check
+        run: poetry run pytest -vv -n auto --cov="taskiq_aio_pika" .

README.md

@@ -12,4 +12,20 @@ broker = AioPikaBroker()
 async def test() -> None:
     print("nothing")
 
-```
+```
+
+
+## Configuration
+
+AioPikaBroker parameters:
+`url` - url to rabbitmq. If None, "amqp://guest:guest@localhost:5672" is used.
+`result_backend` - custom result backend.
+`task_id_generator` - custom task_id genertaor.
+`exchange_name` - name of exchange that used to send messages.
+`exchange_type` - type of the exchange. Used only if `declare_exchange` is True.
+`queue_name` - queue that used to get incoming messages.
+`routing_key` - that used to bind that queue to the exchange.
+`declare_exchange` - whether you want to declare new exchange if it doesn't exist.
+`qos` - number of messages that worker can prefetch.
+`max_connection_pool_size` - maximum number of connections in pool.
+`max_channel_pool_size` - maximum number of channels for each connection.

poetry.lock

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "taskiq-aio-pika"
 version = "0.0.1"
-description = ""
+description = "RabbitMQ broker for taskiq"
 authors = ["Pavel Kirilin <[email protected]>"]
 readme = "README.md"
 
@@ -20,6 +20,9 @@ pre-commit = "^2.20.0"
 yesqa = "^1.4.0"
 autoflake = "^1.4"
 wemake-python-styleguide = "^0.16.1"
+pytest-xdist = { version = "^2.5.0", extras = ["psutil"] }
+anyio = "^3.6.1"
+pytest-cov = "^3.0.0"
 
 [tool.mypy]
 strict = true

pyproject.toml

@@ -1,4 +1,4 @@
 """Taskiq integration with aio-pika."""
-from taskiq_aio_pika.aio_pika_broker import AioPikaBroker
+from taskiq_aio_pika.broker import AioPikaBroker
 
 __all__ = ["AioPikaBroker"]

taskiq_aio_pika/__init__.py

@@ -19,6 +19,7 @@ class AioPikaBroker(AsyncBroker):
 
     def __init__(  # noqa: WPS211
         self,
+        url: Optional[str] = None,
         result_backend: Optional[AsyncResultBackend[_T]] = None,
         task_id_generator: Optional[Callable[[], str]] = None,
         qos: int = 10,
@@ -29,13 +30,39 @@ def __init__(  # noqa: WPS211
         queue_name: str = "taskiq",
         declare_exchange: bool = True,
         routing_key: str = "#",
-        *connection_args: Any,
+        exchange_type: ExchangeType = ExchangeType.TOPIC,
         **connection_kwargs: Any,
     ) -> None:
+        """
+        Construct a new broker.
+
+        :param url: url to rabbitmq. If None,
+            the default "amqp://guest:guest@localhost:5672" is used.
+        :param result_backend: custom result backend.
+
+        :param task_id_generator: custom task_id genertaor.
+        :param qos: number of messages that worker can prefetch.
+        :param loop: specific even loop.
+        :param max_channel_pool_size: maximum number of channels for each connection.
+        :param max_connection_pool_size: maximum number of connections in pool.
+        :param exchange_name: name of exchange that used to send messages.
+        :param queue_name: queue that used to get incoming messages.
+        :param declare_exchange: whether you want to declare new exchange
+            if it doesn't exist.
+        :param routing_key: that used to bind that queue to the exchange.
+        :param exchange_type: type of the exchange.
+            Used only if `declare_exchange` is True.
+        :param connection_kwargs: additional keyword arguments,
+            for connect_robust method of aio-pika.
+        """
         super().__init__(result_backend, task_id_generator)
 
         async def _get_rmq_connection() -> AbstractRobustConnection:
-            return await connect_robust(*connection_args, **connection_kwargs)
+            return await connect_robust(
+                url,
+                loop=loop,
+                **connection_kwargs,
+            )
 
         self._connection_pool: Pool[AbstractRobustConnection] = Pool(
             _get_rmq_connection,
@@ -54,6 +81,7 @@ async def get_channel() -> AbstractChannel:
         )
 
         self._exchange_name = exchange_name
+        self._exchange_type = exchange_type
         self._qos = qos
         self._declare_exchange = declare_exchange
         self._queue_name = queue_name
@@ -65,7 +93,7 @@ async def startup(self) -> None:
             if self._declare_exchange:
                 exchange = await channel.declare_exchange(
                     self._exchange_name,
-                    type=ExchangeType.TOPIC,
+                    type=self._exchange_type,
                 )
             else:
                 exchange = await channel.get_exchange(self._exchange_name, ensure=False)