### Install aiotaskq (Repeat)

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Install the aiotaskq library using pip. This is a repeated installation instruction.

```bash
pip install aiotaskq
```

--------------------------------

### Install aiotaskq

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Install the aiotaskq library using pip. Ensure your pip is up-to-date.

```bash
python -m pip install --upgrade pip
pip install aiotaskq
```

--------------------------------

### Define and Run a Simple Task

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Define a task using the aiotaskq decorator and execute it both synchronously and asynchronously. This example demonstrates basic task definition and execution flow, including starting a Redis instance and the aiotaskq worker.

```python
import asyncio

import aiotaskq


@aiotaskq.task()
def some_task(b: int) -> int:
    # Some task with high cpu usage
    def _naive_fib(n: int) -> int:
        if n <= 2:
            return 1
        return _naive_fib(n - 1) + _naive_fib(n - 2)
    return _naive_fib(b)


async def main():
    async_result = await some_task.apply_async(42)
    sync_result = some_task(42)
    assert async_result == sync_result
    print(f"sync_result == async_result == 165580141. Awesome!")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
```

--------------------------------

### Start Redis for aiotaskq

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Start a Redis instance using Docker. This is required for aiotaskq to function as a message broker.

```bash
docker run --publish 127.0.0.1:6379:6379 redis
```

--------------------------------

### Parallel Workflow Composition with aiotaskq

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

This example shows sequential execution, fan-out to multiple workers using asyncio.gather, and fan-in to a final task. Ensure aiotaskq and asyncio are installed.

```python
import asyncio
import aiotaskq


@aiotaskq.task()
def task_1(data: str) -> str:
    return f"step1({data})"


@aiotaskq.task()
def task_2(data: str, worker_id: int) -> str:
    return f"step2-{worker_id}({data})"


@aiotaskq.task()
def task_3(data: str, worker_id: int) -> str:
    return f"step3-{worker_id}({data})"


@aiotaskq.task()
def task_4(results: list) -> str:
    return f"final({','.join(results)})"


async def main():
    # Sequential step 1
    result_1 = await task_1.apply_async("input")

    # Fan-out: run 5 task_2 workers in parallel
    results_2 = await asyncio.gather(
        *[task_2.apply_async(result_1, worker_id=i) for i in range(5)]
    )

    # Fan-out: run 3 task_3 workers in parallel
    results_3 = await asyncio.gather(
        *[task_3.apply_async(r, worker_id=i) for i, r in enumerate(results_2[:3])]
    )

    # Fan-in: collect and finalize
    final = await task_4.apply_async(list(results_3))
    print(final)
    # Output: final(step3-0(step2-0(step1(input))),step3-1(...),step3-2(...))


asyncio.run(main())
```

--------------------------------

### Celery Workflow Example

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

An example demonstrating how to define a complex workflow using Celery's chord primitive for composing tasks.

```python
from celery import Celery

app = Celery()


@app.task
def task_1(*args, **kwargs):
        pass


@app.task
def task_2(*args, **kwargs):
        pass


@app.task
def task_3(*args, **kwargs):
        pass


@app.task
def task_4(*args, **kwargs):
        pass


if __name__ == "__main__":
    step_1 = task_1.si(some_arg="a")
    step_2 = [task_2.si(some_arg=f"{i}") for i in range(5)]
    step_3 = [task_3.si(some_arg=f"{i}") for i in range(3)]
    step_4 = task_4.si(some_arg="b")
    workflow = chord(
        header=step_1,
        body=chord(
            header=step_2,
            body=chord(
                header=step_3,
                body=step_4,
            ),
        ),
    )
    output = workflow.apply_async().get()
    print(output)
```

--------------------------------

### aiotaskq Advanced Workflow Example

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

An example showcasing how to compose tasks in aiotaskq using native Python and asyncio. This approach offers more flexibility compared to Celery's chord.

```python
import asyncio

from aiotaskq import task


@task()
def task_1(*args, **kwargs):
        pass


@task()
def task_2(*args, **kwargs):
        pass


@task()
def task_3(*args, **kwargs):
        pass


@task()
def task_4(*args, **kwargs):
        pass


# So far the same as celery

# And now the workflow is just native python, and you're free
# to use any `asyncio` library of your choice to help with composing
# your workflow e.g. `trio` to handle more advanced scenarios like
# error propagation, task cancellation etc.
if __name__ == "__main__":
    step_1 = task_1.apply_async()
    step_2 = asyncio.gather(task_2.apply_async(arg=f"{i}" for i in range(5)))
    step_3 = asyncio.gather(task_3.apply_async(arg=f"{i}" for i in range(3)))
    step_4 = task_4.apply_async()
    workflow = [step_1, step_2, step_3, step_4]
    output = await asyncio.gather(workflow)
    print(output)
```

--------------------------------

### Run Docker for Tests

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Start necessary Docker containers for running tests. This command should be executed in a separate terminal.

```bash
./docker.sh
```

--------------------------------

### Start aiotaskq Worker

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Start the aiotaskq worker process. This command assumes your tasks are defined in a module named 'app' within a file 'app.py'.

```bash
python -m aiotaskq worker app.app
```

--------------------------------

### Test Total Spending with aiotaskq

Source: https://github.com/imranariffin/aiotaskq/blob/main/src/tests/apps/sample_app_django/sample_app_django/README.md

Execute the script to test the total spending calculation using aiotaskq. This involves starting the aiotaskq worker and running the task.

```bash
# aiotaskq:
$ LOG_LEVEL=INFO ./src/tests/apps/sample_app_django/scripts/test_show_total_spending_aiotaskq.sh
aiotaskq worker is ready with concurrency=4
...
2025-02-02 20:20:44,752 INFO     api.management.commands.show_total_spending_by_user CMD show_total_spending_by_user --aiotaskq ...
2025-02-02 20:20:44,764 INFO     api.management.commands.show_total_spending_by_user Waiting for 10 tasks to finish
2025-02-02 20:20:44,876 INFO     api.management.commands.show_total_spending_by_user CMD show_total_spending_by_user --aiotaskq took 0.1231 seconds
```

--------------------------------

### Start aiotaskq Worker CLI

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Starts a WorkerManager process that spawns and manages worker subprocesses. The manager subscribes to the Redis tasks channel and round-robins tasks to its worker pool. Requires Redis to be running.

```bash
# Install the package
pip install aiotaskq

# Start Redis (required broker)
docker run --publish 127.0.0.1:6379:6379 redis

# Start workers pointing at your app module (app.py in ./app/ directory)
python -m aiotaskq worker app.app

# Custom concurrency (8 worker processes)
python -m aiotaskq worker app.app --concurrency 8

# Custom Redis URL via environment variable
BROKER_URL=redis://my-redis-host:6379 python -m aiotaskq worker app.app

# Rate-limit each worker to process at most 5 tasks concurrently
python -m aiotaskq worker app.app --worker-rate-limit 5

# Check installed version
python -m aiotaskq --version
# Output: 0.0.17
```

--------------------------------

### Test Total Spending with Celery

Source: https://github.com/imranariffin/aiotaskq/blob/main/src/tests/apps/sample_app_django/sample_app_django/README.md

Execute the script to test the total spending calculation using Celery. This involves starting the Celery worker and running the task.

```bash
# Celery:
$ LOG_LEVEL=INFO ./src/tests/apps/sample_app_django/scripts/test_show_total_spending_celery.sh
Celery worker is ready with concurrency=4
...
2025-02-02 20:24:02,876 INFO     api.management.commands.show_total_spending_by_user CMD show_total_spending_by_user --celery...
2025-02-02 20:24:52,579 INFO     api.management.commands.show_total_spending_by_user Waiting for 10 tasks to finish
2025-02-02 20:24:04,437 INFO     api.management.commands.show_total_spending_by_user CMD show_total_spending_by_user --celery took 1.5609 seconds
```

--------------------------------

### Django Integration with App and autodiscover_tasks

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

App is the aiotaskq application object for use in Django projects, mirroring the Celery app pattern. autodiscover_tasks() iterates all installed Django apps and imports their tasks.py modules so that all @aiotaskq.task()-decorated functions are registered.

```python
# sample_app_django/aiotaskq.py
import os
from aiotaskq import App

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "sample_app_django.settings")

app = App()
app.autodiscover_tasks()  # scans all INSTALLED_APPS for tasks.py modules
```

```python
# api/tasks.py  (auto-discovered by autodiscover_tasks)
import aiotaskq
from .models import Order


@aiotaskq.task()
def compute_total_spending(user_id: int) -> float:
    orders = Order.objects.filter(user_id=user_id)
    return sum(o.amount for o in orders)
```

```python
# api/management/commands/run_report.py
import asyncio
from django.core.management.base import BaseCommand
from api.tasks import compute_total_spending


class Command(BaseCommand):
    def handle(self, *args, **options):
        user_ids = [1, 2, 3, 4, 5]

        async def gather_results():
            return await asyncio.gather(
                *[compute_total_spending.apply_async(uid) for uid in user_ids]
            )

        totals = asyncio.run(gather_results())
        for uid, total in zip(user_ids, totals):
            self.stdout.write(f"User {uid}: ${total:.2f}")
```

```bash
# Start the aiotaskq worker for the Django app
python -m aiotaskq worker sample_app_django.aiotaskq

# Run the management command
python manage.py run_report
```

--------------------------------

### Run Django Migrations and Populate Database

Source: https://github.com/imranariffin/aiotaskq/blob/main/src/tests/apps/sample_app_django/sample_app_django/README.md

Prepare the Django application by applying database migrations and populating the database with initial data using provided scripts.

```bash
# Preparation
./src/tests/apps/sample_app_django/manage.py migrate
./src/tests/apps/sample_app_django/scripts/populate_db.sh
```

--------------------------------

### Activate Development Environment

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Source the activate script to set up the development environment. This is typically used for local development and testing.

```bash
source ./activate.sh
```

--------------------------------

### Run Tests

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Execute the test suite for aiotaskq. This involves activating the development environment and running the test script.

```bash
source ./activate.sh
./test.sh
```

--------------------------------

### Configure aiotaskq via Environment Variables

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

All runtime configuration for aiotaskq is supplied via environment variables. No config files are required. Ensure the BROKER_URL is set to your Redis instance.

```bash
# Redis broker URL (default: redis://127.0.0.1:6379)
export BROKER_URL=redis://my-redis-host:6379

# Or use the legacy REDIS_URL variable (BROKER_URL takes precedence)
export REDIS_URL=redis://my-redis-host:6379

# Serialization format (default: json; only "json" is supported currently)
export AIOTASKQ_SERIALIZATION=json

# Log level for the aiotaskq logger (default: DEBUG)
export AIOTASKQ_LOG_LEVEL=INFO

# Then run your worker
python -m aiotaskq worker myapp.tasks
```

--------------------------------

### Run the Application

Source: https://github.com/imranariffin/aiotaskq/blob/main/README.md

Execute your Python application that utilizes aiotaskq. This should be run after the worker is active.

```bash
python ./app.py
```

--------------------------------

### `Task.apply_async(*args, **kwargs)` — Dispatch a Task to a Worker

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Dispatches a task to a worker process by serializing it and publishing it to Redis. It then awaits and returns the task's result, or raises any exception encountered during execution.

```APIDOC
## `Task.apply_async(*args, **kwargs)` — Dispatch a Task to a Worker

Serializes the task and its arguments, publishes them to the Redis tasks channel, and awaits the result from the Redis results channel. Returns the task's return value directly (or raises if the task raised an exception).

```python
import asyncio
import aiotaskq


@aiotaskq.task()
def join(ls: list, delimiter: str = ",") -> str:
    return delimiter.join([str(x) for x in ls])


@aiotaskq.task()
def power(a: int, b: int = 1) -> int:
    return a ** b


async def main():
    # Keyword arguments are fully supported
    sentence = await join.apply_async(["Hello", "aiotaskq", "world"], delimiter=" ")
    print(sentence)  # Hello aiotaskq world

    # Default arguments work as expected
    squared = await power.apply_async(4, b=2)
    print(squared)  # 16

    # Parallel dispatch with asyncio.gather
    results = await asyncio.gather(
        power.apply_async(2, b=10),
        power.apply_async(3, b=3),
        power.apply_async(5, b=2),
    )
    print(results)  # [1024, 27, 25]

asyncio.run(main())
```
```

--------------------------------

### Dispatch a Task to a Worker with Task.apply_async()

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Use the .apply_async() method on a decorated task to serialize and dispatch it to a worker process via Redis. This method awaits and returns the task's result directly, raising any exceptions that occurred during execution.

```python
import asyncio
import aiotaskq


@aiotaskq.task()
def join(ls: list, delimiter: str = ",") -> str:
    return delimiter.join([str(x) for x in ls])


@aiotaskq.task()
def power(a: int, b: int = 1) -> int:
    return a ** b


async def main():
    # Keyword arguments are fully supported
    sentence = await join.apply_async(["Hello", "aiotaskq", "world"], delimiter=" ")
    print(sentence)  # Hello aiotaskq world

    # Default arguments work as expected
    squared = await power.apply_async(4, b=2)
    print(squared)  # 16

    # Parallel dispatch with asyncio.gather
    results = await asyncio.gather(
        power.apply_async(2, b=10),
        power.apply_async(3, b=3),
        power.apply_async(5, b=2),
    )
    print(results)  # [1024, 27, 25]

asyncio.run(main())
```

--------------------------------

### Add Retry at Call Site with Task.with_retry

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Returns a copy of the task with retry options attached. Useful when retry behavior should vary per call rather than being fixed at definition time. The original task definition remains unmodified.

```python
import asyncio
import aiotaskq


class NetworkError(Exception):
    pass


@aiotaskq.task()
def fetch_data(url: str) -> str:
    # Simulate a network call that may fail
    raise NetworkError(f"Could not reach {url}")


async def main():
    # Original task has no retry configured
    resilient_fetch = fetch_data.with_retry(max_retries=2, on=(NetworkError,))

    try:
        result = await resilient_fetch.apply_async("https://example.com/api")
    except NetworkError as e:
        print(f"Failed after 2 retries: {e}")
    # Output: Failed after 2 retries: Could not reach https://example.com/api

    # Original task is unmodified — no retry
    try:
        result = await fetch_data.apply_async("https://example.com/api")
    except NetworkError as e:
        print(f"Failed immediately (no retry): {e}")

asyncio.run(main())
```

--------------------------------

### Define a Task with Retry Logic

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Configure retry behavior for tasks using the 'retry' option within the @aiotaskq.task() decorator. Specify 'max_retries' for the number of additional attempts and 'on' to define which exceptions trigger a retry.

```python
import asyncio
import aiotaskq


class TransientError(Exception):
    pass


class FatalError(Exception):
    pass


@aiotaskq.task(
    options={
        "retry": {
            "max_retries": 3,       # retry up to 3 additional times
            "on": (TransientError,), # only retry on TransientError
        }
    }
)
def flaky_task(value: int) -> int:
    if value < 0:
        raise TransientError("Temporary failure, will be retried")
    if value == 0:
        raise FatalError("Fatal — will NOT be retried")
    return value * 2


async def main():
    # This will be retried up to 3 times before propagating TransientError
    try:
        result = await flaky_task.apply_async(-1)
    except TransientError as e:
        print(f"Failed after retries: {e}")

    # This raises FatalError immediately without retrying
    try:
        result = await flaky_task.apply_async(0)
    except FatalError as e:
        print(f"Fatal error (no retry): {e}")

    # This succeeds on first attempt
    result = await flaky_task.apply_async(5)
    print(result)  # 10

asyncio.run(main())
```

--------------------------------

### Define a Task with @aiotaskq.task()

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Use the @aiotaskq.task() decorator to convert Python functions (both regular and async) into aiotaskq Tasks. These tasks can be called synchronously or dispatched to a worker process using .apply_async().

```python
import asyncio
import aiotaskq


# Synchronous task
@aiotaskq.task()
def add(x: int, y: int) -> int:
    return x + y


# Async task
@aiotaskq.task()
async def wait(t_s: int) -> int:
    await asyncio.sleep(t_s)
    return t_s


async def main():
    # Synchronous call (runs in the same process)
    sync_result = add(3, 4)
    print(sync_result)  # 7

    # Asynchronous call (dispatched to a worker process via Redis)
    async_result = await add.apply_async(3, 4)
    print(async_result)  # 7

    assert sync_result == async_result

asyncio.run(main())
```

--------------------------------

### `@aiotaskq.task()` — Define a Task

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Converts Python functions (synchronous or async) into aiotaskq Tasks. These tasks can be called directly or dispatched to a worker using `.apply_async()`.

```APIDOC
## `@aiotaskq.task()` — Define a Task

Converts any regular or `async` Python function into an aiotaskq `Task` instance. The decorated function can still be called synchronously like a normal function, or dispatched to a worker via `.apply_async()`.

```python
import asyncio
import aiotaskq


# Synchronous task
@aiotaskq.task()
def add(x: int, y: int) -> int:
    return x + y


# Async task
@aiotaskq.task()
async def wait(t_s: int) -> int:
    await asyncio.sleep(t_s)
    return t_s


async def main():
    # Synchronous call (runs in the same process)
    sync_result = add(3, 4)
    print(sync_result)  # 7

    # Asynchronous call (dispatched to a worker process via Redis)
    async_result = await add.apply_async(3, 4)
    print(async_result)  # 7

    assert sync_result == async_result

asyncio.run(main())
```
```

--------------------------------

### `@aiotaskq.task(options=...)` — Define a Task with Retry Logic

Source: https://context7.com/imranariffin/aiotaskq/llms.txt

Allows defining retry mechanisms for tasks directly in the decorator. You can specify the maximum number of retries and the specific exceptions that should trigger a retry.

```APIDOC
## `@aiotaskq.task(options={"retry": ...})` — Define a Task with Retry Logic

Retry options can be declared at task definition time. `max_retries` controls how many additional attempts are made after the first failure. `on` specifies the exception types that trigger a retry; if the raised exception is not in `on`, no retry occurs.

```python
import asyncio
import aiotaskq


class TransientError(Exception):
    pass


class FatalError(Exception):
    pass


@aiotaskq.task(
    options={
        "retry": {
            "max_retries": 3,       # retry up to 3 additional times
            "on": (TransientError,), # only retry on TransientError
        }
    }
)
def flaky_task(value: int) -> int:
    if value < 0:
        raise TransientError("Temporary failure, will be retried")
    if value == 0:
        raise FatalError("Fatal — will NOT be retried")
    return value * 2


async def main():
    # This will be retried up to 3 times before propagating TransientError
    try:
        result = await flaky_task.apply_async(-1)
    except TransientError as e:
        print(f"Failed after retries: {e}")

    # This raises FatalError immediately without retrying
    try:
        result = await flaky_task.apply_async(0)
    except FatalError as e:
        print(f"Fatal error (no retry): {e}")

    # This succeeds on first attempt
    result = await flaky_task.apply_async(5)
    print(result)  # 10

asyncio.run(main())
```
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.