s,g4ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z m Z mZddlmZddlmZddlmZddlmZmZmZmZmZmZmZddlmZer ddlmZdd l!m"Z"dd l#m$Z$m%Z%m&Z& dd lm'Z'dd lm)Z)ddl*Z!ddl+m,Z,ddl-m.Z.m/Z/m0Z0ddl1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7ddl8m9Z9m:Z:m;Z;ddlm?Z?ddl@mAZAmBZBmCZCddlDmEZEmFZFmGZGmHZHddlImJZJddlKmLZLmMZMddlNmOZOddlPmQZQddlRmSZSddlTmUZUmVZVmWZWddlXmYZYmZZZm[Z[m\Z\m]Z]m^Z^m_Z_m`Z`maZaddlbmcZc ddldmdZeejd#ZhGd$d%eiZjekd&eleDZmd'ZnGd(d)efeZoGd*d+efeZpGd,d-ZqGd.d/eqZrGd0d1erZsGd2d3erZtGd4d5erZuGd6d7erZvy#e $rYQwxYw#e $r dd lm(Z'YLwxYw#e $r d efd!dfd"ZeYwxYw)8N)datetime timedeltatimezone)Enum)shuffle) FrameType) TYPE_CHECKINGCallableListOptionalTupleTypeUnion)uuid4) struct_rusage)Redis)PipelinePubSubPubSubWorkerThread)SIGKILL)SIGTERM)suppress)worker_registration)PUBSUB_CHANNEL_TEMPLATEhandle_command parse_payload)DEFAULT_JOB_MONITORING_INTERVALDEFAULT_LOGGING_DATE_FORMATDEFAULT_LOGGING_FORMAT!DEFAULT_MAINTENANCE_TASK_INTERVALDEFAULT_RESULT_TTLDEFAULT_WORKER_TTL)DequeueTimeoutDeserializationErrorShutDownImminentException) Execution)Group)Job JobStatusRetry)bluegreensetup_loghandlersyellowQueue)StartedJobRegistryclean_registries) RQScheduler)resolve_serializer) is_suspended)HorseMonitorTimeoutExceptionJobTimeoutExceptionUnixSignalDeathPenalty) as_text backend_classcompact ensure_listget_connection_from_queues get_versionnow utcformatutcparse)VERSION) setproctitletitlereturncyN)rEs c/var/www/trellinator.diamondhoofcare.com/public_html/venv/lib/python3.12/site-packages/rq/worker.py setprocnamerKKs z rq.workerc eZdZy) StopRequestedN)__name__ __module__ __qualname__rIrLrJrNrNRsrLrNc#nK|]-}|jdsd|vstt||f/yw)SIG_N) startswithgetattrsignal).0signames rJ rZVs8,3gFXFXY^F_dgovdvWVWw's 555c tjdddk\rtj|jSt |S#t $rYyt$rYywxYw)N) SIG_UNKNOWN)sys version_inforWSignalsname _signamesKeyError ValueError)signums rJ signal_namerh[sY    BQ 6 )>>&).. .V$ $  s4AA A AAceZdZdZdZdZy)DequeueStrategydefault round_robinrandomN)rOrPrQDEFAULT ROUND_ROBINRANDOMrIrLrJrjrjhsGK FrLrjceZdZdZdZdZdZy) WorkerStatusstarted suspendedbusyidleN)rOrPrQSTARTED SUSPENDEDBUSYIDLErIrLrJrrrrnsGI D DrLrrceZdZdZej ZeZe Z e Z dZ dZdZdZdedddedddddeddddfdeeded d ed eed eed eeddeeddedededeee eedgdffdZe dsdedd d eeddeeddedf dZe dtded d eeddeeddeddedf dZeduded deddeefdZeduded deddefdZ dZ!e"d Z#dvd!Z$e"defd"Z%e"defd#Z&d$Z'd%Z(d&Z)deefd'Z*deefd(Z+e"d)Z,e"d*Z-e"defd+Z.d,Z/d-Z0d.ed/ee1fd0Z2d1Z3dwd3Z4dd4e5e6ddde7jpfd5ed6ed7ed8ed9eed:eed;edd>ed?fdEZ?dudFeed>ed?fdGZ@d{d>ed?deefdHZAdedfdIZBd{dJed>ed?fdKZCdLZDdefdMZEdNZFe"eFeDZGdd4e5e6fd5ed6ed7ed8efdOZHdPZIdQZJe"dRZKd4e5e6fd6ed7ed8efdSZLd5efdTZMdUZNdVZOe"dWZPe"dXZQdYZRdZeSd[d\d]d^ddfd_ZTd`ZUdaZVd2ddefdbZWd2ddeXfdcZYddZZ d{deeed:eedee[dffdgZ\dudeeed>ed?fdhZ]diZ^djZ_d{d>ed?fdkZ`d{d>ed?fdlZadmebd>d?fdnZcd|doZddpZeeffdqegjfdrZiy)} BaseWorker rq:worker:T@gN@NFrc connectionrmaintenance_intervaldefault_worker_ttl worker_ttl job_classr) queue_classr1log_job_description!disable_default_exception_handlerprepare_for_workwork_horse_killed_handlerrc 0||_| r| |_n1|r$tjdtd||_n t |_| |_||_|s t|}|sJ|j|}||_ d|_ t|d| |_ t|d| |_t|_t"j |_t'||_d|_t-|Dcgc]H}t/|t0r4|j|||j|j(|j2n|J}}|xst5j6|_||_|j=|j:dd|_g|_ ||_!d|_"d|_#d |_$d |_%d |_&d|_'tP|_)| |_*d|_+d |_,d |_-d |_.d |_/d|_0d|_1d|_2d|_3thjj|_6||_7|rtqjr|_:twjx|_= |j}|j8|jDcgc]&}|jd |j8k(s"|d (}}t|d kDr |d |_Bn7tjd td|_Bnd|_:d|_=d|_Bt/|ttfr|D]}|j|y||j|yycc}wcc}w#tjj$r$tjdtd|_BYwxYw)Nz1default_worker_ttl is deprecated, use worker_ttl.r\) stacklevelr)overrider)rcrr serializerdeath_penalty_classstartingFrrcaddrz@CLIENT LIST command not supported, setting ip_address to unknownunknownzCCLIENT SETNAME command not supported, setting ip_address to unknown)Jdefault_result_ttlrwarningswarnDeprecationWarningr#job_monitoring_intervalrr>_set_connectionrredis_server_versionr;rrrCversionr`python_versionr5r executionr= isinstancestrrrhexrcqueuesvalidate_queues_ordered_queues _exc_handlers_work_horse_killed_handler_shutdown_requested_date_state _is_horse _horse_pid_stop_requested_stopped_job_idloggerlogrlast_cleaned_atsuccessful_job_countfailed_job_counttotal_working_timecurrent_job_working_time birth_date schedulerpubsub pubsub_threadrjrn_dequeue_strategyrsocket gethostnamehostnameosgetpidpidclient_setname client_listgetlen ip_addressWarningredis exceptions ResponseErrorlisttuplepush_exc_handler)selfrrcrr exc_handlerexception_handlersrrrrrrrrrrrqclientclient_adresseshandlers rJ__init__zBaseWorker.__init__sQ(#5 (DO  MMMOano p0DO0DO'>$$8!3F;Jz))*5 $$(!&t[9M(}{S !kk,Z8.2!(  a%  )"nn#(,(@(@ !    ,   #{{1~-/*C'<@%% $ %*##6 #)*!%&)*/0%04*. !NN worker_keyrFc |j}|j|std|z|j|s|j |j |y|t |d}|g||||d|}|j|S)a@Returns a Worker instance, based on the naming conventions for naming the internal Redis keys. Can be used to reverse-lookup Workers by their Redis keys. Args: worker_key (str): The worker key connection (Optional[Redis], optional): Redis connection. Defaults to None. job_class (Optional[Type[Job]], optional): The job class if custom class is being used. Defaults to None. queue_class (Optional[Type[Queue]]): The queue class if a custom class is being used. Defaults to None. serializer (Any, optional): The serializer to use. Defaults to None. Raises: ValueError: If the key doesn't start with `rq:worker:`, the default worker namespace prefix. Returns: worker (Worker): The Worker instance. zNot a valid RQ worker key: %sNF)rrrrr)redis_worker_namespace_prefixrUrfexistssremredis_workers_keysrrefresh) clsrrrrrprefixrcworkers rJ find_by_keyzBaseWorker.find_by_keys422$$V,+B"CD "&D  &wu~6DO"DO $'0@(A$BD ! (+G4H,I(JD % &+G4F,G&HD # #,1':R2S,TD ) V_F $\\#.   doo\`\k\k!DK s:Gc||jyt|jz t|jkDryy)zBMaintenance tasks should run on first startup or every 10 minutes.TsecondsF)rr@rrrs rJshould_run_maintenance_tasksz'BaseWorker.should_run_maintenance_taskss9    ' ED(( (Id>W>W,X XrLc|jjjd}|||jkr3d|ji}|jjj ||S)aConfigures the Redis connection's socket timeout. This will timeout the connection in case any specific command hangs at any given time (eg. BLPOP), but also ensures that the timeout is long enough for those operations. If the connection provided already has an adequate `socket_timeout` defined, skips. Args: connection (Optional[Redis]): The Redis Connection. socket_timeout)connection_poolconnection_kwargsrconnection_timeoutupdate)rrcurrent_socket_timeouttimeout_configs rJrzBaseWorker._set_connectionsf",!;!;!M!M!Q!QRb!c ! )-CdF]F]-].0G0GHN  & & 8 8 ? ? OrLc4td|jdz S)Nr)maxrrs rJdequeue_timeoutzBaseWorker.dequeue_timeouts1doo*++rLc |jdzS)N )rrs rJrzBaseWorker.connection_timeouts##b((rLc`|jD]}|js|jjd|jt ||j tj||jj|||jt|_ y)z1Runs maintenance jobs on each Queue's registries.z!Cleaning registries for queue: %sN)racquire_maintenance_lockrinforcr3rrclean_worker_registryintermediate_queuecleanuprelease_maintenance_lockr@rrrs rJr3zBaseWorker.clean_registriess[[ 1E--/ A5::N (:(:;#99%@((00u=..0 1 #urLcf|jst|j|_|jS)z)Return Redis server version of connection)rr?rrs rJget_redis_server_versionz#BaseWorker.get_redis_server_versions((((3DOO(DD %(((rLc|jD]=}t||jrtdj ||jy)z"Sanity check for the given queues.z&{0} is not of type {1} or string typesN)rrr TypeErrorformatrs rJrzBaseWorker.validate_queuessH[[ jEeT%5%56 H O OPUW[WgWg hii jrLcT|jDcgc]}|jc}Scc}w)zsReturns the queue names of this worker's queues. Returns: List[str]: The queue names. )rrcrs rJ queue_nameszBaseWorker.queue_namess )- 4u 444%cT|jDcgc]}|jc}Scc}w)zReturns the Redis keys representing this worker's queues. Returns: List[str]: The list of strings with queues keys )rrrs rJ queue_keyszBaseWorker.queue_keyss (,{{3e 333rc4|j|jzSz$Returns the worker's Redis hash key.)rrcrs rJrzBaseWorker.keys11DII==rLc(t|jzSr)rrcrs rJpubsub_channel_namezBaseWorker.pubsub_channel_names'22rLc(|jdk\S)z2Only supported by Redis server >= 5.0 is required.)r^rr)rrs rJsupports_redis_streamsz!BaseWorker.supports_redis_streamss,,.);;rLcd|jjdt|t|_t j t j |jt j t j|j|j|jy)zStops the current worker loop but waits for child processes to end gracefully (warm shutdown). Args: signum (Any): Signum frame (Any): Frame z Got signal %sN) rdebugrhr@rrWSIGINTrequest_force_stoprhandle_warm_shutdown_request _shutdownrrgframes rJ request_stopzBaseWorker.request_stopsk  F(;<(+% fmmT%<%<= fnnd&=&=> ))+ rLc0|jtjk(rPd|_|j |j j d|jr|jyy|jr|jt)z If shutdown is requested in the middle of a job, wait until finish before shutting down and save the request in redis TzQStopping after current horse is finished. Press Ctrl+C again for a cold shutdown.N) get_staterrryrset_shutdown_requested_daterr!rstop_schedulerrNrs rJr%zBaseWorker._shutdownss >> |00 0#'D  , , . HHNNn o~~##%~~##%/ !rLrgr'ctrHNotImplementedErrorr&s rJr#zBaseWorker.request_force_stop" !##rLctjtj|jtjtj|jy)zDInstalls signal handlers for handling SIGINT and SIGTERM gracefully.N)rWr"r(rrs rJ_install_signal_handlersz#BaseWorker._install_signal_handlers%s2 fmmT%6%67 fnnd&7&78rLjobct)z To be implemented by subclasses.r.rr3rs rJ execute_jobzBaseWorker.execute_job*s!!rLINFOburst logging_level date_format log_formatmax_jobs max_idle_timewith_schedulerdequeue_strategyc `|j|||||_d} |r|j|||||j |j ||j r|j |jr'|jjd|jn|rdn |j} |j| |} | S|r'|jjd|jn)|'|jjd|j|n]| \} } |j| | |j| dz } |-| |k\r(|jjd|j| n5|j+t-| S#tj j"$r)|jj%d |jY`t&$rYjt($r|jj%d |jd YxYw#|j+wxYw) a:Starts the work loop. Pops and performs all jobs on the current list of queues. When all queues are empty, block and wait for new jobs to arrive on any of the queues, unless `burst` mode is enabled. If `max_idle_time` is provided, worker will die when it's idle for more than the provided value. The return value indicates whether any jobs were processed. Args: burst (bool, optional): Whether to work on burst mode. Defaults to False. logging_level (str, optional): Logging level to use. Defaults to "INFO". date_format (str, optional): Date Format. Defaults to DEFAULT_LOGGING_DATE_FORMAT. log_format (str, optional): Log Format. Defaults to DEFAULT_LOGGING_FORMAT. max_jobs (Optional[int], optional): Max number of jobs. Defaults to None. max_idle_time (Optional[int], optional): Max seconds for worker to be idle. Defaults to None. with_scheduler (bool, optional): Whether to run the scheduler in a separate process. Defaults to False. dequeue_strategy (DequeueStrategy, optional): Which strategy to use to dequeue jobs. Defaults to DequeueStrategy.DEFAULT Returns: worked (bool): Will return True if any job was processed, False otherwise. rTzWorker %s: stopping on requestNzWorker %s: done, quittingz(Worker %s: idle for %d seconds, quittingrz/Worker %s: finished executing %d jobs, quittingz0Worker %s: Redis connection timeout, quitting...z4Worker %s: found an unhandled exception, quitting...)exc_info) bootstrapr_start_schedulerr2check_for_suspensionrrun_maintenance_tasksrrr rrdequeue_job_and_maintain_ttlr6 heartbeatrr TimeoutErrorerrorrN SystemExitteardownbool)rr8r9r:r;r<r=r>r?completed_jobstimeoutresultr3rs rJworkzBaseWorker.work.sD }k:>!1   ! !% Z P %%'. *--e488224++ &FQ&+d1E1EG!>>w VF~  HHMM*EtxxP*6 HHMM*TVZV^V^`mn!'JC$$S%0NN$"a'N+)X5 HHMM*[]a]e]eguv!9Z MMON##!''44HHNN#UW[W_W_`$!HHNN#Y[_[c[cnrNs MMOsXHAF%H&A6FHAF7HAHHH"H$2HHH-pipelinerc|jjd|jt|j|j |j |j}|jd||j|||jrC|j|j|||jj||d|_ yy)zCleans up the execution of a job. It will remove the job from the `StartedJobRegistry` and delete the Execution object. zCleaning up execution of job %srrNrQ)r3rQ) rr!idr2originrrrset_current_job_idremoverremove_executiondelete)rr3rQstarted_job_registrys rJcleanup_executionzBaseWorker.cleanup_executions 8#&&A1 JJ4>>doo  x8##C(#; >> 1 1$..cT\ 1 ] NN ! !cH ! =!DN rLcf|jjd|j|jy)Nz,Worker %s [PID %d]: warm shut down requested)rr rcrrs rJr$z'BaseWorker.handle_warm_shutdown_requests"  DdiiQUQYQYZrLc|jtj|_|jdvrtd|jd|jtjk(ry|jtjk(rF|j j |}|j |dzd|j d|dzz|_y|jtjk(rt|j yy)aRReorder the queues according to the strategy. As this can be defined both in the `Worker` initialization or in the `work` method, it doesn't take the strategy directly, but rather uses the private `_dequeue_strategy` attribute. Args: reference_queue (Union[Queue, str]): The queues to reorder N)rkrmrlzDequeue strategy z: is not allowed. Use `default`, `random` or `round_robin`.r) rrjrnrfrorindexrprrreference_queueposs rJreorder_queueszBaseWorker.reorder_queuess  ! ! )%4%<%>VZVeVe($ "11SVV;N$$;^);Ey008D'+$NN9#3#3hNG  " "3 " :99%##J#Be..>>?'$$&'  + +H 5~~#,,11#,,2OQYZ %*%*"%)"   "%,,S1O+ + .''"  Q+ + sDD H G;A3H#H;H H HHHHHrcl||_||n |j}|j|jd|y)zSets the current job working time in seconds Args: current_job_working_time (float): The current job working time in seconds pipeline (Optional[Pipeline], optional): Pipeline to use. Defaults to None. Nr)rrhsetr)rrrQrs rJset_current_job_working_timez'BaseWorker.set_current_job_working_times3)A%!)!5X4?? "<>VWrLrc||n |j}||j|jdy|j|jd|y)aSets the current job id. If `None` is used it will delete the current job key. Args: job_id (Optional[str], optional): The job id. Defaults to None. pipeline (Optional[Pipeline], optional): The pipeline to use. Defaults to None. Nr)rhdelrrw)rrrQrs rJrWzBaseWorker.set_current_job_ids?"*!5X4?? > OODHHm 4 OODHHmV vrLc|j}|y|jj||j|jS)zqReturns the currently executing job instance. Returns: job (Job): The job instance. N)r}rfetchrr)rrs rJget_current_jobzBaseWorker.get_current_jobs; ((* >~~##FDOOT__MMrLrcl||_||n |j}|j|jd|y)zSets the worker's state. Args: state (str): The state pipeline (Optional[Pipeline], optional): The pipeline to use. Defaults to None. Nr)rrrwr)rrrQrs rJ set_statezBaseWorker.set_states/ !)!5X4?? '51rLcZtjdt|j|y)z:Raise a DeprecationWarning if ``worker.state = X`` is usedz;worker.state is deprecated, use worker.set_state() instead.N)rrrr)rrs rJ _set_statezBaseWorker._set_state(s SUgh urLc|jSrH)rrs rJr*zBaseWorker.get_state-s {{rLcVtjdt|jS)z;Raise a DeprecationWarning if ``worker.state == X`` is usedz;worker.state is deprecated, use worker.get_state() instead.)rrrr*rs rJ _get_statezBaseWorker._get_state0s SUgh~~rLcpt|j|j||||j|_|jj |jj rR|r5|jj|jjy|jjyy)aStarts the scheduler process. This is specifically designed to be run by the worker when running the `work()` method. Instanciates the RQScheduler and tries to acquire a lock. If the lock is acquired, start scheduler. If worker is on burst mode just enqueues scheduled jobs and quits, otherwise, starts the scheduler in a separate process. Args: burst (bool, optional): Whether to work on burst mode. Defaults to False. logging_level (str, optional): Logging level to use. Defaults to "INFO". date_format (str, optional): Date Format. Defaults to DEFAULT_LOGGING_DATE_FORMAT. log_format (str, optional): Log Format. Defaults to DEFAULT_LOGGING_FORMAT. )rr9r:r;rN) r4rrrr acquire_locksacquired_locksenqueue_scheduled_jobs release_locksstart)rr8r9r:r;s rJrCzBaseWorker._start_scheduler7s(% KK'#!   $$& >> ( (557,,.$$& )rLc j|jjd|j|jj |j rL|jj |j ds&d}t|j|j|j }dj|j}|jj5}|j|t}t|}||_||||j |j"|j$|j&|j(d}|j+||t-j.|||j1||j2dz|j5dddy#1swYyxYw) zRegisters its own birth.zRegistering birth of worker %sdeathz1There exists an active worker named {0!r} alreadyr)rrrrrrrr)mapping<N)rr!rcrrrhexistsrfrjoinrrQrZr@rArrrrrrrwrregisterexpirerrk)rmsgrrp right_now now_in_stringrs rJregister_birthzBaseWorker.register_birth[sG 7C ?? ! !$(( +DOO4K4KDHHV]4^ECSZZ 23 3hh$**,- __ % % ' 1 HHSMI%i0M'DO'"/ xx MM"oo<<"&"5"5 G FF3F (  ( (q 1 HHS$//B. / IIK)   s CF))F2c x|jjd|jj5}t j |||j |jdtt|j|jd|jdddy#1swYyxYw)zRegisters its own death.zRegistering deathrrN) rr!rrQr unregisterrwrrAr@rrk)rrs rJregister_deathzBaseWorker.register_deathys *+ __ % % ' 1 * *4 3 FF488Wi&6 7 HHTXXr " IIK    s A1B00B9c|jS)zuThe horse's process ID. Only available in the worker. Will return 0 in the horse part of the fork. )rrs rJ horse_pidzBaseWorker.horse_pids rLct||||j|jjd|jt j t|j|jtj|j}|jjdtdj|y)aBootstraps the worker. Runs the basic tasks that should run when the worker actually starts working. Used so that new workers can focus on the work loop implementation rather than the full bootstraping process. Args: logging_level (str, optional): Logging level to use. Defaults to "INFO". date_format (str, optional): Date Format. Defaults to DEFAULT_LOGGING_DATE_FORMAT. log_format (str, optional): Log Format. Defaults to DEFAULT_LOGGING_FORMAT. z)Worker %s started with PID %d, version %s*** Listening on %s...z, N)r.rrr rrrrC subscriberrrrwrr-r)rr9r:r;qnamess rJrBzBaseWorker.bootstraps -jA   A488RYY[Zab  |++,!!#  .dii6G0HIrLcd}d}|jst|j|r|r<|jj d|jj dt |sL|jj d|j }|jtjd}tjd|jst|j|r|r|j|yy)z;Check to see if workers have been suspended by `rq suspend`NFz Suspended in burst mode, exitingz7Note: There could still be unfinished jobs on the queuez+Worker suspended, run `rq resume` to resumeTr) rr6rrr rNr*rrrrxtimesleep)rr8 before_statenotifieds rJrDzBaseWorker.check_for_suspensions &&<+N @A WX## KL#~~/ |556 JJqM&&<+N  NN< ( rLc:td|jd|y)zwChanges the current procname for the process. This can be used to make `ps -ef` output more readable. r}z: N)rKrc)rmessages rJproclinezBaseWorker.proclines j 2gY78rLcx|jj|jdt|jy)zDSets the date on which the worker received a (warm) shutdown requestshutdown_requested_dateN)rrwrrArrs rJr+z&BaseWorker.set_shutdown_requested_dates( TXX'@)DLiLiBjkrLc||jj|jd}|tt |Sy)z+Fetches shutdown_requested_date from Redis.rNrr|rrBr:)rshutdown_requested_timestamps rJrz"BaseWorker.shutdown_requested_dates<(,';';DHHF_'`$ ' 3G$@AB B 4rLc||jj|jd}|tt |Sy)zFetches death date from Redis.rNr)rdeath_timestamps rJ death_datezBaseWorker.death_dates9//..txxA  &GO45 5 'rLc@|jrb|jrV|jjr$|jjjs|jj d|j t j |jy)a\ Runs periodic maintenance tasks, these include: 1. Check if scheduler should be started. This check should not be run on first run since worker.work() already calls `scheduler.enqueue_scheduled_jobs()` on startup. 2. Cleaning registries No need to try to start scheduler on first run T) auto_startrN)rr_processis_aliverr3r(rrs rJrEz BaseWorker.run_maintenance_taskssh   ~~t~~'>'>dnnF]F]FfFfFh,,,=  $//:rLexcrrrrct|tjjr3|jj d|dt jdy|jjd|z)a> This exception handler allows the pubsub_thread to continue & retry to connect after a connection problem the same way the main worker loop indefinitely retries. redis-py internal mechanism will restore the channels subscriptions once the connection is re-established. ACould not connect to Redis instance: %s Retrying in %d seconds...r\r~zPubsub thread exitin on %sN) rrrrjrrIrrwarning)rrrrs rJ_pubsub_exception_handlerz$BaseWorker._pubsub_exception_handlers[ cE,,<< > HHNNS  JJsO HH  9C? @ rLcj|jjd|t|}t||y)zHandle external commandszReceived message: %sN)rr!rr)rrpayloads rJhandle_payloadzBaseWorker.handle_payloads) -w7(tW%rLcL|jjd|j|jj |_|jj di|j|j i|jjdd|j|_ y)z"Subscribe to this worker's channelzSubscribing to channel %sg?T) sleep_timedaemonexception_handlerNrI) rr rrrrr run_in_threadrrrs rJrzBaseWorker.subscribes  143K3KLoo,,.  P!9!94;N;N OP![[6644;Y;Y7 rLc|jrJ|jdkDr;|j|jz }tt||jdzS|jdzSzGet's the TTL for the next heartbeat. Args: job (Job): The Job Returns: int: The heartbeat TTL. rrrNrrminrrr3remaining_execution_times rJget_heartbeat_ttlzBaseWorker.get_heartbeat_ttlZ ;;3;;?'*{{T5R5R'R $s3T5Q5QRSVXX X//"4 4rLcL|jj5}|j|}tj||||_|j tj||jddd|j S#1swY|j SxYw)zThis method 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. rTN) rrQrr'createrrrrryrk)rr3rQ heartbeat_ttls rJprepare_executionzBaseWorker.prepare_executions__ % % ' 8 2237M&--c=8TDN NN<,,xN @       ~~   ~~s A BB#c:|jr|jjd|j|jj |jj |j j|j jyy)zUnsubscribe from pubsub channelzUnsubscribing from channel %sN) rrr rstoprr unsubscribeclosers rJrzBaseWorker.unsubscribesm    HHMM94;S;S T    # # %    # # % KK # # % KK    rLrN)r)r1cDd}dj|j}|jtj|j d|z|j jdt|d}t}|} |j|jr|j|| t||}|j jdt|||jj|j ||j"|j$|j&|j(}||\}} |j+| |j jd t-|j.t| j0|j3|_|j6rO|j j9d t| j0t-|j:|j.n:|j j9d t| j0|j. |j|S#t<$r?|:t|z j?} tAjB|| z }|d krYWYntDjFjH$ra} |j jKd | |tMjN|||jPz}t||jR}Yd} ~ nd} ~ wwxYw)zDequeues a job while maintaining the TTL. Returns: result (Tuple[Job, Queue]): A tuple with the job and the queue. Nrz Listening on rg?z+Dequeueing jobs on queues %s and timeout %s)rrrr)razDequeued job %s from %s %s: %s (%s)z%s: %srr)*rrrrrrzrrr!r-r@rGrrErr dequeue_anyrrrrrrcr,rUrcrrrr  descriptionr$ total_secondsmathceilrrrjrIrrexponential_backoff_factormax_connection_wait_time) rrNr=rOrconnection_wait_time idle_sinceidle_time_leftr3ridle_forconn_errs rJrFz'BaseWorker.dequeue_job_and_maintain_ttl&s$**,- |(() o./ /v?"U &) ` 44..0&>+E!'>:GLeTZm]de))55((#"nn#(,(@(@ 6%!'JC'''>HHNN#> " "t~~'>'>'B'B //33V^^D NN # # ( ( * (C "  sAB!! B-,B-c^||n |j}|j|jddy)zUsed to keep the worker stats up to date in Redis. Increments the failed job count. Args: pipeline (Optional[Pipeline], optional): A Redis Pipeline. Defaults to None. NrrrhincrbyrrrQrs rJrlz%BaseWorker.increment_failed_job_counts+"*!5X4?? 488%7;rLc^||n |j}|j|jddy)zUsed to keep the worker stats up to date in Redis. Increments the successful job count. Args: pipeline (Optional[Pipeline], optional): A Redis Pipeline. Defaults to None. Nrrrrs rJincrement_successful_job_countz)BaseWorker.increment_successful_job_counts+"*!5X4?? 488%;Q?rLjob_execution_timecZ|j|jd|jy)a,Used to keep the worker stats up to date in Redis. Increments the time the worker has been workig for (in seconds). Args: job_execution_time (timedelta): A timedelta object. pipeline (Optional[Pipeline], optional): A Redis Pipeline. Defaults to None. rN) hincrbyfloatrr)rrrQs rJroz'BaseWorker.increment_total_working_times% dhh(<>P>^>^>`arLc:|jjd|jdjt j |} |j |j|jd}|j }|j|j|jd|jjd|j||||jD]/}|jjd|||g|}|d }|r/y y #t$ri}d}YwxYw) aWalks the exception handler stack to delegate exception handling. If the job cannot be deserialized, it will raise when func_name or the other properties are accessed, which will stop exceptions from being properly logged, so we guard against it here. zHandling exception for %s.)func argumentskwargsz)rrz2[Job %s]: exception raised while executing (%s) %s)extrazInvoking exception handler %sNT)rr!rUr tracebackformat_exception func_nameargsrr%rrVrIr)rr3rArsrrr fallthroughs rJhandle_exceptionzBaseWorker.handle_exceptions 3SVV<WWY77BC  1 ]]SZZXE I  szzSVV<=  A3669V`hm  )) G HHNN:G D!#11K""  $ 1E0I 1s 0D DDc:|jj|y)z7Pushes an exception handler onto the exc handler stack.N)rappend)r handler_funcs rJrzBaseWorker.push_exc_handlers !!,/rLsigctrHr.)rrs rJ kill_horsezBaseWorker.kill_horser0rL)NNN)NNNNN)NN)rrrFrr3r)rr1)r3r)rQr)rar1)NrrHr3r))jrOrPrQrrREDIS_WORKER_KEYSrr9rr1rr)rlog_result_lifespanrrrr"r!rr rrrrLr r classmethodrr rrrrpropertyrrrrr3rrrrrrrr(r%rr#r2r6rr rjrnrPr\r$rcrurrxrWr}rrrr*rrrCrrrrBrDrr+rrrErrrrrrr'rrr rFrGrKr,rlrrrorrrrWrbrrIrLrJr|r|us$0!,>>0KI!$# #-(,$E,0$(+//3$( ?27!%`d%v6smv6 W% v6"v6%SMv6SMv6DK(v6d7m,v6"v6,0v6 !v6$$,Hc3_5UW[5[,\#]%v6p ,0/3 ---DK( - d7m, - , --^)-+//3#'  W% DK( d7m,     h  4 j(7"3 j8GCT j`deh`i j j Ux0 U@Q U]` U UDL ,,,)C)) %) j 5T#Y54DI4>>33<<<"" $$Xi5H$9 " #60"&'+$,;,C,CX$X$X$ X$  X$ 3- X$ }X$X$*X$ X$t "[25n XU XV^_iVj X =# =R\I] = 8J+? 8TW=  N% N 2s 2hz.B 2 3 Z ,E#60 "'"'"' "'  "'H< $60 JJJ J0)$)*9lCC 66 ; YYmrv&&   5U 5s 5 U y  FJ>}>5=c]> %' (>@s#s*AUs* +<8J3G<@x 7K@bybT^b!F007$fnn$rLr|c@eZdZedZefdejfdZde e e e e e dffdZ de de e fd Zd#dZd d de fdZd#dZd#dZd$dZd#dZdZd%d d deddfdZd d d d dedefdZd d d d defdZd d d d de fdZd d d d defdZdZd Zd!Z d"Z!y)&rc|jS)z> HHNN94>> J OO     !lrLr3r)rr1ctj}|jtjd<|jtjd<|dk(r""%&& ; > JJL  e , HHQK'DO MM-44Y L MrLc|jrJ|jdkDr;|j|jz }tt||jdzS|jdzSrrrs rJrzWorker.get_heartbeat_ttl%rrLcdx}x}}t|_ |j|jt5|j \}}}ddd |j dd|_|t"j$k(ry|j'}|y|j(|j*k(rW|j,j/d|j0r|j3|j|j5||dy|t6j8t6j:fvr|j<st|_|r.t#j>|rdt#j@|d nd }d ||d } |j,j/d | |jC|||||j5||| yy#1swYwxYw#t$r|j t|jz j|jdk7r]|j|jdzkDrA|j|jdz|j|j Y"|j|YnCt$r8}|jtjk7r|jYd}~nd}~wwxYw)zThe worker will monitor the work horse and make sure that it either executes successfully or the status of the job is set to failed Args: job (Job): _description_ queue (Queue): _description_ Nrrz4Job stopped by user, moving job to FailedJobRegistryz+Job stopped by user, work-horse terminated.rrsz (signal )rz5Work-horse terminated unexpectedly; waitpid returned z; z$Moving job to FailedJobRegistry (%s))"r@rmrrr7rrxrrNrrGrmaintain_heartbeatsrrEINTRrrEX_OK get_statusrrUrrstopped_callbackexecute_stopped_callbackrur*FINISHEDrhrn WIFSIGNALEDWTERMSIGhandle_work_horse_killed) rr3rretpidret_valrr job_status signal_msgrss rJmonitor_work_horsezWorker.monitor_work_horse4sx%)((6 !--d.J.JLhiD.2.A.A.C+FGVD6 ))!, bhh  ^^%     ! !SVV + HH  S T##,,T-E-EF  # #CuAn # o  2 2I4D4DE E<<"u AHBNN[bLc9R[[%9$:!>3I2X2X2Z[;;"$)F)F#++XZJZ)[NN4#?#?"#DEOO%'')((- !77ekk)   !'s; GG GG GB&K 8K  K .KK c|j||j|||j|||jtj y)zSpawns a work horse to perform the actual work and passes it a job. The worker will wait for the work horse and make sure it executes within the given timeout bounds, or will end the work horse with SIGALRM. N)rr$r8rrrrzr5s rJr6zWorker.execute_jobysD s# S%( U+ |(()rLc|jj5}|j|jdz|t |j |}|j j|j|||jt||d|j}|ddk(r%|jj|jdddy#1swYyxYw)z:Updates worker, execution and job's last heartbeat fields.rrTT)rQxxrN) rrQrGrrrrr[r@rkrZr)rr3rQttlresultss rJr*zWorker.maintain_heartbeatss __ % % ' 08 NN477"  - -a( - C 2237M NN=8N < MM#%M B  % %dii( % C-(cjjT__E e::AsvvF     HHNN6 7 8"2 cjj DIIKHI% 8 8s DFF#rpr[c |jjd|j|jro|j|jk\rV|jj d|j|jd|jd}|j |||ytj|jxsd|j}|jj5}|j||j|j|jz ||dkDrt!j"t$j&t)| z}|j+t,j.||j1||||jjd |j||j|jxsdz nU|jjd |j|j|jxsdz |j3|| |j5|||j7|jjd |jdddy#1swYyxYw)a?Handles the retry of certain job. It will remove the job from the `StartedJobRegistry` and requeue or reschedule the job. Args: job (Job): The job that will be retried. queue (Queue): The queue started_job_registry (StartedJobRegistry): The started registry zHandling retry of job %sz/Job %s has exceeded maximum retry attempts (%d)zJob failed after z retry attemptsr(NrrTrz8Job %s: scheduled for retry at %s, %s attempts remainingz1Job %s: enqueued for retry, %s attempts remaining)rrQz!Finished handling retry of job %s)rr!rUnumber_of_retriesrrrur+ get_interval intervalsrrQrlrornrmrr@rutcrrfr* SCHEDULED schedule_job_handle_retry_resultr\rk) rr3rrpr[rsretry_intervalrQscheduled_times rJhandle_job_retryzWorker.handle_job_retrys 1366:  S%:%:eii%G HH  NPSPVPVX]XaXa b,UYYKGJ  # #Cu # L ++C,A,A,FQX __ % % ' H8  + +X + >  - -cllS^^.KX V!!)hll!;iP^>_!_y22XF""3"JNFF"II!6!6!;!< GFFII!6!6!;!< ((ux(H  " "3 " :     HHNN> G7 H H Hs *FI;;Jc|jjd|j|jj 5} |j |j |j|||js|j|j||j|j|jz ||j|j}|dk7r9|jjd|j|j!|||j#||d|jjd|j|j%|||j'|jsJ|jsJ|j|jz }|jj)d|j*||j,|jjd |j d d d y #t.j0j2$rYwxYw#1swYy xYw) aHandles the successful execution of certain job. It will remove the job from the `StartedJobRegistry`, adding it to the `SuccessfulJobRegistry`, and run a few maintenance tasks including: - Resting the current job ID - Enqueue dependents - Incrementing the job count and working time - Handling of the job successful execution Runs within a loop with the `watch` method so that protects interactions with dependents keys. Args: job (Job): The job that was successful. queue (Queue): The queue started_job_registry (StartedJobRegistry): The started registry z'Handling successful execution of job %srTrz+Saving job %s's successful execution resultF)rQremove_from_queuez'Removing job %s from StartedJobRegistryz1Successfully completed %s job in %ss on worker %sz0Finished handling successful execution of job %sN)rr!rUrrQwatchdependents_keyrqexplicit_transactionmultirrornrmget_result_ttlr_handle_successr r\rkr rrcrr WatchError)rr3rr[rQ result_ttl time_takens rJhandle_job_successzWorker.handle_job_successs" @#&&I __ % % ') 8'NN3#5#56,,S8,D#88!(777J55cllS^^6SU]^!$!3!3D4K4K!LJ!Q'TVYV\V\]++J+JKK XQVKWHHNN#LcffU**3*B $$&>>)><<'JHHMMKS__^hjnjsjsHHNN#UWZW]W]^O) ) P''22Q) ) s+H:GHH73H:6H77H::IrcVt|_|jt|y)z(Called after job has finished execution.N)r@rnrG)rr3rrs rJhandle_execution_endedzWorker.handle_execution_endedFsu  ce]+rLc|j}|jjd t|jdk(}|j ||t |_|jxs|jj}|j|t|j5|jjd|j|j}|jjd|jddd|j|||j |_t%|t&r<|jjd|j|j)||||y |j+|j||j-|||  |jjGdtI|jJtMd|j|.|jjdtOtQ||jRry|jU|jV} | dk(r|jjGdy | dkDr|jjGd| y |jjGdy #1swYxYw#|jjd |jt.j0|_|j|||j4t7j8}d j;t=j>|} |j@|jg|n<#t7j8}d j;t=j>|}YnxYw|jB|g||jE|||| YyxYw)zPerforms the actual work of a job. Will/should only be called inside the work horse's process. Args: job (Job): The Job queue (Queue): The Queue Returns: bool: True after finished. zStarted Job Registry set.r)rzPerforming Job %s ...zFinished performing Job %sNzJob %s returns a Retry object)r3rrpr[T)r3rr[zJob %s raised an exception.r)r3rsrr[FrzJob OKz Result: %rrzResult discarded immediatelyzResult is kept for %s secondsz6Result will never expire, clean up result key manually),r[rr!rrrKr@rmrNrDEFAULT_TIMEOUTrr8rUperformrdsuccess_callback_timeout_resultrr+rVexecute_success_callbackrbr*rh_statusfailure_callback_timeoutr`rArrrexecute_failure_callbackrrur r-rVr,r/rrr]r) rr3rr[rFrN return_valuerArsr`s rJrBzWorker.perform_jobKs %99 230 -0-=-B *  & &s,J K UCNkkET%5%5%E%EG))'3Fsvv)V E6?"{{} ;SVVD E  ' 'UC4P4P Q'CK,.>G%%5 Sg&,,T-E-E|T''CuSg'h2  mU3::%6XO  # HHNN<L0A)B C  # #++D,C,CDJQ <=  a =zJ VWs E E*  HHNN8#&& A#**CK  ' 'UC4P4P Q||~H!;!;X!FGJ L,,,T-E-EQQ L<<>WWY%?%?%JK  "D ! !# 1 1  # #JeRf $ sEA?J5(AJ(A7J5=0J5(J2-J55BO M#"O #7N-O c6|jjS)z?Pops the latest exception handler off of the exc handler stack.)rpoprs rJpop_exc_handlerzWorker.pop_exc_handlers!!%%''rLcF|jy|j||||yrH)r)rr3r4r5rs rJr3zWorker.handle_work_horse_killeds%  * * 2  ''VWfErLcvt||js td|j|jk(S)z;Equality does not take the database/connection into accountz2Cannot compare workers to other types (of workers))r __class__rrc)rothers rJ__eq__z Worker.__eq__s/%0PQ QyyEJJ&&rLc,t|jS)z;The hash does not take the database/connection into account)hashrcrs rJ__hash__zWorker.__hash__sDIIrLrr)F)"rOrPrQr rrrWrbrr r rrrr#r$rr8r6r*r!rArLrKr+r2rVrbrdrBrqr3rvryrIrLrJrrsN 07fnn !hsmXc]H_D]&] ^!Xi5H6N& 5U 5s 5CMJ *02$ 6JJPTJaeJ21HE1H'1H%1H_q1Hf<e<G<Se<|,%,,PS, MuMWMM^(F ' rLrc$eZdZddZdddefdZy) SimpleWorkerr3r)c|j||j|||jtjy)z1Execute job in same thread/process, do not fork()N)rrBrrrrzr5s rJr6zSimpleWorker.execute_jobs2 s# e$ |(()rLrFcl|jdk(rtSt|jxstdzS)z-1" means that jobs never timeout. In this case, we should _not_ do -1 + 60 = 59. We should just stick to DEFAULT_WORKER_TTL. Args: job (Job): The Job Returns: ttl (int): TTL r'r)rNr#r)rr3s rJrzSimpleWorker.get_heartbeat_ttls0 ;;" % % 9'9;b@ @rLNr)rOrPrQr6rrrIrLrJr{r{s* AU As ArLr{c4eZdZdZdZgdZdZdZdZdZ y) HerokuWorkerz Modified version of rq worker which: * stops work horses getting killed with SIGTERM * sends SIGRTMIN to work horses on SIGTERM to the main process which in turn causes the horse to crash `imminent_shutdown_delay` seconds later )f_codef_lastif_linenof_localsf_tracec(tjtj|jtjtjtjtjtj tjy)z>Modified to ignore SIGINT and SIGTERM and only handle SIGRTMINN)rWSIGRTMINrequest_stop_sigrtminr"rDrrs rJrAz%HerokuWorker.setup_work_horse_signalssF foot'A'AB fmmV^^4 fnnfnn5rLc|jdk7rG|jjd|j|j t j y|jjdy)z"If horse is alive send it SIGRTMINrzBWorker %s: warm shut down requested, sending horse SIGRTMIN signal)rz(Warm shut down requested, no horse foundN)rrr rrrWrrrs rJr$z)HerokuWorker.handle_warm_shutdown_requestsM >>Q  HHMM^`d`h`h i OOO 0 HH  G HrLc|jdk(r.|jjd|j||y|jjd|jt jtj |jt jtj |jt j|jy)Nrz@Imminent shutdown, raising ShutDownImminentException immediatelyzBImminent shutdown, raising ShutDownImminentException in %d seconds)imminent_shutdown_delayrrrequest_force_stop_sigrtminrWrSIGALRMalarmr&s rJrz"HerokuWorker.request_stop_sigrtmins  ' '1 , HH  _ `  , ,VU ; HH  TVZVrVr  MM&//4+K+K L MM&..$*J*J K LL55 6rLctfd|jD}|jjdt dt |z|)Nc3:K|]}|t|fywrH)rV)rXattrr's rJrZz;HerokuWorker.request_force_stop_sigrtmin..sSTT75$/0Ssz2raising ShutDownImminentException to cancel job...zshut down imminent (signal: %s))dictframe_propertiesrrr&rh)rrgr'r s ` rJrz(HerokuWorker.request_force_stop_sigrtminsFST=R=RSS MN'(IKX^L_(_aeffrLN) rOrPrQ__doc__rrrAr$rrrIrLrJrrs+ O6 I 7grLrceZdZdZdZy)RoundRobinWorkerze Modified version of Worker that dequeues jobs from the queues using a round-robin strategy. c|jj|}|j|dzd|jd|dzz|_y)Nr)rr_r`s rJrczRoundRobinWorker.reorder_queuessK""((9#33C!GI>AUAUV_X[^_X_A``rLNrOrPrQrrcrIrLrJrrs arLrceZdZdZdZy) RandomWorkerz` Modified version of Worker that dequeues jobs from the queues using a random strategy. c.t|jyrH)rr)rras rJrczRandomWorker.reorder_queuess$$%rLNrrIrLrJrrs &rLr)wrrloggingrrrmrWrr`rrrrrrenumrrtypesrtypingr r r r r rruuidrresourcer ImportErrorrr redis.clientrrrrrrredis.exceptionsrrcommandrrrdefaultsrrr r!r"r#rr$r%r& executionsr'groupr(r3r)r*r+logutilsr,r-r.r/rr1registryr2r3rr4 serializersr5 suspensionr6timeoutsr7r8r9utilsr:r;r<r=r>r?r@rArBrrCrDrKr getLoggerrrrrNrdirrdrhrjrrr|rr{rrrrIrLrJrs  22NNN *AA* !KKXW!&&<<:"+$__    8   ; ' I  7:6{  c4 3`$`$F#TZTnA6A.(g6(gVava&6&I7    *))*V  3 4  s6(F7?G"G7G?G GG G%$G%