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

# 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

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