o i!@sddlZddlmZmZddlmZmZmZddlmZddl m Z er0ddl m Z ddl mZdd lmZdd lmZmZdd lmZmZmZmZGd d d ZGdddeZedZdddedefddZdddeddddfddZdS)N)datetimetimezone) TYPE_CHECKINGAnyOptional)uuid4)Redis)Pipeline) BaseWorker)Job) BaseRegistryStartedJobRegistry)as_textcurrent_timestampnowparse_composite_keyc @seZdZdZdededefddZdedefd d Z e defd d Z e de fd dZ e ddZededededdfddZddZedededdfddZede deddddfddZd(dededfdd Zde ddfd!d"Zdefd#d$Zd%ededdfd&d'ZdS)) Executionz)Class to represent an execution of a job.idjob_id connectioncCs.||_||_||_t}||_||_d|_dSN)rrrr created_atlast_heartbeat_job)selfrrrZ right_nowrD/home/app/Keep/.python/lib/python3.10/site-packages/rq/executions.py__init__s zExecution.__init__otherreturncCst|tsdS|j|jkS)NF) isinstancerr)rrrrr__eq__s  zExecution.__eq__cCs d|jS)Nz rq:execution:) composite_keyrrrrkey$s z Execution.keycCs&|jr|jStj|j|jd|_|jS)N)rr)rr fetchrrr$rrrjob(sz Execution.jobcCs|jd|jS)N:)rrr$rrrr#/szExecution.composite_keycCs||||d}||S)zFetch an execution from Redis.rrr)refresh)clsrrr executionrrrr&3szExecution.fetchcCs\|j|j}|std|jdtjt|dtj d|_ tjt|dtj d|_ dS)z"Refresh execution data from Redis.z Execution z not found in Rediss created_at)tzslast_heartbeatN) rZhgetallr% ValueErrorrr fromtimestampfloatrutcrr)rdatarrrr*:s zExecution.refreshr#cCst|\}}||||dS)z>A combination of job_id and execution_id separated by a colon.r))r)r+r#rr execution_idrrrfrom_composite_keyBs zExecution.from_composite_keyr'ttlpipeliner cCsZtj}|||j|jd}|j||dt|j|dj|||d|jj|||dd|S)zSave execution data to Redis.r))r5r6rrr,r5r6F)r6r5xx) rhexrrsaveExecutionRegistryaddstarted_job_registry add_execution)r+r'r5r6rr,rrrcreateHs zExecution.createNcCs8|dur|n|j}|j|j|d||j|dS)z6Save execution data to Redis and JobExecutionRegistry.N)mapping)rhsetr% serializeexpire)rr5r6rrrrr;RszExecution.savecCs:||j|jj||dt|j|jdj||ddS)zDelete an execution from Redis.)r,r6r7N)deleter%r>Zremove_executionr<rrremove)rr'r6rrrrEYs zExecution.deletecCs|j|j|jdS)N)rrr)rr timestamprr$rrrrC_szExecution.serializer>cCs\t|_||jd|j||j||j|||ddt|j|dj |||ddS)zUpdate execution heartbeat.rT)r5r6r9r7r8N) rrrBr%rGrDr?r<rr=)rr>r5r6rrr heartbeatfs zExecution.heartbeatr)__name__ __module__ __qualname____doc__strrrobjectboolr"propertyr%r r'r# classmethodr&r*r4intr@rr;rEdictrCrrHrrrrrs*   rc@seZdZdZdZdedefddZd dee d ee fd d Z d e d e dddefddZd e dddefddZd!de de de efddZd!de de de e fddZdeddfddZdS)"r<zbClass to represent a registry of job executions. Each job has its own execution registry. zrq:executions:{0}rrcCs||_||_|j||_dSr)rr key_templateformatr%)rrrrrrrwszExecutionRegistry.__init__NrGexception_handlerscCs(|dur|nt}|j|jd|dS)zRemove expired jobs from registry. Removes jobs with an expiry time earlier than timestamp, specified as seconds since the Unix epoch. timestamp defaults to call time if unspecified. Nr)rrZzremrangebyscorer%)rrGrVscorerrrcleanup|szExecutionRegistry.cleanupr,r5r6r r cCs8t|}||j|j|di||j|ddS)aRegister an execution to registry with expiry time of now + ttl, unless it's -1 which is set to +inf Args: execution (Execution): The Execution to add ttl (int, optional): The time to live. Defaults to 0. pipeline (Optional[Pipeline], optional): The Redis Pipeline. Defaults to None. Returns: result (int): The ZADD command result <N)rZzaddr%rrD)rr,r5r6rWrrrr=s zExecutionRegistry.addcCs||j|jS)z"Remove an execution from registry.)Zzremr%r)rr,r6rrrrFszExecutionRegistry.removerstartendcCs$|dd|j|j||DS)&Returns all executions IDs in registrycSsg|]}t|qSr)r).0rrrr sz7ExecutionRegistry.get_execution_ids..)rXrZzranger%)rr[r\rrrget_execution_idssz#ExecutionRegistry.get_execution_idscCs8|||}g}|D]}|tj||j|jdq |S)r]r))r`appendrr&rr)rr[r\Z execution_ids executionsr3rrrget_executionss z ExecutionRegistry.get_executionsr'cCs0|}|D] }|j||dq||jdS)zDelete the registry.)r6r'N)rcrEr%)rr'r6rbr,rrrrEszExecutionRegistry.delete)NN)rrZ)rIrJrKrLrTrMrrrr0listrXrrRrr=rFr`rcr rErrrrr<ps  r<z rq.workerworkerr r'r cCstddlm}|j$}||}tj|||d|_|j|j |d| Wd|jS1s2wY|jS)azPrepares execution for a job. This is called by the main Worker (not the horse) as it prepares for execution. Do not confuse this with worker.prepare_job_execution() which is called by the horse. Args: worker: The worker preparing the execution job: The job to prepare execution for Returns: Execution: The created Execution object r ) WorkerStatusr6N) worker.baserfrr6Zget_heartbeat_ttlrr@r,Z set_stateZBUSYexecute)rer'rfr6Z heartbeat_ttlrrrprepare_executions    rjr6r cCsDtd|j|jd|d|jdur |jj||dd|_dSdS)a@Cleans up the execution of a job. It will remove the job execution record from the StartedJobRegistry and delete the Execution object. Args: worker: The worker to clean up execution for job: The job whose execution is being cleaned up pipeline: Redis pipeline to use for the cleanup zCleaning up execution of job %sNrg)r'r6)loggerdebugrZset_current_job_idr,rE)rer'r6rrrcleanup_executions    rm)loggingrrtypingrrruuidrZredisrZ redis.clientr rhr r'r registryr rutilsrrrrrr< getLoggerrkrjrmrrrrs      ] A