o /i@sjddlmZmZmZddlmZmZddlmZddlm Z er,ddl m Z ddl m Z Gdd d Zd S) )datetime timedeltatimezone) TYPE_CHECKINGOptional)Redis)now)Queue) BaseWorkerc@seZdZdedefddZededefddZdedefd d Zdede fd d Z dede e fd dZ dede fddZdeefddZdeddfddZdddZdS)IntermediateQueue queue_key connectioncCs||_|||_||_dS)N)r get_intermediate_queue_keykeyr)selfr rrU/var/www/html/flask_server/venv/lib/python3.10/site-packages/rq/intermediate_queue.py__init__s  zIntermediateQueue.__init__returncCs |dS)zReturns the intermediate queue key for a given queue key. Args: key (str): The queue key Returns: str: The intermediate queue key z :intermediater)clsr rrrrs z,IntermediateQueue.get_intermediate_queue_keyjob_idcCs|jd|S)zReturns the first seen key for a given job ID. Args: job_id (str): The job ID Returns: str: The first seen key z :first_seen:)rrrrrrget_first_seen_keys z$IntermediateQueue.get_first_seen_keycCs$t|jj||tdddS)zSets the first seen timestamp for a job. Args: job_id (str): The job ID timestamp (float): The timestamp TiQ)nxex)boolrsetrr timestamprrrrset_first_seen*s$z IntermediateQueue.set_first_seencCs.|j||}|rtjt|tjdSdS)zReturns the first seen timestamp for a job. Args: job_id (str): The job ID Returns: Optional[datetime]: The timestamp )tzN)rgetrr fromtimestampfloatrutc)rrrrrrget_first_seen4s z IntermediateQueue.get_first_seencCs&||}|s dSt|tddkS)aReturns whether a job should be cleaned up. A job in intermediate queue should be cleaned up if it has been there for more than 1 minute. Args: job_id (str): The job ID Returns: bool: Whether the job should be cleaned up Fr )minutes)r%rr)rr first_seenrrrshould_be_cleaned_upBs z&IntermediateQueue.should_be_cleaned_upcCsdd|j|jddDS)zlReturns the job IDs in the intermediate queue. Returns: List[str]: The job IDs cSsg|]}|qSr)decode).0rrrr Xsz1IntermediateQueue.get_job_ids..r)rlranger)rrrr get_job_idsRszIntermediateQueue.get_job_idsNcCs|j|jd|dS)zgRemoves a job from the intermediate queue. Args: job_id (str): The job ID r N)rlremrrrrrremoveZszIntermediateQueue.removeworkerr queuer cCsj|}|D],}||}||jvr2|s||q||r q||r2|j||dd||qdS)Nz$Job was stuck in intermediate queue.) exc_string)r. fetch_jobstarted_job_registryr0rr(handle_job_failure)rr1r2job_idsrjobrrrcleanupbs      zIntermediateQueue.cleanup)r1r r2r rN)__name__ __module__ __qualname__strrr classmethodrrrrrrr%r(listr.r0r9rrrrr s   r N)rrrtypingrrredisrrq.utilsrr2r r1r r rrrrs