]jTUddlZddlZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z ddl m Z mZmZmZmZmZmZddlmZddlmZddlmZddlmZdd lmZmZdd lmZm Z dd 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/GddZ0GddZ1ga2eee3d< d$dede4deedeedee5dee4dee5de5dee5dee5dee6d efd!Z7d"ed e1fd#Z8y)%N)datetime timedeltatimezone)AnyCallableDictListOptionalTupleUnion)croniter)Redis)Pipeline)cron_scheduler_registry)DEFAULT_LOGGING_DATE_FORMATDEFAULT_LOGGING_FORMAT)SchedulerNotFound StopRequested)Job)setup_loghandlers)Queue)resolve_serializer)decode_redis_hashnormalize_config_pathnow str_to_date utcformatvalidate_absolute_pathceZdZdZ ddedeedeedeedeedee d eed ee d e d ee d ee dee fdZ de de fdZdefdZdefdZdeddfdZdeeeffdZedeeefddfdZy)CronJobz2Represents a function to be run on a time intervalN queue_namefunc func_nameargskwargsintervalcron job_timeout result_ttlttl failure_ttlmetac X|r |r td|s |s td|r(||_|jd|j|_n|rd|_||_n td|xsd|_|xsi|_||_||_||_ d|_ d|_ |jr8t|jt} | jt|_ || | | | d|_|j j#Dcic] \}}| || c}}|_ycc}}w)Nz0Cannot specify both interval and cron parametersz.Must specify either interval or cron parameter.z)Either func or func_name must be provided)r)r*r+r,r-) ValueErrorr# __module____name__r$r%r&r'r(r" next_run_timelatest_run_timer rget_nextr job_optionsitems)selfr"r#r$r%r&r'r(r)r*r+r,r- cron_iterkvs 1/root/env/lib/python3.12/site-packages/rq/cron.py__init__zCronJob.__init__s OP PMN N ,0DI%)__$5Qt}}o"FDN DI&DNHI I:2 "Lb '/ #' )1537 99 CE2I!*!3!3H!=D &$& , .2-=-=-C-C-EWTQAqDWWs D&D& connectionreturncz|js tdt|j|}|j|jg|j i|j |j}tjtjd|jjd|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#r1rr"enqueuer%r&r7logging getLoggerr3info)r9r?queuejobs r=rCzCronJob.enqueuePsyytu udoo*=emmDIIU UT[[UDDTDTU(#((=9K9K8LJW[WfWfVg)hi c2|jrAt|j|jxs t}|j t S|j r/|jr#|jt|j zSt jS)z@Calculate the next run time based on interval or cron expression)seconds) r(r r5rr6rr'rmax)r9r:s r=get_next_run_timezCronJob.get_next_run_time[sj 99 D,@,@,ICEJI%%h/ / ]]t33'')DMM*JJ J||rIc||js |jsy|jrt|jk\Sy)z Check if this job should run nowTF)r5r(r4rr9s r= should_runzCronJob.should_rungs7##DII   5D... .rItimecn||_|j |j|j|_yy)zrIc|j|j|j|jd}|j |j j Dcic] \}}| || c}}|Scc}}w)z@Convert CronJob instance to a dictionary for monitoring purposes)r$r"r'r()r$r"r'r(updater7r8)r9objr;r<s r=to_dictzCronJob.to_dict|sf// II   T%5%5%;%;%=OTQAqDOP Ps  A2 #A2 datac|di|S)zCreate a CronJob instance from dictionary data for monitoring purposes. Note: The returned CronJob will not have a func attribute and cannot be executed, but contains all the metadata for monitoring. r0r0)clsrXs r= from_dictzCronJob.from_dicts{T{rI) NNNNNNNNNN)r3r2 __qualname____doc__strr rr rintdictr>rrrCrrMboolrPrSrrW classmethodr[r0rIr=r!r!sV< $(#' $!%"&"%)!%)#1X1Xx 1XC= 1X uo 1X  1X3-1Xsm1Xc]1X1Xc]1Xc]1Xtn1Xf % C  8  D ::d: c3h T#s(^ rIr!ceZdZdZej dfdedeee fdefdZ de fdZ de fd Z d,d ed ed eedeedee deedee de dee dee deedefdZdeefdZdeefdZdefdZdZdZdZdefdZedefdZdefd Z d-d!ee!dd fd"Z"d#edd fd$Z#e$dededdfd%Z%e$d.ded&e dedfd'Z&d/d(Z'd-d!ee!dd fd)Z(d/d*Z)edee*fd+Z+y )0 CronSchedulerz*Simple interval-based job scheduler for RQr? logging_levelnamec$||_g|_tj|_t j |_|xs=|jd|jdtjjdd|_ d|_ t|_t|_t#j$t&|_|j(j+s-t-|t&t.t0d|j(_yy)N:rf)levelrh log_format date_formatF)r? _cron_jobssocket gethostnamehostnameosgetpidpiduuiduuid4hexrh config_filer created_atr serializerrDrEr3log hasHandlersrrr propagate)r9r?rgrhs r=r>zCronScheduler.__init__s ",)+#//1  UDMM?!DHH:Qtzz|?O?OPRQR?S>T!U "$'E,.#*#4#4X#>xx##% #17   "'DHH &rIr@cbt||jsy|j|jk(S)z;Equality does not take the database/connection into accountF) isinstance __class__rh)r9others r=__eq__zCronScheduler.__eq__s&%0yyEJJ&&rIc,t|jS)z;The hash does not take the database/connection into account)hashrhrOs r=__hash__zCronScheduler.__hash__sDIIrINr#r"r%r&r'r(r)r*r+r,r-c >t||||||||| | |  } |jj| |jd|j} |r'|j j d| d|d|d| S|r%|j j d| 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!roappendr2r3r|rF)r9r#r"r%r&r'r(r)r*r+r,r-cron_jobjob_keys r=registerzCronScheduler.registers!#!#   x(__%Qt}}o6  HHMML j\QYPZZbc d HHMML j\I^_c^ddef grIc|jS)zGet all registered cron jobs)rorOs r=get_jobszCronScheduler.get_jobss rIct}g}|jD]P}|js|j|j|j ||j |R|S)z$Enqueue all jobs that are due to run)rrorPrCr?rSr)r9 enqueue_time enqueued_jobsrHs r= enqueue_jobszCronScheduler.enqueue_jobss_u ') ?? *C~~ DOO,  .$$S)  * rIct}|jDcgc]}|js|j}}|syt|}||z j }|dkryt|dScc}w)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. <r)rror4min total_seconds)r9 current_timerHnext_job_times closest_timeseconds_until_nexts r=calculate_sleep_intervalz&CronScheduler.calculate_sleep_intervals} u 8<\#J[J[#++\\>* +\9HHJ  "%r**!]s A.A.ctjtj|jtjtj|jy)z.Install signal handlers for graceful shutdown.N)signalSIGINT _request_stopSIGTERMrOs r=_install_signal_handlersz&CronScheduler._install_signal_handlers s2 fmmT%7%78 fnnd&8&89rIcd|jjd|j|t)z#Handle shutdown signals gracefully.z-CronScheduler %s: received shutdown signal %s)r|rFrhr)r9signumframes r=rzCronScheduler._request_stops$  EtyyRXYorIc|jjd|j|j|j  |j |j |j}|dkDr4|jjd|dtj|j#t$r)|jjd|jYn4t$r)|jjd|jYnwxYw|j|jjd|jy#|j|jjd|jwxYw) zStart the cron schedulerzCronScheduler %s: starting...rz Sleeping for z seconds...z,CronScheduler %s: received KeyboardInterruptz CronScheduler %s: stop requestedz#CronScheduler %s: shutdown completeN)r|rFrhrregister_birthr heartbeatrdebugrQsleepKeyboardInterruptrregister_death)r9 sleep_times r=startzCronScheduler.starts'  5tyyA %%'  L!!# !::< >HHNN]:,k#JKJJz* ! U HHMMH$)) T I HHMM||_|jjd|gatj j |r!|jjd|t|dtj j|jdd} tjj||}| |j+d|}|jj|t!|tjj#|}|t$j&|<|jj)||jjd|nt-|}|jjd | |t$j&vr'tj.t$j&|ntj0||jjd |d}tD]E} |jjd| dj2 |j4di| |dz }Gga|jjd|d|dy#t*$rW}|t$j&vrt$j&|=d |d |}|jj|t!||d}~wwxYw#t $r7}d |d|d|}|jj|t!||d}~wt*$r7}d|d|d|}|jj|t+||d}~wwxYw#t*$r;}|jjd| dj2d|dYd}~d}~wwxYw)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: rq_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#rzFailed to register job z from config: T)exc_infozSuccessfully registered z cron jobs from 'rr0)ryr|rF_job_data_registryrspathisabsrrbasenamereplace importlibutilspec_from_file_locationloadererror ImportErrormodule_from_specsysmodules exec_module Exceptionrreload import_moduler3r) r9r module_namespec error_msgmoduleenormalized_path job_countrXs r=load_config_from_filez#CronScheduler.load_config_from_file,sT'  8 FG  77== % HHNN9+G H #; /,BGG,<,<[,I,Q,QRUWZ,[+\]K 4 ~~==k;W<4;;#6"CK= QIHHNN9-%i00"88>+1 K( ''/!G }UV4K@O HHNN..?@ A 2"ckk1$$S[[%AB++O<!I/IZ[\ & rD HHNN:4<;P;P:QR S r %%Q  r   0 ;L[MYZ[\W 4#++- K0A+cRSQTU y)!),!3  4$ 4EoEVV_`k_llpqrpst y)!),!3 2A/ARR[\g[hhlmnlop y) *1 2 r!8f9N9N8O~^_]`alpqq rsW$CI4A,K6M4 K=AKK M 2L M2MM N!0NNc d|jS)z)Redis key for this CronScheduler instancerq:cron_scheduler:)rhrOs r=keyzCronScheduler.keys$DII;//rIc|jt|j|jt |j |j xsdd}|S)z@Convert CronScheduler instance to a dictionary for Redis storagerf)rrrurhrzry)rrr_rurhrrzry)r9rVs r=rWzCronScheduler.to_dictsE txx=II#DOO4++1r   rIpipelinec||n |j}|j|j|j|j |jdy)z2Save CronScheduler instance to Redis hash with TTLN)mappingr)r?hsetrrWexpire)r9rr?s r=savezCronScheduler.saves@!)!5X4?? $,,.9$((B'rIraw_datact|d}|d|_t|jdd|_|d|_t |d|_|d|_y ) z3Restore CronScheduler instance from Redis hash dataT) decode_valuesrrrurrhrzryN) rrrr`getrurhrrzry)r9rrVs r=restorezCronScheduler.restoresY=J swwua()K %c,&78}-rIcd|}|j|}|std|d|||}|j||S)z1Fetch a CronScheduler instance from Redis by namerzCronScheduler with name 'z ' not found)r?rh)hgetallrr)rZrhr?rr schedulers r=fetchzCronScheduler.fetchsX#4&)%%c*#&?v[$QR R:D9 (#rIcleanupcddlm}|rtj|tj|}g}|D]:}|t 5|j ||}|j|ddd<|S#1swYHxYw)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) contextlibrrrget_keysrrr)rZr?rrscheduler_names schedulersrhrs r=allzCronScheduler.alls (  # + +J 71:::F # -D+, -IIdJ7 !!), - - -   - -s $A66A? c"|jjd|jd|jj 5}t j |||j||jdddy#1swYyxYw)zURegister this scheduler's birth in the scheduler registry and save data to Redis hashCronScheduler z: registering birth...N) r|rFrhr?rrrrexecuter9rs r=rzCronScheduler.register_birthso  tyyk1GHI __ % % ' 8 # , ,T8 < IIh        s 8BBc|jjd|jdtj||y)zJRegister this scheduler's death by removing it from the scheduler registryrz: registering death...N)r|rFrhr unregisterrs r=rzCronScheduler.register_deaths0  tyyk1GHI**4:rIc|jj5}|jtj|j t j idd|j|jd|j}|d}|r*|jjd|j dn)|jjd|j ddddy#1swYyxYw) 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?rzaddrget_registry_keyrhrQrrrr|rwarning)r9piperesults zadd_results r=rzCronScheduler.heartbeats__ % % ' s4 II->>@499diikBZ_chlI m KK# &llnG"!*K {:WXY  >$))'O'O'QSWS\S\] =%%e ==rI NNNNNr\NNN)N)T)r@N),r3r2r]r^rDINFOrr r_r`r>rbrrrr r rrar!rr rrfloatrrrrrpropertyrrWrrrrcrrrrrrrr0rIr=reresD4 *1 ''S#X' '2't' #!%!%"&"%)!%)#%%%uo %  % 3- %sm%c]%%c]%c]%tn% %N$w- d7m +%+6:  L2R]R]j0S00  (Xh/(4( ...  % O  UTT/=R2;x'9;T; s  > 2 > >rIrerr#r"r%r&r'r(r)r*r+r,r-r@c |||||||||| | d } tj| tjt} |j d|j} | 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 )rrrDrEr3r2r)r#r"r%r&r'r(r)r*r+r,r-job_dataloggerrs r=rrs> " " Hh'  x (F!4==/2G LL,WI5Mj\Z[ OrIr?ct|}tD]9}tjd|dj|j di|;|S)z8Create a CronScheduler instance with all registered jobsrBzRegistering job: r#r0)rerrDrr3r)r? cron_instancerXs r= create_cronr0sW!Z8M#' )$v,*?*?)@AB &&' rIr)9importlib.utilrrDrsrrprrQrvrrrtypingrrrr r r r r redisr redis.clientrrfrdefaultsrr exceptionsrrrHrlogutilsrrGr serializersrutilsrrrrrrr!rer__annotations__r_r`rarrr0rIr=rsE 22DDD!%I8'+ppuupa>a>J "$DJ# !!"!%!%3 33 5/3 TN 3 sm 3 3- 3#33 #3#3 4.3 3l E m rI