# 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
        """
        v = await self._redis.get(result_key_prefix + self.job_id)
        if v:
            return deserialize_result(v, deserializer=self._deserializer)
        else:
            return None

    async def status(self) -> JobStatus:
        """
        Status of the job.
        """
        if await self._redis.exists(result_key_prefix + self.job_id):
            return JobStatus.complete
        elif await self._redis.exists(in_progress_key_prefix + self.job_id):
            return JobStatus.in_progress
        else:
            score = await self._redis.zscore(self._queue_name, self.job_id)
            if not score:
                return JobStatus.not_found
            return JobStatus.deferred if score > timestamp_ms() else JobStatus.queued

    async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.5) -> bool:
        """
        Abort the job.

        :param timeout: maximum time to wait for the job result before raising ``TimeoutError``,
            will wait forever on None
        :param poll_delay: how often to poll redis for the job result
        :return: True if the job aborted properly, False otherwise
        """

# 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

        """
        v = await self._redis.get(result_key_prefix + self.job_id)
        if v:
            return deserialize_result(v, deserializer=self._deserializer)
        else:
            return None

    async def status(self) -> JobStatus:
        """
        Status of the job.
        """
        if await self._redis.exists(result_key_prefix + self.job_id):
            return JobStatus.complete
        elif await self._redis.exists(in_progress_key_prefix + self.job_id):
            return JobStatus.in_progress
        else:
            score = await self._redis.zscore(self._queue_name, self.job_id)
            if not score:
                return JobStatus.not_found
            return JobStatus.deferred if score > timestamp_ms() else JobStatus.queued

    async def abort(self, *, timeout: Optional[float] = None, poll_delay: float = 0.5) -> bool:
        """
        Abort the job.

        :param timeout: maximum time to wait for the job result before raising ``TimeoutError``,
            will wait forever on None
        :param poll_delay: how often to poll redis for the job result
        :return: True if the job aborted properly, False otherwise
        """

        await self._redis.zadd(abort_jobs_ss, {self.job_id: timestamp_ms()})