class CronJob:
    name: str
    coroutine: WorkerCoroutine
    month: OptionType
    day: OptionType
    weekday: WeekdayOptionType
    hour: OptionType
    minute: OptionType
    second: OptionType
    microsecond: int
    run_at_startup: bool
    unique: bool
    job_id: Optional[str]
    timeout_s: Optional[float]
    keep_result_s: Optional[float]
    keep_result_forever: Optional[bool]
    max_tries: Optional[int]
    next_run: Optional[datetime] = None

    def calculate_next(self, prev_run: datetime) -> None:
        self.next_run = next_cron(
            prev_run,
            month=self.month,
            day=self.day,
            weekday=self.weekday,
            hour=self.hour,
            minute=self.minute,
            second=self.second,
            microsecond=self.microsecond,
        )

    def __repr__(self) -> str:
        return '<CronJob {}>'.format(' '.join(f'{k}={v}' for k, v in self.__dict__.items())) class JobStatus(str, Enum):
class JobDef:
class JobResult(JobDef):
class Job:
class SerializationError(RuntimeError):
class DeserializationError(SerializationError):
    def __post_init__(self) -> None:
    def __init__(
        self,
        job_id: str,
        redis: Redis,
        _queue_name: str = default_queue_name,
        _deserializer: Optional[Deserializer] = None,
    ):
    async def result(
        self, timeout: Optional[float] = None, *, poll_delay: float = 0.5, pole_delay: float = None
    ) -> Any:
    async def info(self) -> Optional[JobDef]:
    async def result_info(self) -> Optional[JobResult]:
    async def status(self) -> JobStatus:
    async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.5) -> bool:
    def __repr__(self) -> str:
def serialize_job(
    function_name: str,
    args: Tuple[Any, ...],
    kwargs: Dict[str, Any],
    job_try: Optional[int],
    enqueue_time_ms: int,
    *,
    serializer: Optional[Serializer] = None,
) -> bytes:
def serialize_result(
    function: str,
    args: Tuple[Any, ...],
    kwargs: Dict[str, Any],
    job_try: int,
    enqueue_time_ms: int,
    success: bool,
    result: Any,
    start_ms: int,
    finished_ms: int,
    ref: str,
    queue_name: str,
    *,
    serializer: Optional[Serializer] = None,
) -> Optional[bytes]:
def deserialize_job(r: bytes, *, deserializer: Optional[Deserializer] = None) -> JobDef:
def deserialize_job_raw(
    r: bytes, *, deserializer: Optional[Deserializer] = None
) -> Tuple[str, Tuple[Any, ...], Dict[str, Any], int, int]:
def deserialize_result(r: bytes, *, deserializer: Optional[Deserializer] = None) -> JobResult: class ArqRedis(Redis):  # type: ignore[misc]
    """
    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 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,
        **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
        super().__init__(**kwargs)

    async def enqueue_job(
        self,
        function: str,
        *args: Any,
        _job_id: Optional[str] = None,
        _queue_name: Optional[str] = None,
        _defer_until: Optional[datetime] = None,
        _defer_by: Union[None, int, float, timedelta] = None,
        _expires: Union[None, int, float, timedelta] = None,
        _job_try: Optional[int] = None,
        **kwargs: Any,
    ) -> Optional[Job]:
        """
        Enqueue a job.

        :param function: Name of the function to call
        :param args: args to pass to the function
        :param _job_id: ID of the job, can be used to enforce job uniqueness
        :param _queue_name: queue of the job, can be used to create job in different queue
        :param _defer_until: datetime at which to run the job
        :param _defer_by: duration to wait before running the job
        :param _expires: if the job still hasn't started after this duration, do not run it
        :param _job_try: useful when re-enqueueing jobs within a job
        :param kwargs: any keyword arguments to pass to the function
        :return: :class:`arq.jobs.Job` instance or ``None`` if a job with this ID already exists
        """
        if _queue_name is None:
            _queue_name = self.default_queue_name
        job_id = _job_id or uuid4().hex
        job_key = job_key_prefix + job_id
        assert not (_defer_until and _defer_by), "use either 'defer_until' or 'defer_by' or neither, not both"

        defer_by_ms = to_ms(_defer_by)
        expires_ms = to_ms(_expires)

        async with self.pipeline(transaction=True) as pipe:
            await pipe.watch(job_key)
            if any(await asyncio.gather(pipe.exists(job_key), pipe.exists(result_key_prefix + job_id))):
                await pipe.reset()
                return None

            enqueue_time_ms = timestamp_ms()
            if _defer_until is not None:
                score = to_unix_ms(_defer_until)
            elif defer_by_ms:
                score = enqueue_time_ms + defer_by_ms
            else:
                score = enqueue_time_ms

            expires_ms = expires_ms or score - enqueue_time_ms + expires_extra_ms

            job = serialize_job(function, args, kwargs, _job_try, enqueue_time_ms, serializer=self.job_serializer)
            pipe.multi()
            pipe.psetex(job_key, expires_ms, job)
            pipe.zadd(_queue_name, {job_id: score})
            try:
                await pipe.execute()
            except WatchError:
                # job got enqueued since we checked 'job_exists'
                return None
        return Job(job_id, redis=self, _queue_name=_queue_name, _deserializer=self.job_deserializer)

    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

    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'))

    async def _get_job_def(self, job_id: bytes, score: int) -> JobDef:
        v = await self.get(job_key_prefix + job_id.decode())
        jd = deserialize_job(v, deserializer=self.job_deserializer)
        jd.score = score
        return jd

    async def queued_jobs(self, *, queue_name: str = default_queue_name) -> List[JobDef]:
        """
        Get information about queued, mostly useful when testing.
        """
        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]) 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
    database: int = 0
    password: Optional[str] = None
    ssl: Union[bool, None, SSLContext] = None
    conn_timeout: int = 1
    conn_retries: int = 5
    conn_retry_delay: int = 1

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

    @classmethod
    def from_dsn(cls, dsn: str) -> 'RedisSettings':
        conf = urlparse(dsn)
        assert conf.scheme in {'redis', 'rediss'}, 'invalid DSN scheme'
        return RedisSettings(
            host=conf.hostname or 'localhost',
            port=conf.port or 6379,
            ssl=conf.scheme == 'rediss',
            password=conf.password,
            database=int((conf.path or '0').strip('/')),
        )

    def __repr__(self) -> str:
        return 'RedisSettings({})'.format(', '.join(f'{k}={v!r}' for k, v in self.__dict__.items())) async def create_pool(
    settings_: RedisSettings = None,
    *,
    retry: int = 0,
    job_serializer: Optional[Serializer] = None,
    job_deserializer: Optional[Deserializer] = None,
    default_queue_name: str = default_queue_name,
) -> 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_

    assert not (
        type(settings.host) is str and settings.sentinel
    ), "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)
            return client.master_for(settings.sentinel_master, redis_class=ArqRedis)

    else:
        pool_factory = functools.partial(
            ArqRedis,
            host=settings.host,
            port=settings.port,
            socket_connect_timeout=settings.conn_timeout,
            ssl=settings.ssl,
        )

    try:
        pool = pool_factory(db=settings.database, password=settings.password, encoding='utf8')
        pool.job_serializer = job_serializer
        pool.job_deserializer = job_deserializer
        pool.default_queue_name = default_queue_name
        await pool.ping()

    except (ConnectionError, OSError, RedisError, asyncio.TimeoutError) as e:
        if retry < settings.conn_retries:
            logger.warning(
                'redis connection error %s:%s %s %s, %d retries remaining...',
                settings.host,
                settings.port,
                e.__class__.__name__,
                e,
                settings.conn_retries - retry,
            )
            await asyncio.sleep(settings.conn_retry_delay)
        else:
            raise
    else:
        if retry > 0:
            logger.info('redis connection successful')
        return pool

    # recursively attempt to create the pool outside the except block to avoid
    # "During handling of the above exception..." madness
    return await create_pool(
        settings,
        retry=retry + 1,
        job_serializer=job_serializer,
        job_deserializer=job_deserializer,
        default_queue_name=default_queue_name,
    ) async def log_redis_info(redis: Redis, log_func: Callable[[str], Any]) -> None:
    async with redis.pipeline(transaction=True) 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} '
        f'mem_usage={mem_usage} '
        f'clients_connected={clients_connected} '
        f'db_keys={key_count}'
    ) def args_to_string(args: Sequence[Any], kwargs: Dict[str, Any]) -> str:
    arguments = ''
    if args:
        arguments = ', '.join(map(repr, args))
    if kwargs:
        if arguments:
            arguments += ', '
        arguments += ', '.join(f'{k}={v!r}' for k, v in sorted(kwargs.items()))
    return truncate(arguments) def ms_to_datetime(unix_ms: int) -> datetime:
    return datetime.fromtimestamp(unix_ms / 1000, tz=timezone.utc) async def poll(step: float = 0.5) -> AsyncGenerator[float, None]:
    loop = asyncio.get_event_loop()
    start = loop.time()
    while True:
        before = loop.time()
        yield before - start
        after = loop.time()
        wait = max([0, step - after + before])
        await asyncio.sleep(wait) def timestamp_ms() -> int:
    return as_int(time() * 1000) @overload
def to_ms(td: None) -> None:
    pass @overload
def to_seconds(td: None) -> None:
    pass def to_unix_ms(dt: datetime) -> int:
    """
    convert a datetime to epoch with milliseconds as int
    """
    return as_int(dt.timestamp() * 1000) def truncate(s: str, length: int = DEFAULT_CURTAIL) -> str:
    """
    Truncate a string and add an ellipsis (three dots) to the end if it was too long

    :param s: string to possibly truncate
    :param length: length to truncate the string to
    """
    if len(s) > length:
        s = s[: length - 1] + '…'
    return s
import asyncio
import inspect
import logging
import signal
from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import partial
from signal import Signals
from time import time
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Sequence, Set, Tuple, Union, cast
from pydantic.utils import import_string
from redis.exceptions import ResponseError, WatchError
from arq.cron import CronJob
from arq.jobs import Deserializer, JobResult, SerializationError, Serializer, deserialize_job_raw, serialize_result
from .connections import ArqRedis, RedisSettings, create_pool, log_redis_info
from .constants import (
    abort_job_max_age,
    abort_jobs_ss,
    default_queue_name,
    health_check_key_suffix,
    in_progress_key_prefix,
    job_key_prefix,
    keep_cronjob_progress,
    result_key_prefix,
    retry_key_prefix,
)
from .utils import args_to_string, ms_to_datetime, poll, timestamp_ms, to_ms, to_seconds, to_unix_ms, truncate
    from .typing import SecondsTimedelta, StartupShutdown, WorkerCoroutine, WorkerSettingsType  # noqa F401

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


class Retry(RuntimeError):
    """
    Special exception to retry the job (if ``max_retries`` hasn't been reached).

    :param defer: duration to wait before rerunning the job
    """

    def __init__(self, defer: Optional['SecondsTimedelta'] = None):
        self.defer_score: Optional[int] = to_ms(defer)

    def __repr__(self) -> str:
        return f'<Retry defer {(self.defer_score or 0) / 1000:0.2f}s>'

    def __str__(self) -> str:
        return repr(self)


class JobExecutionFailed(RuntimeError):
    def __eq__(self, other: Any) -> bool:
        if isinstance(other, JobExecutionFailed):
            return self.args == other.args
        return False


class FailedJobs(RuntimeError):
    def __init__(self, count: int, job_results: List[JobResult]):
        self.count = count
        self.job_results = job_results

    def __str__(self) -> str:
        if self.count == 1 and self.job_results:
            exc = self.job_results[0].result
            return f'1 job failed {exc!r}'
        else:
            return f'{self.count} jobs failed:\n' + '\n'.join(repr(r.result) for r in self.job_results)

    def __repr__(self) -> str:
        return f'<{str(self)}>'


class RetryJob(RuntimeError):
    pass


class Worker:
    """
    Main class for running jobs.

    :param functions: list of functions to register, can either be raw coroutine functions or the
      result of :func:`arq.worker.func`.
    :param queue_name: queue name to get jobs from
    :param cron_jobs:  list of cron jobs to run, use :func:`arq.cron.cron` to create them
    :param redis_settings: settings for creating a redis connection
    :param redis_pool: existing redis pool, generally None
    :param burst: whether to stop the worker once all jobs have been run
    :param on_startup: coroutine function to run at startup
    :param on_shutdown: coroutine function to run at shutdown
    :param on_job_start: coroutine function to run on job start
    :param on_job_end: coroutine function to run on job end
    :param handle_signals: default true, register signal handlers,
      set to false when running inside other async framework
    :param max_jobs: maximum number of jobs to run at a time
    :param job_timeout: default job timeout (max run time)
    :param keep_result: default duration to keep job results for
    :param keep_result_forever: whether to keep results forever
    :param poll_delay: duration between polling the queue for new jobs
    :param queue_read_limit: the maximum number of jobs to pull from the queue each time it's polled. By default it
                             equals ``max_jobs`` * 5, or 100; whichever is higher.
    :param max_tries: default maximum number of times to retry a job
    :param health_check_interval: how often to set the health check key
    :param health_check_key: redis key under which health check is set
    :param ctx: dictionary to hold extra user defined state
    :param retry_jobs: whether to retry jobs on Retry or CancelledError or not
    :param allow_abort_jobs: whether to abort jobs on a call to :func:`arq.jobs.Job.abort`
    :param max_burst_jobs: the maximum number of jobs to process in burst mode (disabled with negative values)
    :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
    """

    def __init__(
        self,
        functions: Sequence[Union[Function, 'WorkerCoroutine']] = (),
        *,
        queue_name: Optional[str] = default_queue_name,
        cron_jobs: Optional[Sequence[CronJob]] = None,
        redis_settings: RedisSettings = None,
        redis_pool: ArqRedis = None,
        burst: bool = False,
        on_startup: Optional['StartupShutdown'] = None,
        on_shutdown: Optional['StartupShutdown'] = None,
        on_job_start: Optional['StartupShutdown'] = None,
        on_job_end: Optional['StartupShutdown'] = None,
        handle_signals: bool = True,
        max_jobs: int = 10,
        job_timeout: 'SecondsTimedelta' = 300,
        keep_result: 'SecondsTimedelta' = 3600,
        keep_result_forever: bool = False,
        poll_delay: 'SecondsTimedelta' = 0.5,
        queue_read_limit: Optional[int] = None,
        max_tries: int = 5,
        health_check_interval: 'SecondsTimedelta' = 3600,
        health_check_key: Optional[str] = None,
        ctx: Optional[Dict[Any, Any]] = None,
        retry_jobs: bool = True,
        allow_abort_jobs: bool = False,
        max_burst_jobs: int = -1,
        job_serializer: Optional[Serializer] = None,
        job_deserializer: Optional[Deserializer] = None,
    ):
        self.functions: Dict[str, Union[Function, CronJob]] = {f.name: f for f in map(func, functions)}
        if queue_name is None:
            if redis_pool is not None:
                queue_name = redis_pool.default_queue_name
            else:
                raise ValueError('If queue_name is absent, redis_pool must be present.')
        self.queue_name = queue_name

        self.cron_jobs: List[CronJob] = []