### Clone Repository and Setup Development Environment

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/contributing.md

Clone the asyncmq repository, navigate into the directory, install Hatch, and create the development environment.

```bash
git clone https://github.com/dymmond/asyncmq
cd asyncmq
pip install hatch
hatch env create
```

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

### Install RabbitMQ Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/rabbitmq.md

Install the asyncmq library with RabbitMQ support using pip. Redis is also required if using the default RabbitMQJobStore.

```bash
pip install "asyncmq[aio-pika]"
```

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

### Initialize Postgres Backend Schema

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/troubleshooting.md

Run this Python code to install or drop the necessary tables for the Postgres backend before starting workers or producers. It uses `anyio.run` to execute the asynchronous function.

```python
import anyio
from asyncmq.core.utils.postgres import install_or_drop_postgres_backend

anyio.run(install_or_drop_postgres_backend)
```

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

### AsyncMQ CLI Command Examples

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/cli.md

Examples of common asyncmq CLI commands for managing queues, jobs, and workers.

```bash
asyncmq info version
asyncmq info backend
```

```bash
asyncmq queue list
asyncmq queue info default
asyncmq queue pause default
asyncmq queue resume default
```

```bash
asyncmq job list --queue default --state waiting
asyncmq job inspect <job_id> --queue default
asyncmq job retry <job_id> --queue default
asyncmq job remove <job_id> --queue default
asyncmq job cancel default <job_id>
```

```bash
asyncmq worker start default --concurrency 4
asyncmq worker list
asyncmq worker register <worker_id> default --concurrency 2
asyncmq worker deregister <worker_id>
```

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

### Install AsyncMQ with Optional Backends

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/installation.md

Install AsyncMQ with optional backend extras for different message queue systems. Choose the appropriate backend for your needs.

```bash
pip install "asyncmq[postgres]"   # asyncpg
```

```bash
pip install "asyncmq[mongo]"      # motor
```

```bash
pip install "asyncmq[aio-pika]"   # RabbitMQ backend
```

```bash
pip install "asyncmq[all]"        # all optional backends
```

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

### Verify AsyncMQ Installation

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/installation.md

Verify the installation by running AsyncMQ commands from the terminal. These commands check the CLI help, version, and backend status.

```bash
asyncmq --help
```

```bash
asyncmq info version
```

```bash
asyncmq info backend
```

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

### Production Worker Bootstrap Example

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/workers.md

Example bootstrap for a production topology, including running multiple queues and the stalled recovery scheduler within an AnyIO task group.

```python
import anyio

from asyncmq.core.stalled import stalled_recovery_scheduler
from asyncmq.queues import Queue
from myapp import tasks  # noqa: F401


async def main() -> None:
    emails = Queue("emails", concurrency=16, rate_limit=40, rate_interval=1.0)
    webhooks = Queue("webhooks", concurrency=8)

    async with anyio.create_task_group() as tg:
        tg.start_soon(emails.run)
        tg.start_soon(webhooks.run)
        tg.start_soon(stalled_recovery_scheduler)


anyio.run(main)
```

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

### Install Base AsyncMQ

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/installation.md

Install the base AsyncMQ package, which includes Redis dependency. Use this for a standard installation.

```bash
pip install asyncmq
```

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

### Example URL for Job Search

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/operations.md

An example URL demonstrating how to search for specific jobs within a queue, filtering by active state, task, and tenant.

```text
/queues/billing/jobs?state=active&task=invoice.charge&q=tenant-88
```

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

### SSE Consumer Example

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/reference.md

A JavaScript example demonstrating how to consume Server-Sent Events (SSE) from the `/events` endpoint and handle specific event types like 'metrics'.

```javascript
const source = new EventSource('/asyncmq/events');
source.addEventListener('metrics', (event) => {
  const payload = JSON.parse(event.data);
  console.log('throughput', payload.throughput);
});
```

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

### Install Postgres Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/postgres-backend.md

Install the Postgres backend for AsyncMQ using pip. This command includes the necessary dependencies for PostgreSQL support.

```bash
pip install "asyncmq[postgres]"
```

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

### Job List Query Examples

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/reference.md

Examples of URLs for querying the job list with various parameters like state, task, and search terms.

```text
/queues/emails/jobs?state=failed&task=send-email&sort=newest
```

```text
/queues/emails/jobs?state=waiting&q=customer-123&job_id=7f3a&page=2&size=50
```

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

### Example AsyncMQ Settings

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/installation.md

Define your application's settings by subclassing Settings and configuring the backend and worker parameters. This example uses Redis.

```python
from asyncmq.backends.redis import RedisBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = RedisBackend("redis://localhost:6379/0")
    worker_concurrency = 4
    scan_interval = 1.0
```

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

### Production Schedule Example

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/schedulers.md

Set up a queue with both a local, code-owned schedule and a durable, operator-visible schedule. This example demonstrates adding a local repeatable task and upserting a durable one using cron syntax.

```python
import anyio

from asyncmq.queues import Queue
from myapp import tasks  # noqa: F401


async def main() -> None:
    queue = Queue("maintenance", concurrency=4, scan_interval=2.0)

    # Local, code-owned housekeeping schedule.
    queue.add_repeatable("maintenance.cleanup_tmp", every=300)

    # Durable, operator-visible schedule.
    await queue.upsert_repeatable(
        "maintenance.rotate_keys",
        cron="0 2 * * 0",
        kwargs={"scope": "prod"},
        retries=5,
    )

    await queue.run()


anyio.run(main)
```

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

### Python Settings Class Example (After)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/release-notes.md

Shows the updated approach for defining custom settings using direct object instantiation, as implemented from version 0.4.1 onwards. This method is preferred for its cleaner syntax.

```python
from asyncmq.conf.global_settings import Settings


class MyCustomSettings(Settings):
    hosts: list[str] = ["example.com"]
```

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

### AsyncMQ Job Enqueue Example (Before 0.3.0)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/release-notes.md

Demonstrates the syntax for enqueuing a job with a delay using the older parameter order before version 0.3.0.

```python
import anyio
from asyncmq.queues import Queue

async def main():
    q = Queue("emails")
    await send_welcome.enqueue(q.backend, "alice@example.com", delay=10)
anyio.run(main)
```

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

### Python Settings Class Example (Before)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/release-notes.md

Illustrates the previous method of defining custom settings using dataclasses before version 0.4.1.

```python
from dataclasses import dataclass, field
from asyncmq.conf.global_settings import Settings


@dataclass
class MyCustomSettings(Settings):
    hosts: list[str] = field(default_factory=lambda: ["example.com"])
```

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

### Start an AsyncMQ Worker

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/quickstart.md

Start a worker process to consume and execute tasks from a specified queue.

```bash
asyncmq worker start default --concurrency 1
```

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

### Install PyJWT Dependency

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/jwt.md

Install the pyjwt library, which is required for JWT authentication.

```bash
pip install pyjwt
```

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

### AsyncMQ Job Enqueue Example (After 0.3.0)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/release-notes.md

Illustrates the updated syntax for enqueuing a job with a delay using the named `backend` parameter, as required from version 0.3.0 onwards.

```python
import anyio
from asyncmq.queues import Queue

async def main():
    q = Queue("emails")
    await send_welcome.enqueue("alice@example.com", backend=q.backend, delay=10)
anyio.run(main)
```

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

### FastAPI Producer Example

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/learn/integration.md

Enqueue tasks from a FastAPI request handler. Ensure your tasks are defined in importable modules.

```python
from fastapi import FastAPI
from asyncmq.queues import Queue
from myapp.tasks import send_email

app = FastAPI()
queue = Queue("emails")


@app.post("/users/{email}")
async def create_user(email: str):
    await send_email.enqueue(email, backend=queue.backend)
    return {"queued": True}
```

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

### Enqueue Task

Source: https://github.com/dymmond/asyncmq/blob/main/README.md

Enqueue a task to be processed by the queue. This example enqueues a 'send_welcome' task with a specific email address.

```python
import anyio
from asyncmq.queues import Queue
from myapp.tasks import send_welcome


async def main() -> None:
    queue = Queue("emails")
    job_id = await send_welcome.enqueue("alice@example.com", backend=queue.backend)
    print("enqueued", job_id)


anyio.run(main)
```

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

### Run a Worker Process

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Start a worker process to consume jobs from the queue. `queue.run()` is for asynchronous execution, while `queue.start()` is a blocking wrapper.

```python
await queue.run()  # async
queue.start()      # blocking wrapper
```

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

### Example Filter URL for Failed Jobs

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/dashboard.md

This URL demonstrates how to filter jobs by state, task name, queue, and sort order. Use this to quickly find and retry specific failed jobs.

```text
/queues/emails/jobs?state=failed&task=send-reminder&q=tenant-42&sort=newest
```

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

### Run Subset of Tests

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/contributing.md

Execute a specific subset of tests, for example, tests located in the 'tests/cli' directory, with quiet output.

```bash
hatch run test:test tests/cli -q
```

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

### Unit Test with InMemoryBackend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/testing.md

Use InMemoryBackend for deterministic unit tests without external dependencies. This example demonstrates enqueuing a job and listing waiting jobs.

```python
import pytest
from asyncmq.backends.memory import InMemoryBackend
from asyncmq.queues import Queue


@pytest.mark.anyio
async def test_enqueue_and_list_waiting():
    backend = InMemoryBackend()
    queue = Queue("test", backend=backend)

    job_id = await queue.add("myapp.tasks.echo", args=["x"])
    waiting = await queue.list_jobs("waiting")

    assert any(job["id"] == job_id for job in waiting)
```

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

### Configure Dashboard Appearance and Mount Prefix

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/dashboard.md

Customize the dashboard's appearance and URL prefix using `DashboardConfig`. This example shows how to define custom titles, descriptions, and colors within your application settings.

```python
from asyncmq.conf.global_settings import Settings
from asyncmq.core.utils.dashboard import DashboardConfig


class AppSettings(Settings):
    secret_key = "replace-in-production"

    @property
    def dashboard_config(self) -> DashboardConfig:
        return DashboardConfig(
            title="My AsyncMQ",
            header_title="Background Jobs",
            description="Queue operations",
            dashboard_url_prefix="/admin",
            sidebar_bg_colour="#CBDC38",
            secret_key=self.secret_key,
        )
```

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

### AsyncMQ Worker CLI Command

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/learn/integration.md

Start an AsyncMQ worker process from the command line. Specify the queue name and desired concurrency.

```bash
asyncmq worker start emails --concurrency 4
```

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

### Run a Queue with Concurrency

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/workers.md

Use `Queue.run()` to start the full queue runtime, including job consumption and scheduling. Set `concurrency` to control the number of jobs a single worker process can handle simultaneously.

```python
from asyncmq.queues import Queue

queue = Queue("emails", concurrency=8)
await queue.run()
```

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

### Example Filter URL for Audit Log

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/dashboard.md

This URL shows how to filter audit log entries by action, queue, status, and limit. Useful for tracking operator actions during incidents.

```text
/audit?action=job.retry&queue=emails&status=success&limit=100
```

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

### Prepare, Build, and Serve Documentation

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/contributing.md

Commands to prepare, build, and serve the project's documentation locally.

```bash
hatch run docs:prepare
```

```bash
hatch run docs:build
```

```bash
hatch run docs:serve
```

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

### Inspecting Flow State

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/flows.md

Provides examples of using queue inspection APIs to monitor the progress and state of jobs within a flow. This includes retrieving jobs waiting for children, checking parent job states, and getting job counts.

```python
waiting_children = await queue.get_waiting_children()
children_page = await queue.get_jobs(["waiting-children"], start=0, end=49)
parent_state = await queue.get_job_state(parent_id)
parent_job = await queue.get_job(parent_id)
```

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

### Initialize SimpleUsernamePasswordBackend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/jwt.md

Set up SimpleUsernamePasswordBackend with a synchronous verify callback function. This backend uses session-backed authentication.

```python
from asyncmq.contrib.dashboard.admin import AsyncMQAdmin
from asyncmq.contrib.dashboard.admin.backends.simple_user import (
    SimpleUsernamePasswordBackend,
)
from asyncmq.contrib.dashboard.admin.protocols import User


def verify(username: str, password: str) -> User | None:
    if username == "admin" and password == "secret":
        return User(id="admin", name="Admin", is_admin=True)
    return None


backend = SimpleUsernamePasswordBackend(verify=verify)
admin = AsyncMQAdmin(enable_login=True, backend=backend)
```

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

### Start AsyncMQ Worker

Source: https://github.com/dymmond/asyncmq/blob/main/README.md

Start an AsyncMQ worker process for a specific queue with a defined concurrency level.

```bash
asyncmq worker start emails --concurrency 1
```

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

### Configure Redis Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/switch-backends.md

Set up the Redis backend by providing the connection URL in the settings.

```python
from asyncmq.backends.redis import RedisBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = RedisBackend("redis://localhost:6379/0")
```

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

### Initialize AsyncMQAdmin

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/dashboard.md

Instantiate the AsyncMQAdmin class to enable the dashboard. Set `enable_login=False` to disable authentication.

```python
from asyncmq.contrib.dashboard.admin import AsyncMQAdmin

admin = AsyncMQAdmin(enable_login=False)
```

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

### Configure Postgres Backend with Settings Class

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/postgres-backend.md

Configure the Postgres backend by defining a settings class that inherits from `asyncmq.conf.global_settings.Settings`. Set the `asyncmq_postgres_backend_url` and instantiate the `PostgresBackend`.

```python
from asyncmq.backends.postgres import PostgresBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    asyncmq_postgres_backend_url = "postgresql://postgres:postgres@localhost:5432/asyncmq"
    backend = PostgresBackend()
```

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

### Display CLI Help

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/cli.md

Use this command to view the main help message for the asyncmq CLI.

```bash
asyncmq --help
```

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

### Queue Initialization with Scan Interval

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/schedulers.md

Demonstrates how to configure the scan interval for a queue.

```APIDOC
## Tuning

Use `scan_interval` to control how often scheduling loops poll:

```python
queue = Queue("emails", scan_interval=1.0)
```
```

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

### Configure PostgreSQL Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/switch-backends.md

Configure the PostgreSQL backend using a connection URL and bootstrap the necessary tables.

```python
from asyncmq.backends.postgres import PostgresBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    asyncmq_postgres_backend_url = "postgresql://postgres:postgres@localhost:5432/asyncmq"
    backend = PostgresBackend()
```

```python
import anyio
from asyncmq.core.utils.postgres import install_or_drop_postgres_backend

anyio.run(install_or_drop_postgres_backend)
```

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

### Get All Jobs by State

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Retrieves all jobs currently in specified states like 'waiting' or 'completed'.

```python
waiting = await queue.get_waiting()
completed = await queue.get_completed()
waiting_children = await queue.get_waiting_children()
```

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

### Get Job Counts

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Retrieves counts of jobs based on their states, either individually or by type.

```python
counts = await queue.get_job_counts("waiting", "delayed", "failed")
total = await queue.get_job_count_by_types("waiting", "delayed")
queued_work = await queue.count()
```

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

### Worker Management Commands

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/reference/cli-reference.md

Manage workers, including starting, listing, registering, and deregistering workers for specific queues.

```bash
asyncmq worker start <queue> --concurrency <n>
```

```bash
asyncmq worker list
```

```bash
asyncmq worker register <worker_id> <queue> --concurrency <n>
```

```bash
asyncmq worker deregister <worker_id>
```

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

### Configure Postgres Backend with DSN

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/postgres-backend.md

Instantiate the `PostgresBackend` directly by passing the Data Source Name (DSN) string to the constructor.

```python
backend = PostgresBackend(dsn="postgresql://...")
```

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

### Get Debounce Job ID (Alias)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/deduplication.md

An alias for `get_deduplication_job_id`, providing compatibility with older BullMQ naming conventions.

```APIDOC
## Get Debounce Job ID (Alias)

### Description
An alias for `get_deduplication_job_id` for compatibility.

### Method
`queue.get_debounce_job_id()`

### Parameters
- `deduplication_key` (string): The logical key to inspect.

### Request Example
```python
owner_job_id = await queue.get_debounce_job_id("search:products")
```

### Response
Returns the job ID of the owner, or null if no owner is found.
```

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

### Get Deduplication Job ID

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/deduplication.md

Retrieves the job ID of the current deduplication owner for a given deduplication key.

```APIDOC
## Get Deduplication Job ID

### Description
Inspects the current deduplication owner for a given logical key.

### Method
`queue.get_deduplication_job_id()`

### Parameters
- `deduplication_key` (string): The logical key to inspect.

### Request Example
```python
owner_job_id = await queue.get_deduplication_job_id("search:products")
```

### Response
Returns the job ID of the owner, or null if no owner is found.
```

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

### Bootstrap Postgres Schema

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/postgres-backend.md

Bootstrap the necessary PostgreSQL tables for the backend using the `install_or_drop_postgres_backend` utility. This function should be run within an `anyio` event loop.

```python
import anyio
from asyncmq.core.utils.postgres import install_or_drop_postgres_backend


anyio.run(install_or_drop_postgres_backend)
```

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

### Configure MongoDB Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/switch-backends.md

Set up the MongoDB backend by specifying the connection URL and database name.

```python
from asyncmq.backends.mongodb import MongoDBBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    asyncmq_mongodb_backend_url = "mongodb://localhost:27017"
    asyncmq_mongodb_database_name = "asyncmq"
    backend = MongoDBBackend()
```

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

### Get Specific Job Information

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Retrieves detailed information about a specific job using its ID, including its state and result.

```python
job = await queue.get_job(job_id)
state = await queue.get_job_state(job_id)
result = await queue.get_job_result(job_id)
```

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

### Info Commands

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/reference/cli-reference.md

Retrieve information about the asyncmq version and backend configuration.

```bash
asyncmq info version
```

```bash
asyncmq info backend
```

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

### Get Queue Information

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/failed-jobs.md

Retrieve detailed information about a specific queue, which can be useful when investigating failed jobs associated with that queue.

```bash
asyncmq queue info default
```

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

### Initialize Queue Chart - JavaScript

Source: https://github.com/dymmond/asyncmq/blob/main/asyncmq/contrib/dashboard/templates/queues/info.html

Sets up a line chart to visualize real-time queue statistics. It configures the chart with multiple datasets for different job states and options for responsiveness and animation.

```javascript
const ctx = document.getElementById('queue-chart').getContext('2d');
const maxPoints = 20;
const data = {
  labels: [],
  datasets: [
    {
      label: 'Waiting',
      data: [],
      borderColor: 'rgba(59,130,246,0.8)',
      fill: false
    },
    {
      label: 'Active',
      data: [],
      borderColor: 'rgba(234,179,8,0.8)',
      fill: false
    },
    {
      label: 'Delayed',
      data: [],
      borderColor: 'rgba(14,165,233,0.8)',
      fill: false
    },
    {
      label: 'Failed',
      data: [],
      borderColor: 'rgba(239,68,68,0.8)',
      fill: false
    },
    {
      label: 'Completed',
      data: [],
      borderColor: 'rgba(34,197,94,0.8)',
      fill: false
    },
  ]
};

const queueChart = new Chart(ctx, {
  type: 'line',
  data,
  options: {
    responsive: true,
    maintainAspectRatio: false,
    scales: {
      x: {},
      y: {
        beginAtZero: true
      }
    },
    animation: {
      duration: 0
    },
  },
});
```

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

### Get Paginated List of Jobs

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Fetches a paginated list of jobs from the queue, allowing filtering by state and specifying sorting order.

```python
page = await queue.get_jobs(["waiting", "delayed"], start=0, end=49, asc=True)
```

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

### Configure In-Memory Backend

Source: https://github.com/dymmond/asyncmq/blob/main/README.md

Set up the in-memory backend for AsyncMQ by defining a settings class that inherits from Settings and assigns an InMemoryBackend instance to the backend attribute.

```python
from asyncmq.backends.memory import InMemoryBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = InMemoryBackend()
```

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

### Record Job Heartbeat

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/workers.md

Record a heartbeat for a job when execution starts. Long-running jobs that may exceed `stalled_threshold` should refresh that heartbeat explicitly.

```python
from asyncmq.core.stalled import record_heartbeat
from asyncmq.tasks import task


@task(queue="video")
async def transcode(job_id: str) -> None:
    for chunk in range(10):
        ...
        await record_heartbeat("video", job_id)
```

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

### Define Custom App Settings

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/settings.md

Create a custom settings class by inheriting from `asyncmq.conf.global_settings.Settings` and configuring backend and other parameters.

```python
# myapp/settings.py
from asyncmq.backends.redis import RedisBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = RedisBackend("redis://localhost:6379/0")
    worker_concurrency = 4
    scan_interval = 1.0
    logging_level = "INFO"
```

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

### Basic Job Flow Creation

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/flows.md

Demonstrates how to create a simple sequential job flow using `FlowProducer`. Jobs are defined with their task IDs and dependencies, then added to the flow.

```python
from asyncmq.flow import FlowProducer
from asyncmq.jobs import Job

producer = FlowProducer(backend)

extract = Job(task_id="etl.extract", args=[], kwargs={})
transform = Job(task_id="etl.transform", args=[], kwargs={}, depends_on=[extract.id])
load = Job(task_id="etl.load", args=[], kwargs={}, depends_on=[transform.id])

ids = await producer.add_flow("etl", [extract, transform, load])
```

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

### Enable Sandbox Execution

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/sandbox.md

Configure sandbox settings by subclassing `Settings` and setting `sandbox_enabled` to `True`. Adjust `sandbox_default_timeout` for execution limits and `sandbox_ctx` to 'spawn' or 'fork' based on your environment.

```python
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    sandbox_enabled = True
    sandbox_default_timeout = 30.0
    sandbox_ctx = "spawn"  # or "fork" depending platform/runtime needs
```

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

### AsyncMQ Worker CLI Command

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/runners.md

This command is used to start an AsyncMQ worker via the command line interface. Specify the queue name and optionally the concurrency level.

```bash
asyncmq worker start <queue> --concurrency <n>
```

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

### Implement Custom Logging Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/logging.md

Create a custom logging configuration by inheriting from `LoggingConfig` and implementing `configure` and `get_logger` methods. Override `settings.logging_config` to use your custom implementation.

```python
from asyncmq.logging import LoggingConfig


class MyLoggingConfig(LoggingConfig):
    def configure(self) -> None:
        ...

    def get_logger(self):
        ...
```

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

### RabbitMQBackend Constructor Parameters

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/rabbitmq.md

Understand the parameters for the RabbitMQBackend constructor. Provide a custom job_store for alternative metadata persistence or a redis_url if not using the default job store.

```python
RabbitMQBackend(
    rabbit_url: str,
    job_store: BaseJobStore | None = None,
    redis_url: str | None = None,
    prefetch_count: int = 1,
)
```

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

### Get Debounce Job ID (Compatibility)

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/deduplication.md

Access the deduplication owner using the older BullMQ `debounce` getter name for compatibility. This function serves the same purpose as `get_deduplication_job_id`.

```python
owner_job_id = await queue.get_debounce_job_id("search:products")
```

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

### Initialize Queue

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Instantiate a Queue object with a given name. If the backend is not specified, it defaults to the global setting.

```python
from asyncmq.queues import Queue

queue = Queue("default")
```

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

### Get Deduplication Owner Job ID

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/deduplication.md

Inspect the current deduplication owner for a given key. This API helps in understanding which job is currently holding the deduplication lock.

```python
owner_job_id = await queue.get_deduplication_job_id("search:products")
```

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

### Initialize Line Chart for Metrics

Source: https://github.com/dymmond/asyncmq/blob/main/asyncmq/contrib/dashboard/templates/metrics/metrics.html

Sets up a line chart to visualize historical metrics like throughput, retries, and failures. It configures chart options for responsiveness and animation.

```javascript
const initialHistory = {{ metrics_history | tojson }};
const maxPoints = 60;
const metricsCtx = document.getElementById('metrics-chart').getContext('2d');
const metricsData = {
  labels: [],
  datasets: [
    {
      label: 'Throughput',
      data: [],
      borderColor: 'rgba(59,130,246,0.8)',
      fill: false,
    },
    {
      label: 'Retries',
      data: [],
      borderColor: 'rgba(234,179,8,0.8)',
      fill: false,
    },
    {
      label: 'Failures',
      data: [],
      borderColor: 'rgba(239,68,68,0.8)',
      fill: false,
    },
  ],
};
const metricsChart = new Chart(metricsCtx, {
  type: 'line',
  data: metricsData,
  options: {
    responsive: true,
    maintainAspectRatio: false,
    scales: {
      x: { display: true },
      y: { beginAtZero: true }
    },
    animation: { duration: 0 },
  },
});
```

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

### Mount AsyncMQ Admin with URL Prefix

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/jwt.md

Mount the AsyncMQ Admin ASGI app under a specific URL prefix using `with_url_prefix=True` to ensure correct link generation for login, logout, and sidebar navigation.

```python
from fastapi import FastAPI
from asyncmq.contrib.dashboard.admin import AsyncMQAdmin

app = FastAPI()
admin = AsyncMQAdmin(enable_login=True, backend=backend)
app.mount("/ops", admin.get_asgi_app(with_url_prefix=True))
```

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

### Task with Custom Retry Backoff

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/jobs.md

Example of defining a task with custom retry logic using a bounded backoff function. Specifies queue, max retries, and the backoff strategy.

```python
from asyncmq.tasks import task


def bounded_backoff(attempt: int) -> float:
    return min(60.0, 2.0**attempt)


@task(queue="billing", retries=5, backoff=bounded_backoff)
async def charge_invoice(invoice_id: str) -> None:
    ...
```

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

### Configure RabbitMQ Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/how-to/switch-backends.md

Configure the RabbitMQ backend with its AMQP URL and a Redis URL for the metadata store.

```python
from asyncmq.backends.rabbitmq import RabbitMQBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = RabbitMQBackend(
        rabbit_url="amqp://guest:guest@localhost/",
        redis_url="redis://localhost:6379/0",  # metadata store
    )
```

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

### List Queues and Jobs with AsyncMQ CLI

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/troubleshooting.md

Use these commands to inspect queue status and waiting jobs. Ensure you replace `<queue>` with your actual queue name.

```bash
asyncmq info backend
asyncmq queue list
asyncmq job list --queue <queue> --state waiting
```

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

### Configure Queue Rate Limiting

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/workers.md

Configure a queue with `rate_limit` and `rate_interval` to apply a token-bucket limiter within the worker process. This limits the rate at which jobs start, independent of the overall concurrency.

```python
queue = Queue("outbound-api", concurrency=20, rate_limit=10, rate_interval=1.0)
```

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

### Configure RabbitMQ Backend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/backends/rabbitmq.md

Configure the RabbitMQ backend by defining a settings class that inherits from asyncmq.conf.global_settings.Settings. Specify RabbitMQ and Redis URLs, and optionally prefetch count.

```python
from asyncmq.backends.rabbitmq import RabbitMQBackend
from asyncmq.conf.global_settings import Settings


class AppSettings(Settings):
    backend = RabbitMQBackend(
        rabbit_url="amqp://guest:guest@localhost/",
        redis_url="redis://localhost:6379/0",
        prefetch_count=10,
    )
```

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

### Initialize JWTAuthBackend

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/dashboard/jwt.md

Configure JWTAuthBackend with a secret, algorithms, and optional claims for user identification. Ensure the secret is strong and kept secure.

```python
from asyncmq.contrib.dashboard.admin import AsyncMQAdmin
from asyncmq.contrib.dashboard.admin.backends.jwt import JWTAuthBackend

backend = JWTAuthBackend(
    secret="change-me-with-strong-secret",
    algorithms=["HS256"],
    audience=None,
    issuer=None,
    user_claim="sub",
    user_name_claim="name",
    leeway=0,
)

admin = AsyncMQAdmin(enable_login=True, backend=backend)
```

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

### Queue Constructor

Source: https://github.com/dymmond/asyncmq/blob/main/docs/en/docs/features/queues.md

Initializes a new Queue instance. You can specify the queue name, backend, concurrency, rate limiting, and scan intervals.

```APIDOC
## Constructor

```python
Queue(
    name: str,
    backend: BaseBackend | None = None,
    concurrency: int = 3,
    rate_limit: int | None = None,
    rate_interval: float = 1.0,
    scan_interval: float | None = None,
)
```

- If `backend` is omitted, `asyncmq.monkay.settings.backend` is used.
- `scan_interval` overrides global settings for delayed/repeatable scanner loops.
```

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

### Fetch and Refresh Metrics History

Source: https://github.com/dymmond/asyncmq/blob/main/asyncmq/contrib/dashboard/templates/metrics/metrics.html

Asynchronously fetches the latest metrics history from the server. It makes a GET request to the '/metrics/history' endpoint with a limit, parses the JSON response, and then calls renderHistory to update the dashboard.

```javascript
async function refreshHistory() {
  try {
    const resp = await fetch('{{ url_prefix }}/metrics/history?limit=60', {
      headers: {
        'Accept': 'application/json'
      },
    });
    if (!resp.ok) {
      return;
    }
    const payload = await resp.json();
    co
```