# 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
            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


# 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

            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]: