101def job(
102    job_type: Optional[JobType] = None,
103    name: Optional[str] = None,
104    keep_result: Optional[SecondsTimedelta] = None,
105    timeout: Optional[SecondsTimedelta] = None,
106    keep_result_forever: Optional[bool] = None,
107    max_tries: Optional[int] = None,
108) -> Callable[[ArqCallable[Any]], _job[Any]]:
109    """
110    Creates an async enqueueable job from the provided function. The function may be
111    synchronous or a coroutine. If synchronous, the job will be run in either a
112    thread or process depending on its `JobType`.
113
114    Synchronous jobs are required to specify their `job_type`. If a job type is
115    specified for a coroutine, a warning will be thrown (but will still execute).
116    """
117    return lambda func: _job(
118        func=func,
119        job_type=job_type,
120        # inherited
121        name=name or func.__qualname__,
122        timeout_s=to_seconds(timeout),
123        keep_result_s=to_seconds(keep_result),
124        keep_result_forever=keep_result_forever,
125        max_tries=max_tries,
126    )

 5class JobType(Enum):
 6    """Indicates the performance characteristic of the job to run."""
 7
 8    IO_BOUND = "io-bound"
 9    """
10    IO-bound tasks typically spend a majority of their time waiting for external
11    services to complete, such as when read / writing to disk or sending / receiving
12    network packets. Since they do not hold the GIL during that downtime, IO-bound
13    jobs run in a ThreadPoolExecutor.
14    """
15
16    CPU_BOUND = "cpu-bound"
17    """
18    CPU-bound tasks typically perform many contiguous operations (like complex
19    calculations) rather than wait for external services. Because they operate 
20    continuously and hold the GIL, CPU-bound jobs run in a ProcessPoolExecutor.
21    """

 24class BaseSettings(type):
 25    """
 26    A metaclass for defining WorkerSettings to pass to an arq process. This enables
 27    using the built-in JobType and executor pool logic, as well as secure remote job
 28    serialization and parsing.
 29    """
 30
 31    def __new__(
 32        cls, clsname: str, bases: Tuple[Any, ...], attrs: Dict[str, Any]
 33    ) -> type:
 34        attrs.update(
 35            {
 36                "on_startup": BaseSettings.on_startup,
 37                "on_shutdown": BaseSettings.on_shutdown,
 38                "job_serializer": BaseSettings.job_serializer,
 39                "job_deserializer": BaseSettings.job_deserializer,
 40            }
 41        )
 42        return super().__new__(cls, clsname, bases, attrs)
 43
 44    @staticmethod
 45    async def on_startup(ctx: Context) -> None:
 46        """
 47        Starts the thread and process pool executors for downstream synchronous job
 48        execution.
 49        """
 50        with styled_text(Fore.BLUE, Style.DIM):
 51            print("[justjobs] Starting executors...")
 52
 53        # we're ok creating pools for all the types since the executors don't
 54        # spin up the threads / processes unless a task is scheduled to run in one
 55        ctx["_executors"] = {
 56            JobType.IO_BOUND: ThreadPoolExecutor(max_workers=MAX_THREAD_WORKERS),
 57            JobType.CPU_BOUND: ProcessPoolExecutor(max_workers=MAX_PROCESS_WORKERS),
 58        }
 59
 60    @staticmethod
 61    async def on_shutdown(ctx: Context) -> None:
 62        """
 63        Gracefully shuts down the available thread and process pool executors.
 64        """
 65        for executor in ctx["_executors"].values():
 66            executor.shutdown(wait=True)
 67
 68        del ctx["_executors"]
 69
 70        with styled_text(Fore.BLUE, Style.DIM):
 71            print("[justjobs] Gracefully shutdown executors ✔")
 72
 73    @staticmethod
 74    def job_serializer(job: Any) -> bytes:
 75        """
 76        Serializes the given job using dill and signs it using blake2b. The serialized
 77        job and its signature are returned for later verification.
 78        """
 79        serialized: bytes = dill.dumps(job)
 80        signer = blake2b(key=SERIALIZATION_SECRET)
 81        signer.update(serialized)
 82        # must be hexdigest to ensure no premature byte delimiters
 83        sig = signer.hexdigest()
 84        return (sig + "|").encode("utf-8") + serialized
 85
 86    @staticmethod
 87    def job_deserializer(packed: bytes) -> Any:
 88        """
 89        Extracts the signature from the serialized job and compares it with the job
 90        function. If the signatures match, the job is deserialized and executed. If
 91        not, an error is raised.
 92        """
 93        sig, serialized = packed.split(b"|", 1)
 94        signer = blake2b(key=SERIALIZATION_SECRET)
 95        signer.update(serialized)
 96
 97        if not compare_digest(sig.decode("utf-8"), signer.hexdigest()):
 98            raise ValueError(
 99                "Invalid job signature! Has someone tampered with your job queue?"
100            )
101
102        return dill.loads(serialized)
103
104    def create_pool(cls, **kwargs: Any) -> Broker:
105        """
106        Creates an ArqRedis instance using this class' RedisSettings and job
107        serializers. This function technically returns an instance of Broker,
108        so the pool creation is delayed until the returned object is either
109        awaited or entered.
110        """
111        if not hasattr(cls, "redis_settings"):
112            raise AttributeError(
113                "You must first define some RedisSettings on this worker class before"
114                " trying to create a pool from them."
115            )
116
117        return Broker(
118            redis_settings=cls.redis_settings,
119            packj=cls.job_serializer,
120            unpackj=cls.job_deserializer,
121            kwargs=kwargs,
122        )