### Install arq

Source: https://arq-docs.helpmanual.io/

Install the arq package using pip.

```bash
pip install arq
```

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

### Worker.run

Source: https://arq-docs.helpmanual.io/

Starts the worker process to begin consuming jobs.

```APIDOC
## Worker.run

### Description
Starts the worker loop to process jobs from the Redis queue.

### Method
Method call
```

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

### Worker Execution Methods

Source: https://arq-docs.helpmanual.io/

Methods for starting and managing the worker process.

```APIDOC
## run()

### Description
Sync function to run the worker, finally closes worker connections.

## async_run()

### Description
Asynchronously run the worker, does not close connections. Useful when testing.

## _run_check(retry_jobs, max_burst_jobs)

### Description
Run arq.worker.Worker.async_run(), check for failed jobs and raise arq.worker.FailedJobs if any jobs have failed.

### Returns
- **int** - Number of completed jobs
```

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

### Run arq Jobs and Workers

Source: https://arq-docs.helpmanual.io/

Commands to enqueue jobs and start the worker process.

```bash
python demo.py
```

```bash
arq demo.WorkerSettings
```

```bash
arq demo.WorkerSettings --burst
```

```bash
arq demo.WorkerSettings --watch path/to/src
```

```bash
arq --help
```

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

### Implement Simple arq Job Queue

Source: https://arq-docs.helpmanual.io/

A complete example demonstrating job enqueuing and worker configuration using Redis and asyncio.

```python
import asyncio
from httpx import AsyncClient
from arq import create_pool
from arq.connections import RedisSettings

# Here you can configure the Redis connection.
# The default is to connect to localhost:6379, no password.
REDIS_SETTINGS = RedisSettings()

async def download_content(ctx, url):
    session: AsyncClient = ctx['session']
    response = await session.get(url)
    print(f'{url}: {response.text:.80}...')
    return len(response.text)

async def startup(ctx):
    ctx['session'] = AsyncClient()

async def shutdown(ctx):
    await ctx['session'].aclose()

async def main():
    redis = await create_pool(REDIS_SETTINGS)
    for url in ('https://facebook.com', 'https://microsoft.com', 'https://github.com'):
        await redis.enqueue_job('download_content', url)

# WorkerSettings defines the settings to use when creating the work,
# It's used by the arq CLI.
# redis_settings might be omitted here if using the default settings
# For a list of all available settings, see https://arq-docs.helpmanual.io/#arq.worker.Worker
class WorkerSettings:
    functions = [download_content]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = REDIS_SETTINGS

if __name__ == '__main__':
    asyncio.run(main())
```

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

### Health check output format

Source: https://arq-docs.helpmanual.io/

Example output string returned by the health check command.

```text
Mar-01 17:41:22 j_complete=0 j_failed=0 j_retried=0 j_ongoing=0 queued=0
```

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

### Worker output during job cancellation

Source: https://arq-docs.helpmanual.io/

Example console output showing a worker shutting down and cancelling an ongoing job.

```text
➤  arq slow_job.WorkerSettings
12:42:38: Starting worker for 1 functions: the_task
12:42:38: redis_version=4.0.9 mem_usage=904.50K clients_connected=4 db_keys=3
12:42:38:  10.23s → c3dd4acc171541b9ac10b1d791750cde:the_task() delayed=10.23s
^C12:42:40: shutdown on SIGINT ◆ 0 jobs complete ◆ 0 failed ◆ 0 retries ◆ 1 ongoing to cancel
12:42:40:   1.16s ↻ c3dd4acc171541b9ac10b1d791750cde:the_task cancelled, will be run again


➤  arq slow_job.WorkerSettings
12:42:50: Starting worker for 1 functions: the_task
12:42:50: redis_version=4.0.9 mem_usage=904.61K clients_connected=4 db_keys=4
12:42:50:  21.78s → c3dd4acc171541b9ac10b1d791750cde:the_task() try=2 delayed=21.78s
12:42:55:   5.00s ← c3dd4acc171541b9ac10b1d791750cde:the_task ●
^C12:42:57: shutdown on SIGINT ◆ 1 jobs complete ◆ 0 failed ◆ 0 retries ◆ 0 ongoing to cancel
```

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

### Run Arq Worker

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Function to create and run an Arq worker. This is a common entry point for starting a worker process.

```python
def run_worker(settings_cls: 'WorkerSettingsType', **kwargs: Any) -> Worker:
    worker = create_worker(settings_cls, **kwargs)
    worker.run()
    return worker
```

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

### Get All Job Results

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Fetches and returns results for all jobs currently stored in Redis, sorted by enqueue time.

```python
async def all_job_results(self) -> list[JobResult]:
        """
        Get results for all jobs in redis.
        """
        keys = await self.keys(result_key_prefix + '*')
        results = await asyncio.gather(*[self._get_job_result(k) for k in keys])
        return sorted(results, key=attrgetter('enqueue_time'))
```

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

### GET /all_job_results

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Retrieves results for all jobs currently stored in Redis.

```APIDOC
## GET /all_job_results

### Description
Get results for all jobs in redis.

### Method
GET

### Response
#### Success Response (200)
- **results** (list[JobResult]) - A list of job results sorted by enqueue time.
```

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

### Get Job Result Information

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Retrieves detailed information about a specific job using its key. Raises KeyError if the job is not found.

```python
async def _get_job_result(self, key: bytes) -> JobResult:
        job_id = key[len(result_key_prefix) :].decode()
        job = Job(job_id, self, _deserializer=self.job_deserializer)
        r = await job.result_info()
        if r is None:
            raise KeyError(f'job "{key.decode()}" not found')
        r.job_id = job_id
        return r
```

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

### Get Queued Jobs

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Retrieves information about all jobs currently in the queue, primarily for testing purposes. Defaults to the default queue if none is specified.

```python
async def queued_jobs(self, *, queue_name: Optional[str] = None) -> list[JobDef]:
        """
        Get information about queued, mostly useful when testing.
        """
        if queue_name is None:
            queue_name = self.default_queue_name
        jobs = await self.zrange(queue_name, withscores=True, start=0, end=-1)
        return await asyncio.gather(*[self._get_job_def(job_id, int(score)) for job_id, score in jobs])
```

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

### Calculate Next Cron Execution Time

Source: https://arq-docs.helpmanual.io/_modules/arq/cron

Finds the next datetime that matches the specified cron parameters, starting from one second after the previous datetime. Handles various scheduling options like month, day, weekday, hour, minute, second, and microsecond.

```python
import dataclasses
import inspect
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional, Union

from .typing import WEEKDAYS, OptionType, SecondsTimedelta, WeekdayOptionType, WorkerCoroutine
from .utils import import_string, to_seconds


@dataclass
class Options:
    month: OptionType
    day: OptionType
    weekday: WeekdayOptionType
    hour: OptionType
    minute: OptionType
    second: OptionType
    microsecond: int


def next_cron(
    previous_dt: datetime,
    *,
    month: OptionType = None,
    day: OptionType = None,
    weekday: WeekdayOptionType = None,
    hour: OptionType = None,
    minute: OptionType = None,
    second: OptionType = 0,
    microsecond: int = 123_456,
) -> datetime:
    """
    Find the next datetime matching the given parameters.
    """
    dt = previous_dt + timedelta(seconds=1)
    if isinstance(weekday, str):
        weekday = WEEKDAYS.index(weekday.lower())
    options = Options(
        month=month, day=day, weekday=weekday, hour=hour, minute=minute, second=second, microsecond=microsecond
    )

    while True:
        next_dt = _get_next_dt(dt, options)
        # print(dt, next_dt)
        if next_dt is None:
            return dt
        dt = next_dt


def _get_next_dt(dt_: datetime, options: Options) -> Optional[datetime]:  # noqa: C901
    for field, v in dataclasses.asdict(options).items():
        if v is None:
            continue
        if field == 'weekday':
            next_v = dt_.weekday()
        else:
            next_v = getattr(dt_, field)
        if isinstance(v, int):
            mismatch = next_v != v
        elif isinstance(v, (set, list, tuple)):
            mismatch = next_v not in v
        else:
            raise RuntimeError(v)
        # print(field, v, next_v, mismatch)
        if mismatch:
            micro = max(dt_.microsecond - options.microsecond, 0)
            if field == 'month':
                if dt_.month == 12:
                    return datetime(dt_.year + 1, 1, 1, tzinfo=dt_.tzinfo)
                else:
                    return datetime(dt_.year, dt_.month + 1, 1, tzinfo=dt_.tzinfo)
            elif field in ('day', 'weekday'):
                return (
                    dt_
                    + timedelta(days=1)
                    - timedelta(hours=dt_.hour, minutes=dt_.minute, seconds=dt_.second, microseconds=micro)
                )
            elif field == 'hour':
                return dt_ + timedelta(hours=1) - timedelta(minutes=dt_.minute, seconds=dt_.second, microseconds=micro)
            elif field == 'minute':
                return dt_ + timedelta(minutes=1) - timedelta(seconds=dt_.second, microseconds=micro)
            elif field == 'second':
                return dt_ + timedelta(seconds=1) - timedelta(microseconds=micro)
            else:
                if field != 'microsecond':
                    raise RuntimeError(field)
                return dt_ + timedelta(microseconds=options.microsecond - dt_.microsecond)
    return None
```

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

### Initialize ArqRedis Client

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

ArqRedis is a subclass of redis.asyncio.Redis that adds support for job serialization and queue management.

```python
class ArqRedis(BaseRedis):
    """
    Thin subclass of ``redis.asyncio.Redis`` which adds :func:`arq.connections.enqueue_job`.

    :param redis_settings: an instance of ``arq.connections.RedisSettings``.
    :param job_serializer: a function that serializes Python objects to bytes, defaults to pickle.dumps
    :param job_deserializer: a function that deserializes bytes into Python objects, defaults to pickle.loads
    :param default_queue_name: the default queue name to use, defaults to ``arq.queue``.
    :param expires_extra_ms: the default length of time from when a job is expected to start
     after which the job expires, defaults to 1 day in ms.
    :param kwargs: keyword arguments directly passed to ``redis.asyncio.Redis``.
    """

    def __init__(
        self,
        pool_or_conn: Optional[ConnectionPool] = None,
        job_serializer: Optional[Serializer] = None,
        job_deserializer: Optional[Deserializer] = None,
        default_queue_name: str = default_queue_name,
        expires_extra_ms: int = expires_extra_ms,
        **kwargs: Any,
    ) -> None:
        self.job_serializer = job_serializer
        self.job_deserializer = job_deserializer
        self.default_queue_name = default_queue_name
        if pool_or_conn:
            kwargs['connection_pool'] = pool_or_conn
        self.expires_extra_ms = expires_extra_ms
        super().__init__(**kwargs)
```

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

### GET /queued_jobs

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Retrieves information about jobs currently in the queue, primarily for testing purposes.

```APIDOC
## GET /queued_jobs

### Description
Get information about queued, mostly useful when testing.

### Method
GET

### Parameters
#### Query Parameters
- **queue_name** (string) - Optional - The name of the queue to retrieve jobs from. Defaults to the default queue name.
```

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

### Worker Initialization

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Configuration parameters for initializing the arq Worker instance.

```APIDOC
## Worker Initialization

### Description
Initializes the arq Worker with functions, queue settings, and operational constraints.

### Parameters
#### Request Body
- **functions** (Sequence) - Optional - List of functions or coroutines to register.
- **queue_name** (str) - Optional - Name of the queue to poll.
- **cron_jobs** (Sequence) - Optional - List of cron jobs to execute.
- **redis_settings** (RedisSettings) - Optional - Redis connection settings.
- **burst** (bool) - Optional - Whether to run in burst mode.
- **max_jobs** (int) - Optional - Maximum number of concurrent jobs (default: 10).
- **job_timeout** (SecondsTimedelta) - Optional - Default job timeout in seconds (default: 300).
- **keep_result** (SecondsTimedelta) - Optional - Duration to keep job results (default: 3600).
- **poll_delay** (SecondsTimedelta) - Optional - Delay between polling for new jobs (default: 0.5).
- **max_tries** (int) - Optional - Default maximum retries for a job (default: 5).
- **log_results** (bool) - Optional - Whether to log results for successful jobs (default: true).
```

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

### Define Redis Connection Settings

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Use the RedisSettings dataclass to configure connection parameters, including SSL, sentinel, and retry logic.

```python
@dataclass
class RedisSettings:
    """
    No-Op class used to hold redis connection redis_settings.

    Used by :func:`arq.connections.create_pool` and :class:`arq.worker.Worker`.
    """

    host: Union[str, list[tuple[str, int]]] = 'localhost'
    port: int = 6379
    unix_socket_path: Optional[str] = None
    database: int = 0
    username: Optional[str] = None
    password: Optional[str] = None
    ssl: bool = False
    ssl_keyfile: Optional[str] = None
    ssl_certfile: Optional[str] = None
    ssl_cert_reqs: str = 'required'
    ssl_ca_certs: Optional[str] = None
    ssl_ca_data: Optional[str] = None
    ssl_check_hostname: bool = False
    conn_timeout: int = 1
    conn_retries: int = 5
    conn_retry_delay: int = 1
    max_connections: Optional[int] = None

    sentinel: bool = False
    sentinel_master: str = 'mymaster'

    retry_on_timeout: bool = False
    retry_on_error: Optional[list[type[Exception]]] = None
    retry: Optional[Retry] = None
```

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

### JobDef Dataclass

Source: https://arq-docs.helpmanual.io/_modules/arq/jobs

Represents the definition and metadata of a job. Used for jobs that are queued or deferred but not yet started.

```python
@dataclass
class JobDef:
    function: str
    args: tuple[Any, ...]
    kwargs: dict[str, Any]
    job_try: int
    enqueue_time: datetime
    score: Optional[int]
    job_id: Optional[str]

    def __post_init__(self) -> None:
        if isinstance(self.score, float):
            self.score = int(self.score)
```

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

### Configure Job Functions with func

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Use this wrapper to define custom settings for job functions, such as timeouts, result retention, and retry limits.

```python
def func(
    coroutine: Union[str, Function, 'WorkerCoroutine'],
    *,
    name: Optional[str] = None,
    keep_result: Optional['SecondsTimedelta'] = None,
    timeout: Optional['SecondsTimedelta'] = None,
    keep_result_forever: Optional[bool] = None,
    max_tries: Optional[int] = None,
) -> Function:
    """
    Wrapper for a job function which lets you configure more settings.

    :param coroutine: coroutine function to call, can be a string to import
    :param name: name for function, if None, ``coroutine.__qualname__`` is used
    :param keep_result: duration to keep the result for, if 0 the result is not kept
    :param keep_result_forever: whether to keep results forever, if None use Worker default, wins over ``keep_result``
    :param timeout: maximum time the job should take
    :param max_tries: maximum number of tries allowed for the function, use 1 to prevent retrying
    """
    if isinstance(coroutine, Function):
        return coroutine

    if isinstance(coroutine, str):
        name = name or coroutine
        coroutine_: WorkerCoroutine = import_string(coroutine)
    else:
        coroutine_ = coroutine

    if not inspect.iscoroutinefunction(coroutine_):
        raise RuntimeError(f'{coroutine_} is not a coroutine function')
    timeout = to_seconds(timeout)
    keep_result = to_seconds(keep_result)

    return Function(name or coroutine_.__qualname__, coroutine_, timeout, keep_result, keep_result_forever, max_tries)
```

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

### POST /create_pool

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Initializes a new Redis connection pool for job enqueuing.

```APIDOC
## POST /create_pool

### Description
Create a new redis pool, retrying up to conn_retries times if the connection fails.

### Method
POST

### Parameters
#### Request Body
- **settings_** (RedisSettings) - Optional - Redis connection settings.
- **retry** (int) - Optional - Number of retries.
- **job_serializer** (Serializer) - Optional - Custom job serializer.
- **job_deserializer** (Deserializer) - Optional - Custom job deserializer.
- **default_queue_name** (string) - Optional - Default queue name.
- **expires_extra_ms** (int) - Optional - Extra expiration time in milliseconds.
```

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

### RedisSettings

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Configuration class for Redis connection settings.

```APIDOC
## RedisSettings

### Description
Data class used to hold Redis connection settings for arq pools and workers.

### Parameters
- **host** (Union[str, list[tuple[str, int]]]) - Optional - Redis host or list of sentinel nodes
- **port** (int) - Optional - Redis port
- **database** (int) - Optional - Redis database index
- **username** (str) - Optional - Redis username
- **password** (str) - Optional - Redis password
- **ssl** (bool) - Optional - Enable SSL
- **sentinel** (bool) - Optional - Enable Redis Sentinel mode
```

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

### Implement custom job serialization with MsgPack

Source: https://arq-docs.helpmanual.io/

Configure custom serializers and deserializers in both the connection pool and WorkerSettings to replace the default pickle module.

```python
import asyncio

import msgpack  # installable with "pip install msgpack"

from arq import create_pool
from arq.connections import RedisSettings


async def the_task(ctx):
    return 42


async def main():
    redis = await create_pool(
        RedisSettings(),
        job_serializer=msgpack.packb,
        job_deserializer=lambda b: msgpack.unpackb(b, raw=False),
    )
    await redis.enqueue_job('the_task')


class WorkerSettings:
    functions = [the_task]
    job_serializer = msgpack.packb
    # refer to MsgPack's documentation as to why raw=False is required
    job_deserializer = lambda b: msgpack.unpackb(b, raw=False)


if __name__ == '__main__':
    asyncio.run(main())
```

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

### Get Job Definition

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Retrieves the definition of a queued job, including its score (deferral time). Raises RuntimeError if the job is not found.

```python
async def _get_job_def(self, job_id: bytes, score: int) -> JobDef:
        key = job_key_prefix + job_id.decode()
        v = await self.get(key)
        if v is None:
            raise RuntimeError(f'job "{key}" not found')
        jd = deserialize_job(v, deserializer=self.job_deserializer)
        jd.score = score
        jd.job_id = job_id.decode()
        return jd
```

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

### Worker Initialization Properties

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Details the various properties used during the initialization of an Arq worker.

```APIDOC
## Worker Initialization Properties

### `keep_result_forever`

(bool) - Whether to keep job results forever.

### `poll_delay_s`

(float) - The delay in seconds between polling for new jobs.

### `queue_read_limit`

(int) - The limit for reading from the job queue.

### `max_tries`

(int) - The maximum number of times a job can be retried.

### `health_check_interval`

(float) - The interval in seconds for health checks.

### `health_check_key`

(str) - The key used for health checks.

### `redis_settings`

(Optional[RedisSettings]) - Settings for the Redis connection.

### `ctx`

(dict) - Context dictionary for the worker.

### `timezone`

(tzinfo) - The timezone to use for job scheduling.

### `retry_jobs`

(bool) - Whether to retry jobs on Retry and CancelledError.

### `allow_abort_jobs`

(bool) - Whether to allow aborting jobs.

### `allow_pick_jobs`

(bool) - Whether to allow picking jobs from the queue.

### `max_burst_jobs`

(int) - Maximum number of jobs to burst.

### `job_serializer`

(Callable) - Function to serialize jobs.

### `job_deserializer`

(Callable) - Function to deserialize jobs.

### `expires_extra_ms`

(int) - Extra milliseconds before job results expire.

### `log_results`

(bool) - Whether to log job results.
```

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

### Signal Handling and Shutdown

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Methods for managing OS signals and graceful task completion during shutdown.

```python
    def _add_signal_handler(self, signum: Signals, handler: Callable[[Signals], None]) -> None:
        try:
            self.loop.add_signal_handler(signum, partial(handler, signum))
        except NotImplementedError:  # pragma: no cover
            logger.debug('Windows does not support adding a signal handler to an eventloop')
```

```python
    def _jobs_started(self) -> int:
        return self.jobs_complete + self.jobs_retried + self.jobs_failed + len(self.tasks)
```

```python
    async def _sleep_until_tasks_complete(self) -> None:
        """
        Sleeps until all tasks are done. Used together with asyncio.wait_for()
        """
        while len(self.tasks):
            await asyncio.sleep(0.1)
```

```python
    async def _wait_for_tasks_to_complete(self, signum: Signals) -> None:
        """
        Wait for tasks to complete, until `wait_for_job_completion_on_signal_second` has been reached.
        """
        with contextlib.suppress(asyncio.TimeoutError):
            await asyncio.wait_for(
                self._sleep_until_tasks_complete(),
                self._job_completion_wait,
            )
        logger.info(
            'shutdown on %s, wait complete ◆ %d jobs complete ◆ %d failed ◆ %d retries ◆ %d ongoing to cancel',
            signum.name,
            self.jobs_complete,
            self.jobs_failed,
            self.jobs_retried,
            sum(not t.done() for t in self.tasks.values()),
        )
        for t in self.tasks.values():
            if not t.done():
                t.cancel()
        self.main_task and self.main_task.cancel()
        self.on_stop and self.on_stop(signum)
```

```python
    def handle_sig_wait_for_completion(self, signum: Signals) -> None:
        """
        Alternative signal handler that allow tasks to complete within a given time before shutting down the worker.
        Time can be configured using `wait_for_job_completion_on_signal_second`.
        The worker will stop picking jobs when signal has been received.
        """
        sig = Signals(signum)
        logger.info('Setting allow_pick_jobs to `False`')
        self.allow_pick_jobs = False
        logger.info(
            'shutdown on %s ◆ %d jobs complete ◆ %d failed ◆ %d retries ◆ %d to be completed',
            sig.name,
```

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

### Initialize Arq Worker

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Worker configuration including Redis settings, signal handlers, and task tracking initialization.

```python
        self.keep_result_forever = keep_result_forever
        self.poll_delay_s = to_seconds(poll_delay)
        self.queue_read_limit = queue_read_limit or max(max_jobs * 5, 100)
        self._queue_read_offset = 0
        self.max_tries = max_tries
        self.health_check_interval = to_seconds(health_check_interval)
        if health_check_key is None:
            self.health_check_key = self.queue_name + health_check_key_suffix
        else:
            self.health_check_key = health_check_key
        self._pool = redis_pool
        if self._pool is None:
            self.redis_settings: Optional[RedisSettings] = redis_settings or RedisSettings()
        else:
            self.redis_settings = None
        # self.tasks holds references to run_job coroutines currently running
        self.tasks: dict[str, asyncio.Task[Any]] = {}
        # self.job_tasks holds references the actual jobs running
        self.job_tasks: dict[str, asyncio.Task[Any]] = {}
        self.main_task: Optional[asyncio.Task[None]] = None
        try:
            self.loop = asyncio.get_event_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)
        self.ctx = ctx or {}
        max_timeout = max(f.timeout_s or self.job_timeout_s for f in self.functions.values())
        self.in_progress_timeout_s = (max_timeout or 0) + 10
        self.jobs_complete = 0
        self.jobs_retried = 0
        self.jobs_failed = 0
        self._last_health_check: float = 0
        self._last_health_check_log: Optional[str] = None
        self._handle_signals = handle_signals
        self._job_completion_wait = job_completion_wait
        if self._handle_signals:
            if self._job_completion_wait:
                self._add_signal_handler(signal.SIGINT, self.handle_sig_wait_for_completion)
                self._add_signal_handler(signal.SIGTERM, self.handle_sig_wait_for_completion)
            else:
                self._add_signal_handler(signal.SIGINT, self.handle_sig)
                self._add_signal_handler(signal.SIGTERM, self.handle_sig)
        self.on_stop: Optional[Callable[[Signals], None]] = None
        # whether or not to retry jobs on Retry and CancelledError
        self.retry_jobs = retry_jobs
        self.allow_abort_jobs = allow_abort_jobs
        self.allow_pick_jobs: bool = True
        self.aborting_tasks: set[str] = set()
        self.max_burst_jobs = max_burst_jobs
        self.job_serializer = job_serializer
        self.job_deserializer = job_deserializer
        self.expires_extra_ms = expires_extra_ms
        self.log_results = log_results

        # default to system timezone
        self.timezone = datetime.now().astimezone().tzinfo if timezone is None else timezone
```

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

### Job Class Initialization

Source: https://arq-docs.helpmanual.io/_modules/arq/jobs

Initializes a Job object, which serves as a reference to a job in the Arq system. Requires a job ID and a Redis connection.

```python
class Job:
    """
    Holds data a reference to a job.
    """

    __slots__ = 'job_id', '_redis', '_queue_name', '_deserializer'

    def __init__(
        self,
        job_id: str,
        redis: 'Redis[bytes]',
        _queue_name: str = default_queue_name,
        _deserializer: Optional[Deserializer] = None,
    ):
        self.job_id = job_id
        self._redis = redis
        self._queue_name = _queue_name
        self._deserializer = _deserializer
```

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

### Get Worker Kwargs

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Utility function to extract keyword arguments for the Worker class from a settings class. Ensures only valid worker parameters are included.

```python
def get_kwargs(settings_cls: 'WorkerSettingsType') -> dict[str, NameError]:
    worker_args = set(inspect.signature(Worker).parameters.keys())
    d = settings_cls if isinstance(settings_cls, dict) else settings_cls.__dict__
    return {k: v for k, v in d.items() if k in worker_args}
```

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

### Log Redis Server Information

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Uses a pipeline to fetch server, memory, and client metrics from Redis and logs them using a provided function.

```python
async def log_redis_info(redis: 'Redis[bytes]', log_func: Callable[[str], Any]) -> None:
    async with redis.pipeline(transaction=False) as pipe:
        pipe.info(section='Server')
        pipe.info(section='Memory')
        pipe.info(section='Clients')
        pipe.dbsize()
        info_server, info_memory, info_clients, key_count = await pipe.execute()

    redis_version = info_server.get('redis_version', '?')
    mem_usage = info_memory.get('used_memory_human', '?')
    clients_connected = info_clients.get('connected_clients', '?')

    log_func(
        f'redis_version={redis_version} mem_usage={mem_usage} clients_connected={clients_connected} db_keys={key_count}'
    )
```

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

### Abort a job

Source: https://arq-docs.helpmanual.io/

Shows how to use job.abort() to stop a running or queued job, requiring allow_abort_jobs=True in WorkerSettings.

```python
import asyncio
from arq import create_pool
from arq.connections import RedisSettings


async def do_stuff(ctx):
    print('doing stuff...')
    await asyncio.sleep(10)
    return 'stuff done'


async def main():
    redis = await create_pool(RedisSettings())
    job = await redis.enqueue_job('do_stuff')
    await asyncio.sleep(1)
    await job.abort()


class WorkerSettings:
    functions = [do_stuff]
    allow_abort_jobs = True


if __name__ == '__main__':
    asyncio.run(main())
```

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

### Job.info() Method

Source: https://arq-docs.helpmanual.io/_modules/arq/jobs

Retrieves all available information about a job without waiting for its result. It first checks for result information and then for job definition details.

```python
async def info(self) -> Optional[JobDef]:
    """
    All information on a job, including its result if it's available, does not wait for the result.
    """
    info: Optional[JobDef] = await self.result_info()
    if not info:
        v = await self._redis.get(job_key_prefix + self.job_id)
        if v:
            info = deserialize_job(v, deserializer=self._deserializer)
    if info:
        s = await self._redis.zscore(self._queue_name, self.job_id)
        info.score = None if s is None else int(s)
    return info
```

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

### Retry jobs with deferral

Source: https://arq-docs.helpmanual.io/

Demonstrates raising the arq.worker.Retry exception to defer job execution.

```python
import asyncio
from httpx import AsyncClient
from arq import create_pool, Retry
from arq.connections import RedisSettings

async def download_content(ctx, url):
    session: AsyncClient = ctx['session']
    response = await session.get(url)
    if response.status_code != 200:
        # retry the job with increasing back-off
        # delays will be 5s, 10s, 15s, 20s
        # after max_tries (default 5) the job will permanently fail
        raise Retry(defer=ctx['job_try'] * 5)
    return len(response.text)

async def startup(ctx):
    ctx['session'] = AsyncClient()

async def shutdown(ctx):
    await ctx['session'].aclose()

async def main():
    redis = await create_pool(RedisSettings())
    await redis.enqueue_job('download_content', 'https://httpbin.org/status/503')

class WorkerSettings:
    functions = [download_content]
    on_startup = startup
    on_shutdown = shutdown

if __name__ == '__main__':
    asyncio.run(main())
```

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

### Run Worker Methods

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Methods to execute the worker synchronously, asynchronously, or with failure checking.

```python
[docs]    def run(self) -> None:
        """
        Sync function to run the worker, finally closes worker connections.
        """
        self.main_task = self.loop.create_task(self.main())
        try:
            self.loop.run_until_complete(self.main_task)
        except asyncio.CancelledError:  # pragma: no cover
            # happens on shutdown, fine
            pass
        finally:
            self.loop.run_until_complete(self.close())
```

```python
[docs]    async def async_run(self) -> None:
        """
        Asynchronously run the worker, does not close connections. Useful when testing.
        """
        self.main_task = self.loop.create_task(self.main())
        await self.main_task
```

```python
[docs]    async def run_check(self, retry_jobs: Optional[bool] = None, max_burst_jobs: Optional[int] = None) -> int:
        """
        Run :func:`arq.worker.Worker.async_run`, check for failed jobs and raise :class:`arq.worker.FailedJobs`
        if any jobs have failed.

        :return: number of completed jobs
        """
        if retry_jobs is not None:
            self.retry_jobs = retry_jobs
        if max_burst_jobs is not None:
            self.max_burst_jobs = max_burst_jobs
        await self.async_run()
        if self.jobs_failed:
            failed_job_results = [r for r in await self.pool.all_job_results() if not r.success]
            raise FailedJobs(self.jobs_failed, failed_job_results)
        else:
            return self.jobs_complete
```

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

### Manage job concurrency and execution

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Handles semaphore acquisition, job state verification, and task creation.

```python
    def _release_sem_dec_counter_on_complete(self) -> None:
        self.job_counter = self.job_counter - 1
        self.sem.release()
```

```python
    [docs]    async def start_jobs(self, job_ids: list[bytes]) -> None:
        """
        For each job id, get the job definition, check it's not running and start it in a task
        """
        for job_id_b in job_ids:
            await self.sem.acquire()

            if self.job_counter >= self.max_jobs:
                self.sem.release()
                return None

            self.job_counter = self.job_counter + 1

            job_id = job_id_b.decode()
            in_progress_key = in_progress_key_prefix + job_id
            async with self.pool.pipeline(transaction=True) as pipe:
                await pipe.watch(in_progress_key)
                ongoing_exists = await pipe.exists(in_progress_key)
                score = await pipe.zscore(self.queue_name, job_id)
                if ongoing_exists or not score or score > timestamp_ms():
                    # job already started elsewhere, or already finished and removed from queue
                    # if score > ts_now,
                    # it means probably the job was re-enqueued with a delay in another worker
                    self.job_counter = self.job_counter - 1
                    self.sem.release()
                    logger.debug('job %s already running elsewhere', job_id)
                    continue

                pipe.multi()
                pipe.psetex(in_progress_key, int(self.in_progress_timeout_s * 1000), b'1')
                try:
                    await pipe.execute()
                except (ResponseError, WatchError):
                    # job already started elsewhere since we got 'existing'
                    self.job_counter = self.job_counter - 1
                    self.sem.release()
                    logger.debug('multi-exec error, job %s already started elsewhere', job_id)
                else:
                    t = self.loop.create_task(self.run_job(job_id, int(score)))
                    t.add_done_callback(lambda _: self._release_sem_dec_counter_on_complete())
                    self.tasks[job_id] = t
```

```python
    async def run_job(self, job_id: str, score: int) -> None:  # noqa: C901
        start_ms = timestamp_ms()
        async with self.pool.pipeline(transaction=True) as pipe:
            pipe.get(job_key_prefix + job_id)
            pipe.incr(retry_key_prefix + job_id)
```

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

### Execute Job Lifecycle and Error Handling

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Handles job expiration, deserialization, and failure reporting using Redis pipelines and asynchronous task management.

```python
pipe.expire(retry_key_prefix + job_id, 88400)
            if self.allow_abort_jobs:
                pipe.zrem(abort_jobs_ss, job_id)
                v, job_try, _, abort_job = await pipe.execute()
            else:
                v, job_try, _ = await pipe.execute()
                abort_job = False

        function_name, enqueue_time_ms = '<unknown>', 0
        args: tuple[Any, ...] = ()
        kwargs: dict[Any, Any] = {}

        async def job_failed(exc: BaseException) -> None:
            self.jobs_failed += 1
            result_data_ = serialize_result(
                function=function_name,
                args=args,
                kwargs=kwargs,
                job_try=job_try,
                enqueue_time_ms=enqueue_time_ms,
                success=False,
                result=exc,
                start_ms=start_ms,
                finished_ms=timestamp_ms(),
                ref=f'{job_id}:{function_name}',
                serializer=self.job_serializer,
                queue_name=self.queue_name,
                job_id=job_id,
            )
            await asyncio.shield(self.finish_failed_job(job_id, result_data_))

        if not v:
            logger.warning('job %s expired', job_id)
            return await job_failed(JobExecutionFailed('job expired'))

        try:
            function_name, args, kwargs, enqueue_job_try, enqueue_time_ms = deserialize_job_raw(
                v, deserializer=self.job_deserializer
            )
        except SerializationError as e:
            logger.exception('deserializing job %s failed', job_id)
            return await job_failed(e)

        if abort_job:
            t = (timestamp_ms() - enqueue_time_ms) / 1000
            logger.info('%6.2fs ⊘ %s:%s aborted before start', t, job_id, function_name)
            return await job_failed(asyncio.CancelledError())

        try:
            function: Union[Function, CronJob] = self.functions[function_name]
        except KeyError:
            logger.warning('job %s, function %r not found', job_id, function_name)
            return await job_failed(JobExecutionFailed(f'function {function_name!r} not found'))

        if hasattr(function, 'next_run'):
            # cron_job
            ref = function_name
            keep_in_progress: Optional[float] = keep_cronjob_progress
        else:
            ref = f'{job_id}:{function_name}'
            keep_in_progress = None

        if enqueue_job_try and enqueue_job_try > job_try:
            job_try = enqueue_job_try
            await self.pool.setex(retry_key_prefix + job_id, 88400, str(job_try))

        max_tries = self.max_tries if function.max_tries is None else function.max_tries
        if job_try > max_tries:
            t = (timestamp_ms() - enqueue_time_ms) / 1000
            logger.warning('%6.2fs ! %s max retries %d exceeded', t, ref, max_tries)
            self.jobs_failed += 1
            result_data = serialize_result(
                function_name,
                args,
                kwargs,
                job_try,
                enqueue_time_ms,
                False,
                JobExecutionFailed(f'max {max_tries} retries exceeded'),
                start_ms,
                timestamp_ms(),
                ref,
                self.queue_name,
                job_id=job_id,
                serializer=self.job_serializer,
            )
            return await asyncio.shield(self.finish_failed_job(job_id, result_data))

        result = no_result
        exc_extra = None
        finish = False
        timeout_s = self.job_timeout_s if function.timeout_s is None else function.timeout_s
        incr_score: Optional[int] = None
        job_ctx = {
            'job_id': job_id,
            'job_try': job_try,
            'enqueue_time': ms_to_datetime(enqueue_time_ms),
            'score': score,
        }
        ctx = {**self.ctx, **job_ctx}

        if self.on_job_start:
            await self.on_job_start(ctx)

        start_ms = timestamp_ms()
        success = False
        try:
            s = args_to_string(args, kwargs)
            extra = f' try={job_try}' if job_try > 1 else ''
            if (start_ms - score) > 1200:
                extra += f' delayed={(start_ms - score) / 1000:0.2f}s'
            logger.info('%6.2fs → %s(%s)%s', (start_ms - enqueue_time_ms) / 1000, ref, s, extra)
            self.job_tasks[job_id] = task = self.loop.create_task(function.coroutine(ctx, *args, **kwargs))

            # run repr(result) and extra inside try/except as they can raise exceptions
            try:
                result = await asyncio.wait_for(task, timeout_s)
            except (Exception, asyncio.CancelledError) as e:
                exc_extra = getattr(e, 'extra', None)
                if callable(exc_extra):
                    exc_extra = exc_extra()
                raise
            else:
                result_str = '' if result is None or not self.log_results else truncate(repr(result))
            finally:
```

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

### Schedule cron jobs

Source: https://arq-docs.helpmanual.io/

Configuring periodic tasks using the cron helper in WorkerSettings.

```python
from arq import cron

async def run_regularly(ctx):
    print('run foo job at 9.12am, 12.12pm and 6.12pm')

class WorkerSettings:
    cron_jobs = [
        cron(run_regularly, hour={9, 12, 18}, minute=12)
    ]
```

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

### Worker Run Methods

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Provides methods for running the Arq worker, both synchronously and asynchronously, and for checking job status.

```APIDOC
## Worker Run Methods

### `run()`

Synchronous function to run the worker, finally closes worker connections.

#### Method

`run`

### `async_run()`

Asynchronously run the worker, does not close connections. Useful when testing.

#### Method

`async_run`

### `run_check()`

Run :func:`arq.worker.Worker.async_run`, check for failed jobs and raise :class:`arq.worker.FailedJobs` if any jobs have failed.

#### Parameters

- **retry_jobs** (Optional[bool]) - Whether to retry jobs on Retry and CancelledError.
- **max_burst_jobs** (Optional[int]) - Maximum number of jobs to burst.

#### Returns

- **int** - Number of completed jobs.

#### Raises

- **FailedJobs** - If any jobs have failed.
```

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

### Poll for and execute jobs

Source: https://arq-docs.helpmanual.io/_modules/arq/worker

Main loop iteration for polling the queue and managing job lifecycle.

```python
    async for _ in poll(self.poll_delay_s):
            await self._poll_iteration()

            if self.burst:
                if 0 <= self.max_burst_jobs <= self._jobs_started():
                    await asyncio.gather(*self.tasks.values())
                    return None
                queued_jobs = await self.pool.zcard(self.queue_name)
                if queued_jobs == 0:
                    await asyncio.gather(*self.tasks.values())
                    return None
```

```python
    async def _poll_iteration(self) -> None:
        """
        Get ids of pending jobs from the main queue sorted-set data structure and start those jobs, remove
        any finished tasks from self.tasks.
        """
        count = self.queue_read_limit
        if self.burst and self.max_burst_jobs >= 0:
            burst_jobs_remaining = self.max_burst_jobs - self._jobs_started()
            if burst_jobs_remaining < 1:
                return
            count = min(burst_jobs_remaining, count)
        if self.allow_pick_jobs:
            if self.job_counter < self.max_jobs:
                now = timestamp_ms()
                job_ids = await self.pool.zrangebyscore(
                    self.queue_name, min=float('-inf'), start=self._queue_read_offset, num=count, max=now
                )

                await self.start_jobs(job_ids)

        if self.allow_abort_jobs:
            await self._cancel_aborted_jobs()

        for job_id, t in list(self.tasks.items()):
            if t.done():
                del self.tasks[job_id]
                # required to make sure errors in run_job get propagated
                t.result()

        await self.heart_beat()
```

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

### Create Arq Redis Pool with Sentinel

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Creates a new Arq Redis pool using Sentinel for high availability. Retries connection up to a specified number of times.

```python
def pool_factory(*args: Any, **kwargs: Any) -> ArqRedis:
            client = Sentinel(
                *args,
                sentinels=settings.host,
                ssl=settings.ssl,
                **kwargs,
            )
            redis = client.master_for(settings.sentinel_master, redis_class=ArqRedis)
            return cast(ArqRedis, redis)
```

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

### Create Arq Redis Pool

Source: https://arq-docs.helpmanual.io/_modules/arq/connections

Creates a new Arq Redis pool with specified settings, serializers, and queue configurations. Supports retries on connection failure.

```python
async def create_pool(
    settings_: Optional[RedisSettings] = None,
    *,
    retry: int = 0,
    job_serializer: Optional[Serializer] = None,
    job_deserializer: Optional[Deserializer] = None,
    default_queue_name: str = default_queue_name,
    expires_extra_ms: int = expires_extra_ms,
) -> ArqRedis:
    """
    Create a new redis pool, retrying up to ``conn_retries`` times if the connection fails.

    Returns a :class:`arq.connections.ArqRedis` instance, thus allowing job enqueuing.
    """
    settings: RedisSettings = RedisSettings() if settings_ is None else settings_

    if isinstance(settings.host, str) and settings.sentinel:
        raise RuntimeError("str provided for 'host' but 'sentinel' is true; list of sentinels expected")

    if settings.sentinel:

        def pool_factory(*args: Any, **kwargs: Any) -> ArqRedis:
            client = Sentinel(
                *args,
                sentinels=settings.host,
                ssl=settings.ssl,
                **kwargs,
            )
            redis = client.master_for(settings.sentinel_master, redis_class=ArqRedis)
            return cast(ArqRedis, redis)

    else:
        pool_factory = functools.partial(
            ArqRedis,
            host=settings.host,
            port=settings.port,
            unix_socket_path=settings.unix_socket_path,
```