o ia@sUddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z m Z ddl mZmZmZmZddlmZddlmZddlmZddlmZdd lmZmZmZdd lmZmZdd l m!Z!dd l"m#Z#dd l$m%Z%ddl&m'Z'ddl(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1GdddZ2GdddZ3ga4e5e6e7d<dddddedddf dede8dee9dee6dee:dee8dee:de:dee:dee:dee6d e6fd!d"Z;d#ed e3fd$d%Z/home/app/Keep/.python/lib/python3.10/site-packages/rq/cron.py Ysz$CronJob.__init__..) ValueErrorr" __module____name__r#r$r%r&r'r!next_enqueue_timelatest_enqueue_timer rget_nextr job_optionsitems)selfr!r"r#r$r%r&r'r(r)r*r+r, cron_iterr.r.r3__init__(s8   zCronJob.__init__ connectionreturncCsf|jstdt|j|d}|j|jg|jRi|j|j}t t  d|jj d|j|S)z:Enqueue this job to its queue and update the next run timezUCronJob has no function to enqueue. It may have been created for monitoring purposes.r@z Enqueued job z to queue ) r"r5rr!enqueuer$r%r;logging getLoggerr7info)r=r@queuejobr.r.r3rC[s $"zCronJob.enqueuecCsH|jrt|j|jp t}|tS|jr!|jr!|jt|jdStjS)z@Calculate the next run time based on interval or cron expression)seconds) r'r r9rr:rr&rmax)r=r>r.r.r3get_next_enqueue_timefs   zCronJob.get_next_enqueue_timecCs&|js|jsdS|jrt|jkSdS)z Check if this job should run nowTF)r9r'r8rr=r.r.r3 should_runrs  zCronJob.should_runtimecCs,||_|jdus |jdur||_dSdS)zd?Z)dEd@dAZ*edee+fdBdCZ,d S)G CronSchedulerz*Simple interval-based job scheduler for RQr@ logging_levelnamecCs||_g|_t|_t|_|p$|jd|jdt j dd|_ d|_ t |_t|_tt|_|jsJt|tttdd|j_dSdS)N:ra)levelrcZ log_formatZ date_formatF)r@ _cron_jobssocket gethostnamehostnameosgetpidpiduuiduuid4hexrc config_filer created_atr serializerrDrEr7log hasHandlersrrr propagate)r=r@rbrcr.r.r3r?s$  ,   zCronScheduler.__init__rAcCst||jsdS|j|jkS)z;Equality does not take the database/connection into accountF) isinstance __class__rc)r=otherr.r.r3__eq__s  zCronScheduler.__eq__cCs t|jS)z;The hash does not take the database/connection into account)hashrcrLr.r.r3__hash__s zCronScheduler.__hash__Nr"r!r$r%r&r'r(r)r*r+r,c  Cst||||||||| | | d } |j| |jd|j} |r2|jd| d|d|d| S|rD|jd| d|d|d| S) z2Register a function to be run at regular intervals) r!r"r$r%r&r'r(r)r*r+r,r-z Registered 'z ' to run on z every z secondsz with cron schedule '')r rgappendr6r7rtrF)r=r"r!r$r%r&r'r(r)r*r+r,Zcron_jobjob_keyr.r.r3registers*   zCronScheduler.registercCs|jS)zGet all registered cron jobs)rgrLr.r.r3get_jobsszCronScheduler.get_jobscCsBt}g}|jD]}|r||j||||q|S)z$Enqueue all jobs that are due to run)rrgrMrCr@rOr~)r=Z enqueue_timeZ enqueued_jobsrHr.r.r3 enqueue_jobss    zCronScheduler.enqueue_jobscCsHt}dd|jD}|sdSt|}||}|dkrdSt|dS)zCalculate how long to sleep until the next job is due. Returns the number of seconds to sleep, with a maximum of 60 seconds to ensure we check regularly. cSsg|]}|jr|jqSr.)r8r0rHr.r.r3 -sz:CronScheduler.calculate_sleep_interval..<r)rrgmin total_seconds)r= current_timeZnext_job_timesZ closest_timeZseconds_until_nextr.r.r3calculate_sleep_interval$s  z&CronScheduler.calculate_sleep_intervalcCs$ttj|jttj|jdS)z.Install signal handlers for graceful shutdown.N)signalSIGINT _request_stopSIGTERMrLr.r.r3_install_signal_handlers?sz&CronScheduler._install_signal_handlerscCs|jd|j|t)z#Handle shutdown signals gracefully.z-CronScheduler %s: received shutdown signal %s)rtrFrcr)r=signumframer.r.r3rDszCronScheduler._request_stopc Cs |jd|j||zgz' |}|r|||}|dkr8|j d|dt |qt yI|jd|jYnt yY|jd|jYnwW||jd|jd SW||jd|jd S||jd|jw) zStart the cron schedulerzCronScheduler %s: starting...Trz Sleeping for z seconds...z,CronScheduler %s: received KeyboardInterruptz CronScheduler %s: stop requestedz#CronScheduler %s: shutdown completeN)rtrFrcrregister_birthrsave_jobs_data heartbeatrdebugrNsleepKeyboardInterruptrregister_death)r=ZenqueuedZ sleep_timer.r.r3startIs6   zCronScheduler.start config_pathc Cs||_|jd|gatj|r|jd|t|dtj | dd}z;t j ||}|dus?|jdurNd|}|j|t|t j |}|tj|<|j||jd|Wnty}z|tjvrztj|=d |d |}|j|t||d}~wwt|}|jd |z|tjvrt tj|nt ||jd |WnDty}zd |d|d|}|j|t||d}~wty}zd|d|d|}|j|t||d}~wwd}tD]B} |jd| djz|jdi| |d7}WqtyH}z|jjd| djd|ddWYd}~qd}~wwga|jd|d|ddS)a Dynamically load a cron config file and register all jobs with this Cron instance. Supports both dotted import paths (e.g. 'app.cron_config') and file paths (e.g. '/path/to/app/cron_config.py', 'app/cron_config.py'). The .py extension is recommended for file paths for clarity. Jobs defined in the config file must use the global `rq.cron.register` function. Args: config_path: Path to the cron_config.py file or module path. z Loading cron configuration from zLoading absolute file path: Zrq_cron_config_r-_Nz!Could not create module spec for z&Successfully loaded config from file: z#Failed to load configuration file 'z': zNormalized path: z(Successfully loaded config from module: z'Failed to import configuration module 'z ' (from 'z'): z#An error occurred while importing 'rzRegistering job from config: r"r zFailed to register job z from config: T)exc_infozSuccessfully registered z cron jobs from 'r}r.)rqrtrF_job_data_registryrkpathisabsrrbasenamereplace importlibutilspec_from_file_locationloadererror ImportErrormodule_from_specsysmodules exec_module Exceptionrreload import_moduler7r) r=r module_namespec error_msgmoduleeZnormalized_pathZ job_countrRr.r.r3load_config_from_fileesn                0z#CronScheduler.load_config_from_filecCs d|jS)z)Redis key for this CronScheduler instancerq:cron_scheduler:)rcrLr.r.r3keys zCronScheduler.keyc Cs>|jt|j|jt|j|jpdtdd|j Dd}|S)z@Convert CronScheduler instance to a dictionary for Redis storageracSg|]}|qSr.rQrr.r.r3rz)CronScheduler.to_dict..)rjrmrcrrrq cron_jobs) rjr[rmrcrrrrqrUdumpsrg)r=rPr.r.r3rQszCronScheduler.to_dictpipelinecCs8|dur|n|j}|j|j|d||jddS)z2Save CronScheduler instance to Redis hash with TTLN)mappingr)r@hsetrrQexpire)r=rr@r.r.r3saveszCronScheduler.savecCs,tdd|jD}|j|jd|dS)zSave cron jobs data to Redis.cSrr.rrr.r.r3rrz0CronScheduler.save_jobs_data..rN)rUrrgr@rr)r=rRr.r.r3rszCronScheduler.save_jobs_dataraw_datac Cst|dd}|d|_t|dd|_|d|_t|d|_|d|_|d r`zt |d }d d |D|_ Wd St j t tfy_}z|jd |g|_ WYd }~d Sd }~wwg|_ d S)z4Restore CronScheduler instance from Redis hash data.T)Z decode_valuesrjrmrrcrrrqrcSsg|]}t|qSr.)r rX)r0job_datar.r.r3rsz)CronScheduler.restore..zFailed to restore cron jobs: N)rrjr]rSrmrcrrrrqrUrVrgJSONDecodeErrorKeyError TypeErrorrtwarning)r=rrPZ jobs_datarr.r.r3restores      zCronScheduler.restorecCsBd|}||}|std|d|||d}|||S)z2Fetch a CronScheduler instance from Redis by name.rzCronScheduler with name 'z ' not found)r@rc)Zhgetallrr)rWrcr@rr schedulerr.r.r3fetchs    zCronScheduler.fetchTcleanupc Cstddlm}|r t|t|}g}|D]!}|t|||}||Wdn1s2wYq|S)aReturns all CronScheduler instances from the registry Args: connection: Redis connection to use cleanup: If True, removes stale entries from registry before fetching schedulers Returns: List of CronScheduler instances r)suppressN) contextlibrr rZget_keysrrr~)rWr@rrZscheduler_namesZ schedulersrcrr.r.r3alls      zCronScheduler.allcCsd|jd|jd|j}t|||||WddS1s+wYdS)zURegister this scheduler's birth in the scheduler registry and save data to Redis hashCronScheduler z: registering birth...N) rtrFrcr@rr rrexecuter=rr.r.r3rs     "zCronScheduler.register_birthcCs&|jd|jdt||dS)zJRegister this scheduler's death by removing it from the scheduler registryrz: registering death...N)rtrFrcr unregisterrr.r.r3rszCronScheduler.register_deathcCs|jI}|jt|jtiddd||jd| }|d}|r3|j d|jdn|j d|jdWddSWddS1sQwYdS) zSend a heartbeat to update this scheduler's last seen timestamp in the registry and extend the scheduler's Redis hash TTL. T)xxchxrrz: heartbeat sent successfullyz4: heartbeat failed - scheduler not found in registryN) r@rZzaddr get_registry_keyrcrNrrrrtrr)r=piperesultsZ zadd_resultr.r.r3r!s  "zCronScheduler.heartbeatcCs0|jt|j}|durdStj|tjdS)zReturn the UTC datetime of the last heartbeat, or None if no heartbeat recorded Returns: datetime: UTC datetime of the last heartbeat, or None if scheduler not found in registry N)tz) r@Zzscorer rrcr fromtimestamprutc)r=Zscorer.r.r3last_heartbeat1szCronScheduler.last_heartbeatr/)rAN)T)-r7r6rYrZrDINFOr rr[r]r?r^rzr|rrrrTr\r rlistrrfloatrrrrrpropertyrrQr rrrr_rrrrrrrr.r.r.r3r`s       ' U    r`rr"r!r$r%r&r'r(r)r*r+r,rAc  Cs\|||||||||| | d } t| tt} |jd|j} | d| d|| S)a Register a function to be run as a cron job by adding its definition to a temporary global registry. This function should typically be called from within a cron configuration file that will be loaded using `CronScheduler.load_config_from_file()`. Example (in your cron_config.py): from rq import cron from my_app.tasks import my_func cron.register(my_func, 'default', interval=60) # Run every 60 seconds Returns: dict: The job data dictionary added to the registry. ) r"r!r$r%r&r'r(r)r*r+r,r-zCron config: Adding job 'z' to registry for queue )rr~rDrEr7r6r)r"r!r$r%r&r'r(r)r*r+r,rloggerrr.r.r3rEs"  rr@cCs>t|d}tD]}td|dj|jdi|q|S)z8Create a CronScheduler instance with all registered jobsrBzRegistering job: r"Nr.)r`rrDrr7r)r@Z cron_instancerRr.r.r3 create_cron{s r)=importlib.utilrrUrDrkrrhrrNrnrrrtypingrrrrr Zredisr Z redis.clientr rar defaultsrrr exceptionsrrrHrZlogutilsrrGrZ serializersrutilsrrrrrrrrrr r`rrr\__annotations__r[rTr]rrr.r.r.r3s~         , $|    6