# 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
        'st': start_ms,
        'ft': finished_ms,
        'q': queue_name,
    }
    if serializer is None:
        serializer = pickle.dumps
    try:
        return serializer(data)
    except Exception:
        logger.warning('error serializing result of %s', ref, exc_info=True)

    # use string in case serialization fails again
    data.update(r='unable to serialize result', s=False)
    try:
        return serializer(data)
    except Exception:
        logger.critical('error serializing result of %s even after replacing result', ref, exc_info=True)
    return None


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

# 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

        'st': start_ms,
        'ft': finished_ms,
        'q': queue_name,
    }
    if serializer is None:
        serializer = pickle.dumps
    try:
        return serializer(data)
    except Exception:
        logger.warning('error serializing result of %s', ref, exc_info=True)

    # use string in case serialization fails again
    data.update(r='unable to serialize result', s=False)
    try:
        return serializer(data)
    except Exception:
        logger.critical('error serializing result of %s even after replacing result', ref, exc_info=True)
    return None


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

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