# Path: arq/constants.py

# 
# Path: arq/utils.py
# 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)


import asyncio
import logging
import pickle
import warnings
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Dict, Optional, Tuple
from redis.asyncio import Redis
from .constants import abort_jobs_ss, default_queue_name, in_progress_key_prefix, job_key_prefix, result_key_prefix
from .utils import ms_to_datetime, poll, timestamp_ms
            job_try=d['t'],
            enqueue_time=ms_to_datetime(d['et']),
            score=None,
        )
    except Exception as e:
        raise DeserializationError('unable to deserialize job') from e


def deserialize_job_raw(
    r: bytes, *, deserializer: Optional[Deserializer] = None
) -> Tuple[str, Tuple[Any, ...], Dict[str, Any], int, int]:
    if deserializer is None:
        deserializer = pickle.loads
    try:
        d = deserializer(r)
        return d['f'], d['a'], d['k'], d['t'], d['et']
    except Exception as e:
        raise DeserializationError('unable to deserialize job') from e


def deserialize_result(r: bytes, *, deserializer: Optional[Deserializer] = None) -> JobResult:
    if deserializer is None:
        deserializer = pickle.loads
    try:
        d = deserializer(r)
        return JobResult(
            job_try=d['t'],
            function=d['f'],
            args=d['a'],
            kwargs=d['k'],

# Path: arq/constants.py

# 
# Path: arq/utils.py
# 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)


# Path: arq/jobs.py
import asyncio
import logging
import pickle
import warnings
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Dict, Optional, Tuple
from redis.asyncio import Redis
from .constants import abort_jobs_ss, default_queue_name, in_progress_key_prefix, job_key_prefix, result_key_prefix
from .utils import ms_to_datetime, poll, timestamp_ms

            job_try=d['t'],
            enqueue_time=ms_to_datetime(d['et']),
            score=None,
        )
    except Exception as e:
        raise DeserializationError('unable to deserialize job') from e


def deserialize_job_raw(
    r: bytes, *, deserializer: Optional[Deserializer] = None
) -> Tuple[str, Tuple[Any, ...], Dict[str, Any], int, int]:
    if deserializer is None:
        deserializer = pickle.loads
    try:
        d = deserializer(r)
        return d['f'], d['a'], d['k'], d['t'], d['et']
    except Exception as e:
        raise DeserializationError('unable to deserialize job') from e


def deserialize_result(r: bytes, *, deserializer: Optional[Deserializer] = None) -> JobResult:
    if deserializer is None:
        deserializer = pickle.loads
    try:
        d = deserializer(r)
        return JobResult(
            job_try=d['t'],
            function=d['f'],
            args=d['a'],
            kwargs=d['k'],

            enqueue_time=ms_to_datetime(d['et']),