# Path: arq/constants.py

# 
# Path: arq/jobs.py
# 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:
# 
# Path: arq/utils.py
# def timestamp_ms() -> int:
#     return as_int(time() * 1000)
# 
# @overload
# def to_ms(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)


import asyncio
import functools
import logging
import ssl
from dataclasses import dataclass
from datetime import datetime, timedelta
from operator import attrgetter
from typing import Any, Callable, Generator, List, Optional, Tuple, Union
from urllib.parse import urlparse
from uuid import uuid4
from pydantic.validators import make_arbitrary_type_validator
from redis.asyncio import ConnectionPool, Redis
from redis.asyncio.sentinel import Sentinel
from redis.exceptions import RedisError, WatchError
from .constants import default_queue_name, job_key_prefix, result_key_prefix
from .jobs import Deserializer, Job, JobDef, JobResult, Serializer, deserialize_job, serialize_job
from .utils import timestamp_ms, to_ms, to_unix_ms
        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,

# Path: arq/constants.py

# 
# Path: arq/jobs.py
# 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:
# 
# Path: arq/utils.py
# def timestamp_ms() -> int:
#     return as_int(time() * 1000)
# 
# @overload
# def to_ms(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)


# Path: arq/connections.py
import asyncio
import functools
import logging
import ssl
from dataclasses import dataclass
from datetime import datetime, timedelta
from operator import attrgetter
from typing import Any, Callable, Generator, List, Optional, Tuple, Union
from urllib.parse import urlparse
from uuid import uuid4
from pydantic.validators import make_arbitrary_type_validator
from redis.asyncio import ConnectionPool, Redis
from redis.asyncio.sentinel import Sentinel
from redis.exceptions import RedisError, WatchError
from .constants import default_queue_name, job_key_prefix, result_key_prefix
from .jobs import Deserializer, Job, JobDef, JobResult, Serializer, deserialize_job, serialize_job
from .utils import timestamp_ms, to_ms, to_unix_ms

        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,