AHjxddlmZddlmZmZmZddlmZddlmZddl m Z er ddl m Z ddl mZGd d Zy ) ) annotations)datetime timedeltatimezone) TYPE_CHECKING)Redis)now)Queue) BaseWorkerc^eZdZd dZed dZd dZddZddZddZ ddZ ddZ dd Z y )IntermediateQueuecL||_|j||_||_y)N) queue_keyget_intermediate_queue_keykey connection)selfrrs M/root/tools/cai/cai_env/lib/python3.12/site-packages/rq/intermediate_queue.py__init__zIntermediateQueue.__init__s""229=$c |dS)zReturns the intermediate queue key for a given queue key. Args: key (str): The queue key Returns: str: The intermediate queue key z :intermediate)clsrs rrz,IntermediateQueue.get_intermediate_queue_keysM**rc$|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:)rrjob_ids rget_first_seen_keyz$IntermediateQueue.get_first_seen_key!s((<x00rct|jj|j|t j ddS)zSets the first seen timestamp for a job. Args: job_id (str): The job ID timestamp (float): The timestamp TiQ)nxex)boolrsetrr timestamprs rset_first_seenz IntermediateQueue.set_first_seen,s=DOO''(?(?(GIZ_chq'rssrc|jj|j|}|r.tjt |t jSy)zReturns the first seen timestamp for a job. Args: job_id (str): The job ID Returns: Optional[datetime]: The timestamp )tzN)rgetrr fromtimestampfloatrutc)rrr$s rget_first_seenz IntermediateQueue.get_first_seen6sEOO''(?(?(GH ))% *:x||L Lrc^|j|}|syt|z tdkDS)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,r r)rr first_seens rshould_be_cleaned_upz&IntermediateQueue.should_be_cleaned_upDs1((0 uz!Ia$888rc|jj|jddDcgc]}|jc}Scc}w)zlReturns the job IDs in the intermediate queue. Returns: List[str]: The job IDs r)rlrangerdecoders r get_job_idszIntermediateQueue.get_job_idsTs6 /3oo.D.DTXXqRT.UVF VVVsAcR|jj|jd|y)zgRemoves a job from the intermediate queue. Args: job_id (str): The job ID r N)rlremrrs rremovezIntermediateQueue.remove\s TXXq&1rc,|j}|D]}|j|}||jvs#|s|j|7|j |rI|j |s[|j ||d|j|y)Nz$Job was stuck in intermediate queue.) exc_string)r5 fetch_jobstarted_job_registryr8r%r0handle_job_failure)rworkerqueuejob_idsrjobs rcleanupzIntermediateQueue.cleanupds""$ (F//&)CU777KK'&&v.,,V4--c5Ek-lKK'! (rN)rstrrr)rrCreturnrC)rrCrDrC)rrCrDr")rrCrDzdatetime | None)rDz list[str])rrCrDNone)r>r r?r rDrE) __name__ __module__ __qualname__r classmethodrrr%r,r0r5r8rBrrrrrs@%  + + 1t 9 W2(rrN) __future__rrrrtypingrredisrrq.utilsr r?r r>r rrrrrNs,"22 "h(h(r