HEX

File: //home/arjun/projects/env/lib/python3.10/site-packages/celery/__pycache__/beat.cpython-310.pyc
o

-we�_�@sdZddlZddlZddlZddlZddlZddlZddlZddlZddl	m
Z
ddlmZddl
mZddlmZmZddlmZddlmZdd	lmZdd
lmZmZddlmZdd
lmZmZm Z ddl!m"Z"ddl#m$Z$m%Z%ddl&m'Z'ddl(m)Z)m*Z*ddl+m,Z,m-Z-ddl.m/Z/m0Z0dZ1edd�Z2e,e3�Z4e4j5e4j6e4j7e4j8f\Z5Z6Z7Z8dZ9Gdd�de:�Z;Gdd�d�Z<eGdd�d��Z=dd�Z>d d!�Z?Gd"d#�d#�Z@Gd$d%�d%e@�ZAGd&d'�d'�ZBGd(d)�d)e�ZCze�WneDy�dZEYn	wGd*d+�d+e�ZEd.d,d-�ZFdS)/zThe periodic task scheduler.�N)�timegm)�
namedtuple)�total_ordering)�Event�Thread)�ensure_multiprocessing)�
reset_signals)�Process)�maybe_evaluate�reprcall)�cached_property�)�__version__�	platforms�signals)�reraise)�crontab�maybe_schedule)�is_numeric_value)�load_extension_class_names�symbol_by_name)�
get_logger�iter_open_logger_fds)�humanize_seconds�maybe_make_aware)�SchedulingError�
ScheduleEntry�	Scheduler�PersistentScheduler�Service�EmbeddedService�event_t)�time�priority�entryi,c@seZdZdZdS)rz*An error occurred while scheduling a task.N)�__name__�
__module__�__qualname__�__doc__�r)r)�D/home/arjun/projects/env/lib/python3.10/site-packages/celery/beat.pyr,src@s(eZdZdZdd�Zdd�Zdd�ZdS)	�BeatLazyFuncaoA lazy function declared in 'beat_schedule' and called before sending to worker.

    Example:

        beat_schedule = {
            'test-every-5-minutes': {
                'task': 'test',
                'schedule': 300,
                'kwargs': {
                    "current": BeatCallBack(datetime.datetime.now)
                }
            }
        }

    cOs||_||d�|_dS)N)�args�kwargs��_func�_func_params)�self�funcr,r-r)r)r*�__init__As�zBeatLazyFunc.__init__cC�|��S�N)�delay�r1r)r)r*�__call__H�zBeatLazyFunc.__call__cCs|j|jdi|jd��S)Nr,r-r.r7r)r)r*r6KszBeatLazyFunc.delayN)r%r&r'r(r3r8r6r)r)r)r*r+0s
r+c@s�eZdZdZdZdZdZdZdZdZ	dZ
			ddd�Zdd	�ZeZ
dd
d�ZeZZdd
�Zdd�Zdd�Zdd�Zdd�Zdd�Zdd�Zdd�ZdS)ra�An entry in the scheduler.

    Arguments:
        name (str): see :attr:`name`.
        schedule (~celery.schedules.schedule): see :attr:`schedule`.
        args (Tuple): see :attr:`args`.
        kwargs (Dict): see :attr:`kwargs`.
        options (Dict): see :attr:`options`.
        last_run_at (~datetime.datetime): see :attr:`last_run_at`.
        total_run_count (int): see :attr:`total_run_count`.
        relative (bool): Is the time relative to when the server starts?
    Nrr)FcCsb|
|_||_||_||_|r|ni|_|r|ni|_t||	|jd�|_|p(|��|_	|p-d|_
dS)N)�appr)r:�name�taskr,r-�optionsr�schedule�default_now�last_run_at�total_run_count)r1r;r<r@rAr>r,r-r=�relativer:r)r)r*r3sszScheduleEntry.__init__cCs|jr|j��S|j��Sr5)r>�nowr:r7r)r)r*r?�szScheduleEntry.default_nowcCs(|jdit||p|��|jdd���S)z8Return new instance, with date and count fields updated.r
)r@rANr))�	__class__�dictr?rA)r1r@r)r)r*�_next_instance�s



�zScheduleEntry._next_instancec	Cs*|j|j|j|j|j|j|j|j|jffSr5)	rDr;r<r@rAr>r,r-r=r7r)r)r*�
__reduce__�s�zScheduleEntry.__reduce__cCs&|j�|j|j|j|j|jd��dS)z�Update values from another entry.

        Will only update "editable" fields:
            ``task``, ``schedule``, ``args``, ``kwargs``, ``options``.
        )r<r>r,r-r=N)�__dict__�updater<r>r,r-r=�r1�otherr)r)r*rI�s
�zScheduleEntry.updatecCs|j�|j�S)z.See :meth:`~celery.schedules.schedule.is_due`.)r>�is_duer@r7r)r)r*rL�szScheduleEntry.is_duecCstt|����Sr5)�iter�vars�itemsr7r)r)r*�__iter__�szScheduleEntry.__iter__cCs,dj|t|j|jp
d|jpi�t|�jd�S)Nz%<{name}: {0.name} {call} {0.schedule}r))�callr;)�formatrr<r,r-�typer%r7r)r)r*�__repr__�s
�zScheduleEntry.__repr__cCst|t�r
t|�t|�kStSr5)�
isinstancer�id�NotImplementedrJr)r)r*�__lt__�s
zScheduleEntry.__lt__cCs(dD]}t||�t||�krdSqdS)N)r<r,r-r=r>FT)�getattr)r1rK�attrr)r)r*�editable_fields_equal�s
�z#ScheduleEntry.editable_fields_equalcCs
|�|�S)z�Test schedule entries equality.

        Will only compare "editable" fields:
        ``task``, ``schedule``, ``args``, ``kwargs``, ``options``.
        )r[rJr)r)r*�__eq__�s
zScheduleEntry.__eq__)
NNNNNr)NNFNr5)r%r&r'r(r;r>r,r-r=r@rAr3r?�_default_nowrF�__next__�nextrGrIrLrPrTrXr[r\r)r)r)r*rOs2
�
rcCs|sgSdd�|D�S)NcSs g|]}t|t�r|�n|�qSr)�rUr+)�.0�vr)r)r*�
<listcomp>�s��z(_evaluate_entry_args.<locals>.<listcomp>r))�
entry_argsr)r)r*�_evaluate_entry_args�s
�recCs|siSdd�|��D�S)NcSs&i|]\}}|t|t�r|�n|�qSr)r`)ra�krbr)r)r*�
<dictcomp>�s��z*_evaluate_entry_kwargs.<locals>.<dictcomp>)rO)�entry_kwargsr)r)r*�_evaluate_entry_kwargs�s
�ric@sDeZdZdZeZdZeZdZ	dZ
dZdZe
Z
		d>dd�Zdd	�Zd?d
d�Zd@d
d�Zdd�Zefdd�Zeejfdd�Zeeejejfdd�Zdd�Zdd�Zdd�ZdAdd�Z d d!�Z!d"d#�Z"d$d%�Z#d&d'�Z$d(d)�Z%d*d+�Z&d,d-�Z'd.d/�Z(d0d1�Z)d2d3�Z*d4d5�Z+d6d7�Z,e-e+e,�Ze.d8d9��Z/e.d:d;��Z0e-d<d=��Z1dS)Bra�Scheduler for periodic tasks.

    The :program:`celery beat` program may instantiate this class
    multiple times for introspection purposes, but then with the
    ``lazy`` argument set.  It's important for subclasses to
    be idempotent when this argument is set.

    Arguments:
        schedule (~celery.schedules.schedule): see :attr:`schedule`.
        max_interval (int): see :attr:`max_interval`.
        lazy (bool): Don't set up the schedule.
    N�rFcKst||_t|dur
in|�|_|p|jjp|j|_|p|jj|_d|_d|_	|dur-|jj
n||_|s8|��dSdSr5)
r:r
�data�conf�beat_max_loop_interval�max_interval�amqp�Producer�_heap�old_schedulers�beat_sync_every�sync_every_tasks�setup_schedule)r1r:r>rnrp�lazyrtr-r)r)r*r3�s ����zScheduler.__init__cCsJi}|jjjr|jjjsd|vrdtddd�ddid�|d<|�|�dS)Nzcelery.backend_cleanup�0�4�*�expiresi��)r<r>r=)r:rl�result_expires�backend�supports_autoexpirer�update_from_dict)r1rk�entriesr)r)r*�install_default_entries
s
�

�z!Scheduler.install_default_entriesc
Cs�td|j|j�z
|j||dd�}Wnty/}ztd|t��dd�WYd}~dSd}~ww|rAt|d�rAt	d|j|j
�dSt	d	|j�dS)
Nz#Scheduler: Sending due task %s (%s)F)�producer�advancezMessage Error: %s
%sT��exc_inforVz%s sent. id->%sz%s sent.)�infor;r<�apply_async�	Exception�error�	traceback�format_stack�hasattr�debugrV)r1r$r��result�excr)r)r*�apply_entrys
���zScheduler.apply_entry�{�G�z��cCs|r
|dkr
||S|S)Nrr))r1�n�driftr)r)r*�adjust"szScheduler.adjustcCs|��Sr5)rL)r1r$r)r)r*rL'r9zScheduler.is_duecCs4|j}t|���}||���|jd||�pdS)z9Return a utc timestamp, make sure heapq in correct order.g��.Ar)r�rr?�utctimetuple�microsecond)r1r$�next_time_to_run�mktimer��as_nowr)r)r*�_when*s
�
�zScheduler._whencCs\d}g|_|j��D]}|��\}}|j�||�||rdn|�p!d||��q
||j�dS)z:Populate the heap with the data contained in the schedule.�rN)rqr>�valuesrL�appendr�)r1r!�heapifyr#r$rL�next_call_delayr)r)r*�
populate_heap4s
��zScheduler.populate_heapcCs�|j}|j}|jdus|�|j|j�st�|j�|_|��|j}|s%|S|d}|d}	|�|	�\}
}|
rh||�}||ur\|�	|	�}
|j
|	|jd�||||�|
|�|d|
��dS|||�||d|�S||�}|t
|�ru||�S||�S)z�Run a tick - one iteration of the scheduler.

        Executes one due task per call.

        Returns:
            float: preferred delay in seconds for next call.
        Nr�)r�r
)r�rnrq�schedules_equalrrr>�copyr�rL�reserver�r�r�r)r1r!�min�heappop�heappushr�rn�H�eventr$rLr��verify�
next_entry�adjusted_next_time_to_runr)r)r*�tickDs<	
�
�
��zScheduler.tickcCs�||urdurdS|dus|durdSt|���t|���kr$dS|��D]\}}|�|�}|s6dS||kr=dSq(dS)NTF)�set�keysrO�get)r1�
old_schedules�
new_schedulesr;�	old_entry�	new_entryr)r)r*r�ls�
�zScheduler.schedules_equalcCs.|jpt��|j|jkp|jo|j|jkSr5)�
_last_syncr"�	monotonic�
sync_everyrt�_tasks_since_syncr7r)r)r*�should_sync{s�
�zScheduler.should_synccCst|�}|j|j<|Sr5)r_r>r;)r1r$r�r)r)r*r��szScheduler.reserveTc	
KsN|r|�|�n|}|jj�|j�}z�zLt|j�}t|j�}|r>|j	||fd|i|j
��WW|jd7_|��r=|�
�SS|j|j||fd|i|j
��WW|jd7_|��r^|�
�SSty�}ztttdj||d��t��d�WYd}~nd}~wwW|jd7_|��r�|�
�dSdS|jd7_|��r�|�
�ww)Nr�r
z-Couldn't apply scheduled task {0.name}: {exc})r�r�)r�r:�tasksr�r<rer,rir-r�r=r�r��_do_sync�	send_taskr�rrrR�sysr�)	r1r$r�r�r-r<rdrhr�r)r)r*r��sV

��
����
����
�����
�zScheduler.apply_asynccOs|jj|i|��Sr5)r:r��r1r,r-r)r)r*r���zScheduler.send_taskcCs |�|j�|�|jjj�dSr5)r�rk�
merge_inplacer:rl�
beat_scheduler7r)r)r*ru�szScheduler.setup_schedulecCs:ztd�|��Wt��|_d|_dSt��|_d|_w)Nzbeat: Synchronizing schedule...r)r��syncr"r�r�r�r7r)r)r*r��s



�zScheduler._do_synccCsdSr5r)r7r)r)r*r��szScheduler.synccCs|��dSr5)r�r7r)r)r*�close�szScheduler.closecKs&|jdd|ji|��}||j|j<|S)Nr:r))�Entryr:r>r;)r1r-r$r)r)r*�add�sz
Scheduler.addcCs4t||j�r|j|_|S|jdit|||jd���S�N)r;r:r))rUr�r:rE)r1r;r$r)r)r*�_maybe_entry�szScheduler._maybe_entrycs"�j��fdd�|��D��dS)Ncsi|]\}}|��||��qSr))r�)rar;r$r7r)r*rg�s��z.Scheduler.update_from_dict.<locals>.<dictcomp>)r>rIrO)r1�dict_r)r7r*r~�s�zScheduler.update_from_dictc	Cs�|j}t|�t|�}}||AD]}|�|d�q|D]#}|jdit||||jd���}|�|�r:||�|�q|||<qdSr�)r>r��popr�rEr:r�rI)r1�br>�A�B�keyr$r)r)r*r��s 

�zScheduler.merge_inplacecCsdd�}|j�||jjj�S)NcSstd||�dS)Nz9beat: Connection error: %s. Trying again in %s seconds...)r�)r��intervalr)r)r*�_error_handler�s�z3Scheduler._ensure_connected.<locals>._error_handler)�
connection�ensure_connectionr:rl�broker_connection_max_retries)r1r�r)r)r*�_ensure_connected�s
�zScheduler._ensure_connectedcCs|jSr5�rkr7r)r)r*�get_schedule�szScheduler.get_schedulecCs
||_dSr5r��r1r>r)r)r*�set_schedule��
zScheduler.set_schedulecCs
|j��Sr5)r:�connection_for_writer7r)r)r*r��s
zScheduler.connectioncCs|j|��dd�S)NF)�auto_declare)rpr�r7r)r)r*r��szScheduler.producercCsdS)N�r)r7r)r)r*r��szScheduler.info)NNNFNr5)r�)NT)2r%r&r'r(rr�r>�DEFAULT_MAX_INTERVALrnr�rtr�r��loggerr3r�r�r�rLrr�r!�heapqr�r�r�r�r�r�r�r�r�r�r�rur�r�r�r�r�r~r�r�r�r��propertyrr�r�r�r)r)r)r*r�sZ
�




�(



rcs�eZdZdZeZdZdZ�fdd�Zdd�Z	dd	�Z
d
d�Zdd
�Zdd�Z
dd�Zdd�Zeee�Zdd�Zdd�Zedd��Z�ZS)rz+Scheduler backed by :mod:`shelve` database.)r�z.dbz.datz.bakz.dirNcs"|�d�|_t�j|i|��dS)N�schedule_filename)r�r��superr3r��rDr)r*r3�szPersistentScheduler.__init__c	CsL|jD] }t�tj��t�|j|�Wd�n1swYqdSr5)�known_suffixesr�ignore_errno�errno�ENOENT�os�remover�)r1�suffixr)r)r*�
_remove_dbs
���zPersistentScheduler._remove_dbcCs|jj|jdd�S)NT)�	writeback)�persistence�openr�r7r)r)r*�_open_scheduler�z"PersistentScheduler._open_schedulecCs"td|j|dd�|��|��S)Nz'Removing corrupted schedule file %r: %rTr�)r�r�r�r�)r1r�r)r)r*� _destroy_open_corrupted_schedules
�z4PersistentScheduler._destroy_open_corrupted_schedulec
CsFz|��|_|j��Wnty$}z|�|�|_WYd}~nd}~ww|��|jjj}|j�	d�}|durG||krGt
d||�|j��|jjj}|j�	d�}|durn||krnddd�}t
d||||�|j��|j�
di�}|�|jjj�|�|j�|j�t||d	��|��td
d�dd
�|��D���dS)N�tzz%Reset: Timezone changed from %r to %r�utc_enabled�enabled�disabled)TFz Reset: UTC changed from %s to %sr)rr�r�zCurrent schedule:
�
css�|]}t|�VqdSr5)�repr)rar$r)r)r*�	<genexpr>4s�
�z5PersistentScheduler.setup_schedule.<locals>.<genexpr>)r��_storer�r�r��_create_scheduler:rl�timezoner��warning�clear�
enable_utc�
setdefaultr�r�r�r>rIrr�r��joinr�)r1r�r��	stored_tz�utc�
stored_utc�choicesrr)r)r*rusB
��



�
��z"PersistentScheduler.setup_schedulecCs�dD]h}z|jdWn,ty7zi|jd<Wnty2}z
|�|�|_WYd}~Yqd}~wwYdSwd|jvrItd�|j��dSd|jvrZtd�|j��dSd|jvrhtd�|j��dSdS)	N)r
r�rrz+DB Reset: Account for new __version__ fieldr�z"DB Reset: Account for new tz fieldr�z+DB Reset: Account for new utc_enabled field)r��KeyErrorr�r�r�)r1�_r�r)r)r*r�7s6����


�

�
�z$PersistentScheduler._create_schedulecCs
|jdS�Nr�r�r7r)r)r*r�Nr�z PersistentScheduler.get_schedulecCs||jd<dSrrr�r)r)r*r�Qsz PersistentScheduler.set_schedulecCs|jdur|j��dSdSr5)r�r�r7r)r)r*r�Us
�zPersistentScheduler.synccCs|��|j��dSr5)r�r�r�r7r)r)r*r�YszPersistentScheduler.closecCsd|j��S)Nz    . db -> )r�r7r)r)r*r�]szPersistentScheduler.info)r%r&r'r(�shelver�r�r�r3r�r�r�rur�r�r�r�r>r�r�r��
__classcell__r)r)r�r*r�s$&
rc@s`eZdZdZeZ		ddd�Zdd�Zddd	�Zd
d�Z	ddd
�Z
		ddd�Zedd��Z
dS)rzCelery periodic task service.NcCsB||_|p|jj|_|p|j|_|p|jj|_t�|_t�|_	dSr5)
r:rlrmrn�
scheduler_cls�beat_schedule_filenamer�r�_is_shutdown�_is_stopped)r1r:rnr�rr)r)r*r3gs�
�zService.__init__cCs|j|j|j|j|jffSr5)rDrnr�rr:r7r)r)r*rGss�zService.__reduce__Fc	Cs�td�tdt|jj��tjj|d�|r"tjj|d�t	�
d�zNz/|j��sQ|j�
�}|rL|dkrLtdt|dd��t�|�|j��rL|j��|j��r)Wnttfyb|j��YnwW|��dSW|��dS|��w)	Nzbeat: Starting...z#beat: Ticking with max interval->%s)�senderzcelery beatgzbeat: Waking up %s.zin )�prefix)r�r�r�	schedulerrnr�	beat_init�send�beat_embedded_initr�set_process_titler
�is_setr�r"�sleepr�r��KeyboardInterrupt�
SystemExitr�r�)r1�embedded_processr�r)r)r*�startws6
�



�



�����z
Service.startcC�|j��|j��dSr5)rr�rr�r7r)r)r*r���
zService.synccCs*td�|j��|o|j��dSdS)Nzbeat: Shutting down...)r�r
r�r�wait)r1rr)r)r*�stop�s
zService.stop�celery.beat_schedulerscCs0|j}tt|��}t|j|d�|j||j|d�S)N)�aliases)r:r�rnrv)r�rErrrr:rn)r1rv�extension_namespace�filenamerr)r)r*�
get_scheduler�s�zService.get_schedulercCr4r5)r!r7r)r)r*r�szService.scheduler)NNN)F)Fr)r%r&r'r(rrr3rGrr�rr!rrr)r)r)r*rbs
�


�rcs0eZdZdZ�fdd�Zdd�Zdd�Z�ZS)�	_Threadedz(Embedded task scheduler using threading.cs2t���||_t|fi|��|_d|_d|_dS)NT�Beat)r�r3r:r�service�daemonr;�r1r:r-r�r)r*r3�s


z_Threaded.__init__cCrr5)r:�set_currentr$rr7r)r)r*�run�rz
_Threaded.runcCs|jjdd�dS)NT)r)r$rr7r)r)r*r�r�z_Threaded.stop)r%r&r'r(r3r(rrr)r)r�r*r"�s
r"cs,eZdZ�fdd�Zdd�Zdd�Z�ZS)�_Processcs,t���||_t|fi|��|_d|_dS)Nr#)r�r3r:rr$r;r&r�r)r*r3�s

z_Process.__init__cCsPtdd�t�tjtjtjgtt���|j	�
�|j	��|jj
dd�dS)NF)�fullT)r)rr�close_open_fdsr��	__stdin__�
__stdout__�
__stderr__�listrr:�set_defaultr'r$rr7r)r)r*r(�s
��

z_Process.runcCs|j��|��dSr5)r$r�	terminater7r)r)r*r�s
z
_Process.stop)r%r&r'r3r(rrr)r)r�r*r)�s	r)cKs<|�dd�s
tdurt|fddi|��St|fd|i|��S)z�Return embedded clock service.

    Arguments:
        thread (bool): Run threaded instead of as a separate process.
            Uses :mod:`multiprocessing` by default, if available.
    �threadFNrnr
)r�r)r")r:rnr-r)r)r*r �sr r5)Gr(r�r�r�r�rr�r"r��calendarr�collectionsr�	functoolsr�	threadingrr�billiardr�billiard.commonr�billiard.contextr	�kombu.utils.functionalr
r�kombu.utils.objectsrr�rrr�
exceptionsr�	schedulesrr�utils.functionalr�
utils.importsrr�	utils.logrr�
utils.timerr�__all__r!r%r�r�r�r�r�r�r�rr+rrerirrrr"�NotImplementedErrorr)r r)r)r)r*�<module>sf
�w		 kF
�