"""Schedules define the intervals at which periodic tasks run."""from__future__importannotationsimportrefrombisectimportbisect,bisect_leftfromcollectionsimportnamedtuplefromdatetimeimportdatetime,timedelta,tzinfofromtypingimportAny,Callable,Iterable,Mapping,Sequence,Unionfromkombu.utils.objectsimportcached_propertyfromceleryimportCeleryfrom.importcurrent_appfrom.utils.collectionsimportAttributeDictfrom.utils.timeimport(ffwd,humanize_seconds,localize,maybe_make_aware,maybe_timedelta,remaining,timezone,weekday,yearmonth)__all__=('ParseException','schedule','crontab','crontab_parser','maybe_schedule','solar',)schedstate=namedtuple('schedstate',('is_due','next'))CRON_PATTERN_INVALID="""\Invalid crontab pattern. Valid range is {min}-{max}. \'{value}' was found.\"""CRON_INVALID_TYPE="""\Argument cronspec needs to be of any of the following types: \int, str, or an iterable type. {type!r} was given.\"""CRON_REPR="""\<crontab: {0._orig_minute}{0._orig_hour}{0._orig_day_of_month}{0._orig_month_of_year}\{0._orig_day_of_week} (m/h/dM/MY/d)>\"""SOLAR_INVALID_LATITUDE="""\Argument latitude {lat} is invalid, must be between -90 and 90.\"""SOLAR_INVALID_LONGITUDE="""\Argument longitude {lon} is invalid, must be between -180 and 180.\"""SOLAR_INVALID_EVENT="""\Argument event "{event}" is invalid, must be one of {all_events}.\"""Cronspec=Union[int,str,Iterable[int]]defcronfield(s:Cronspec|None)->Cronspec:return'*'ifsisNoneelses
[文档]classParseException(Exception):"""Raised by :class:`crontab_parser` when the input can't be parsed."""
[文档]classschedule(BaseSchedule):"""Schedule for periodic task. Arguments: run_every (float, ~datetime.timedelta): Time interval. relative (bool): If set to True the run time will be rounded to the resolution of the interval. nowfun (Callable): Function returning the current date and time (:class:`~datetime.datetime`). app (Celery): Celery app instance. """relative:bool=Falsedef__init__(self,run_every:float|timedelta|None=None,relative:bool=False,nowfun:Callable|None=None,app:Celery|None=None)->None:self.run_every=maybe_timedelta(run_every)self.relative=relativesuper().__init__(nowfun=nowfun,app=app)
[文档]defis_due(self,last_run_at:datetime)->tuple[bool,datetime]:"""Return tuple of ``(is_due, next_time_to_check)``. Notes: - next time to check is in seconds. - ``(True, 20)``, means the task should be run now, and the next time to check is in 20 seconds. - ``(False, 12.3)``, means the task is not due, but that the scheduler should check again in 12.3 seconds. The next time to check is used to save energy/CPU cycles, it does not need to be accurate but will influence the precision of your schedule. You must also keep in mind the value of :setting:`beat_max_loop_interval`, that decides the maximum number of seconds the scheduler can sleep between re-checking the periodic task intervals. So if you have a task that changes schedule at run-time then your next_run_at check will decide how long it will take before a change to the schedule takes effect. The max loop interval takes precedence over the next check at value returned. .. admonition:: Scheduler max interval variance The default max loop interval may vary for different schedulers. For the default scheduler the value is 5 minutes, but for example the :pypi:`django-celery-beat` database scheduler the value is 5 seconds. """last_run_at=self.maybe_make_aware(last_run_at)rem_delta=self.remaining_estimate(last_run_at)remaining_s=max(rem_delta.total_seconds(),0)ifremaining_s==0:returnschedstate(is_due=True,next=self.seconds)returnschedstate(is_due=False,next=remaining_s)
[文档]classcrontab_parser:"""Parser for Crontab expressions. Any expression of the form 'groups' (see BNF grammar below) is accepted and expanded to a set of numbers. These numbers represent the units of time that the Crontab needs to run on: .. code-block:: bnf digit :: '0'..'9' dow :: 'a'..'z' number :: digit+ | dow+ steps :: number range :: number ( '-' number ) ? numspec :: '*' | range expr :: numspec ( '/' steps ) ? groups :: expr ( ',' expr ) * The parser is a general purpose one, useful for parsing hours, minutes and day of week expressions. Example usage: .. code-block:: pycon >>> minutes = crontab_parser(60).parse('*/15') [0, 15, 30, 45] >>> hours = crontab_parser(24).parse('*/4') [0, 4, 8, 12, 16, 20] >>> day_of_week = crontab_parser(7).parse('*') [0, 1, 2, 3, 4, 5, 6] It can also parse day of month and month of year expressions if initialized with a minimum of 1. Example usage: .. code-block:: pycon >>> days_of_month = crontab_parser(31, 1).parse('*/3') [1, 4, 7, 10, 13, 16, 19, 22, 25, 28, 31] >>> months_of_year = crontab_parser(12, 1).parse('*/2') [1, 3, 5, 7, 9, 11] >>> months_of_year = crontab_parser(12, 1).parse('2-12/2') [2, 4, 6, 8, 10, 12] The maximum possible expanded value returned is found by the formula: :math:`max_ + min_ - 1` """ParseException=ParseException_range=r'(\w+?)-(\w+)'_steps=r'/(\w+)?'_star=r'\*'def__init__(self,max_:int=60,min_:int=0):self.max_=max_self.min_=min_self.pats:tuple[tuple[re.Pattern,Callable],...]=((re.compile(self._range+self._steps),self._range_steps),(re.compile(self._range),self._expand_range),(re.compile(self._star+self._steps),self._star_steps),(re.compile('^'+self._star+'$'),self._expand_star),)
def_parse_part(self,part:str)->list[int]:forregex,handlerinself.pats:m=regex.match(part)ifm:returnhandler(m.groups())returnself._expand_range((part,))def_expand_range(self,toks:Sequence[str])->list[int]:fr=self._expand_number(toks[0])iflen(toks)>1:to=self._expand_number(toks[1])ifto<fr:# Wrap around max_ if necessaryreturn(list(range(fr,self.min_+self.max_))+list(range(self.min_,to+1)))returnlist(range(fr,to+1))return[fr]def_range_steps(self,toks:Sequence[str])->list[int]:iflen(toks)!=3ornottoks[2]:raiseself.ParseException('empty filter')returnself._expand_range(toks[:2])[::int(toks[2])]def_star_steps(self,toks:Sequence[str])->list[int]:ifnottoksornottoks[0]:raiseself.ParseException('empty filter')returnself._expand_star()[::int(toks[0])]def_expand_star(self,*args:Any)->list[int]:returnlist(range(self.min_,self.max_+self.min_))def_expand_number(self,s:str)->int:ifisinstance(s,str)ands[0]=='-':raiseself.ParseException('negative numbers not supported')try:i=int(s)exceptValueError:try:i=yearmonth(s)exceptKeyError:try:i=weekday(s)exceptKeyError:raiseValueError(f'Invalid weekday literal {s!r}.')max_val=self.min_+self.max_-1ifi>max_val:raiseValueError(f'Invalid end range: {i} > {max_val}.')ifi<self.min_:raiseValueError(f'Invalid beginning range: {i} < {self.min_}.')returni
[文档]classcrontab(BaseSchedule):"""Crontab schedule. A Crontab can be used as the ``run_every`` value of a periodic task entry to add :manpage:`crontab(5)`-like scheduling. Like a :manpage:`cron(5)`-job, you can specify units of time of when you'd like the task to execute. It's a reasonably complete implementation of :command:`cron`'s features, so it should provide a fair degree of scheduling needs. You can specify a minute, an hour, a day of the week, a day of the month, and/or a month in the year in any of the following formats: .. attribute:: minute - A (list of) integers from 0-59 that represent the minutes of an hour of when execution should occur; or - A string representing a Crontab pattern. This may get pretty advanced, like ``minute='*/15'`` (for every quarter) or ``minute='1,13,30-45,50-59/2'``. .. attribute:: hour - A (list of) integers from 0-23 that represent the hours of a day of when execution should occur; or - A string representing a Crontab pattern. This may get pretty advanced, like ``hour='*/3'`` (for every three hours) or ``hour='0,8-17/2'`` (at midnight, and every two hours during office hours). .. attribute:: day_of_week - A (list of) integers from 0-6, where Sunday = 0 and Saturday = 6, that represent the days of a week that execution should occur. - A string representing a Crontab pattern. This may get pretty advanced, like ``day_of_week='mon-fri'`` (for weekdays only). (Beware that ``day_of_week='*/2'`` does not literally mean 'every two days', but 'every day that is divisible by two'!) .. attribute:: day_of_month - A (list of) integers from 1-31 that represents the days of the month that execution should occur. - A string representing a Crontab pattern. This may get pretty advanced, such as ``day_of_month='2-30/2'`` (for every even numbered day) or ``day_of_month='1-7,15-21'`` (for the first and third weeks of the month). .. attribute:: month_of_year - A (list of) integers from 1-12 that represents the months of the year during which execution can occur. - A string representing a Crontab pattern. This may get pretty advanced, such as ``month_of_year='*/3'`` (for the first month of every quarter) or ``month_of_year='2-12/2'`` (for every even numbered month). .. attribute:: nowfun Function returning the current date and time (:class:`~datetime.datetime`). .. attribute:: app The Celery app instance. It's important to realize that any day on which execution should occur must be represented by entries in all three of the day and month attributes. For example, if ``day_of_week`` is 0 and ``day_of_month`` is every seventh day, only months that begin on Sunday and are also in the ``month_of_year`` attribute will have execution events. Or, ``day_of_week`` is 1 and ``day_of_month`` is '1-7,15-21' means every first and third Monday of every month present in ``month_of_year``. """def__init__(self,minute:Cronspec='*',hour:Cronspec='*',day_of_week:Cronspec='*',day_of_month:Cronspec='*',month_of_year:Cronspec='*',**kwargs:Any)->None:self._orig_minute=cronfield(minute)self._orig_hour=cronfield(hour)self._orig_day_of_week=cronfield(day_of_week)self._orig_day_of_month=cronfield(day_of_month)self._orig_month_of_year=cronfield(month_of_year)self._orig_kwargs=kwargsself.hour=self._expand_cronspec(hour,24)self.minute=self._expand_cronspec(minute,60)self.day_of_week=self._expand_cronspec(day_of_week,7)self.day_of_month=self._expand_cronspec(day_of_month,31,1)self.month_of_year=self._expand_cronspec(month_of_year,12,1)super().__init__(**kwargs)
[文档]@classmethoddeffrom_string(cls,crontab:str)->crontab:""" Create a Crontab from a cron expression string. For example ``crontab.from_string('* * * * *')``. .. code-block:: text ┌───────────── minute (0–59) │ ┌───────────── hour (0–23) │ │ ┌───────────── day of the month (1–31) │ │ │ ┌───────────── month (1–12) │ │ │ │ ┌───────────── day of the week (0–6) (Sunday to Saturday) * * * * * """minute,hour,day_of_month,month_of_year,day_of_week=crontab.split(" ")returncls(minute,hour,day_of_week,day_of_month,month_of_year)
@staticmethoddef_expand_cronspec(cronspec:Cronspec,max_:int,min_:int=0)->set[Any]:"""Expand cron specification. Takes the given cronspec argument in one of the forms: .. code-block:: text int (like 7) str (like '3-5,*/15', '*', or 'monday') set (like {0,15,30,45} list (like [8-17]) And convert it to an (expanded) set representing all time unit values on which the Crontab triggers. Only in case of the base type being :class:`str`, parsing occurs. (It's fast and happens only once for each Crontab instance, so there's no significant performance overhead involved.) For the other base types, merely Python type conversions happen. The argument ``max_`` is needed to determine the expansion of ``*`` and ranges. The argument ``min_`` is needed to determine the expansion of ``*`` and ranges for 1-based cronspecs, such as day of month or month of year. The default is sufficient for minute, hour, and day of week. """ifisinstance(cronspec,int):result={cronspec}elifisinstance(cronspec,str):result=crontab_parser(max_,min_).parse(cronspec)elifisinstance(cronspec,set):result=cronspecelifisinstance(cronspec,Iterable):result=set(cronspec)# type: ignoreelse:raiseTypeError(CRON_INVALID_TYPE.format(type=type(cronspec)))# assure the result does not precede the min or exceed the maxfornumberinresult:ifnumber>=max_+min_ornumber<min_:raiseValueError(CRON_PATTERN_INVALID.format(min=min_,max=max_-1+min_,value=number))returnresultdef_delta_to_next(self,last_run_at:datetime,next_hour:int,next_minute:int)->ffwd:"""Find next delta. Takes a :class:`~datetime.datetime` of last run, next minute and hour, and returns a :class:`~celery.utils.time.ffwd` for the next scheduled day and time. Only called when ``day_of_month`` and/or ``month_of_year`` cronspec is specified to further limit scheduled task execution. """datedata=AttributeDict(year=last_run_at.year)days_of_month=sorted(self.day_of_month)months_of_year=sorted(self.month_of_year)defday_out_of_range(year:int,month:int,day:int)->bool:try:datetime(year=year,month=month,day=day)exceptValueError:returnTruereturnFalsedefis_before_last_run(year:int,month:int,day:int)->bool:returnself.maybe_make_aware(datetime(year,month,day,next_hour,next_minute),naive_as_utc=False)<last_run_atdefroll_over()->None:for_inrange(2000):flag=(datedata.dom==len(days_of_month)orday_out_of_range(datedata.year,months_of_year[datedata.moy],days_of_month[datedata.dom])or(is_before_last_run(datedata.year,months_of_year[datedata.moy],days_of_month[datedata.dom])))ifflag:datedata.dom=0datedata.moy+=1ifdatedata.moy==len(months_of_year):datedata.moy=0datedata.year+=1else:breakelse:# Tried 2000 times, we're most likely in an infinite loopraiseRuntimeError('unable to rollover, ''time specification is probably invalid')iflast_run_at.monthinself.month_of_year:datedata.dom=bisect(days_of_month,last_run_at.day)datedata.moy=bisect_left(months_of_year,last_run_at.month)else:datedata.dom=0datedata.moy=bisect(months_of_year,last_run_at.month)ifdatedata.moy==len(months_of_year):datedata.moy=0roll_over()while1:th=datetime(year=datedata.year,month=months_of_year[datedata.moy],day=days_of_month[datedata.dom])ifth.isoweekday()%7inself.day_of_week:breakdatedata.dom+=1roll_over()returnffwd(year=datedata.year,month=months_of_year[datedata.moy],day=days_of_month[datedata.dom],hour=next_hour,minute=next_minute,second=0,microsecond=0)def__repr__(self)->str:returnCRON_REPR.format(self)def__reduce__(self)->tuple[type,tuple[Cronspec,Cronspec,Cronspec,Cronspec,Cronspec],Any]:return(self.__class__,(self._orig_minute,self._orig_hour,self._orig_day_of_week,self._orig_day_of_month,self._orig_month_of_year),self._orig_kwargs)def__setstate__(self,state:Mapping[str,Any])->None:# Calling super's init because the kwargs aren't necessarily passed in# the same form as they are stored by the superclasssuper().__init__(**state)
[文档]defremaining_delta(self,last_run_at:datetime,tz:tzinfo|None=None,ffwd:type=ffwd)->tuple[datetime,Any,datetime]:# caching global ffwdlast_run_at=self.maybe_make_aware(last_run_at)now=self.maybe_make_aware(self.now())dow_num=last_run_at.isoweekday()%7# Sunday is day 0, not day 7execute_this_date=(last_run_at.monthinself.month_of_yearandlast_run_at.dayinself.day_of_monthanddow_numinself.day_of_week)execute_this_hour=(execute_this_dateandlast_run_at.day==now.dayandlast_run_at.month==now.monthandlast_run_at.year==now.yearandlast_run_at.hourinself.hourandlast_run_at.minute<max(self.minute))ifexecute_this_hour:next_minute=min(minuteforminuteinself.minuteifminute>last_run_at.minute)delta=ffwd(minute=next_minute,second=0,microsecond=0)else:next_minute=min(self.minute)execute_today=(execute_this_dateandlast_run_at.hour<max(self.hour))ifexecute_today:next_hour=min(hourforhourinself.hourifhour>last_run_at.hour)delta=ffwd(hour=next_hour,minute=next_minute,second=0,microsecond=0)else:next_hour=min(self.hour)all_dom_moy=(self._orig_day_of_month=='*'andself._orig_month_of_year=='*')ifall_dom_moy:next_day=min([dayfordayinself.day_of_weekifday>dow_num]orself.day_of_week)add_week=next_day==dow_numdelta=ffwd(weeks=add_weekand1or0,weekday=(next_day-1)%7,hour=next_hour,minute=next_minute,second=0,microsecond=0,)else:delta=self._delta_to_next(last_run_at,next_hour,next_minute)returnself.to_local(last_run_at),delta,self.to_local(now)
[文档]defremaining_estimate(self,last_run_at:datetime,ffwd:type=ffwd)->timedelta:"""Estimate of next run time. Returns when the periodic task should run next as a :class:`~datetime.timedelta`. """# pylint: disable=redefined-outer-name# caching global ffwdreturnremaining(*self.remaining_delta(last_run_at,ffwd=ffwd))
[文档]defis_due(self,last_run_at:datetime)->tuple[bool,datetime]:"""Return tuple of ``(is_due, next_time_to_run)``. If :setting:`beat_cron_starting_deadline` has been specified, the scheduler will make sure that the `last_run_at` time is within the deadline. This prevents tasks that could have been run according to the crontab, but didn't, from running again unexpectedly. Note: Next time to run is in seconds. SeeAlso: :meth:`celery.schedules.schedule.is_due` for more information. """rem_delta=self.remaining_estimate(last_run_at)rem_secs=rem_delta.total_seconds()rem=max(rem_secs,0)due=rem==0deadline_secs=self.app.conf.beat_cron_starting_deadlinehas_passed_deadline=Falseifdeadline_secsisnotNone:# Make sure we're looking at the latest possible feasible run# date when checking the deadline.last_date_checked=last_run_atlast_feasible_rem_secs=rem_secswhilerem_secs<0:last_date_checked=last_date_checked+abs(rem_delta)rem_delta=self.remaining_estimate(last_date_checked)rem_secs=rem_delta.total_seconds()ifrem_secs<0:last_feasible_rem_secs=rem_secs# if rem_secs becomes 0 or positive, second-to-last# last_date_checked must be the last feasible run date.# Check if the last feasible date is within the deadline# for runninghas_passed_deadline=-last_feasible_rem_secs>deadline_secsifhas_passed_deadline:# Should not be due if we've passed the deadline for looking# at past runsdue=Falseifdueorhas_passed_deadline:rem_delta=self.remaining_estimate(self.now())rem=max(rem_delta.total_seconds(),0)returnschedstate(due,rem)
[文档]defmaybe_schedule(s:int|float|timedelta|BaseSchedule,relative:bool=False,app:Celery|None=None)->float|timedelta|BaseSchedule:"""Return schedule from number, timedelta, or actual schedule."""ifsisnotNone:ifisinstance(s,(float,int)):s=timedelta(seconds=s)ifisinstance(s,timedelta):returnschedule(s,relative,app=app)else:s.app=appreturns
[文档]classsolar(BaseSchedule):"""Solar event. A solar event can be used as the ``run_every`` value of a periodic task entry to schedule based on certain solar events. Notes: Available event values are: - ``dawn_astronomical`` - ``dawn_nautical`` - ``dawn_civil`` - ``sunrise`` - ``solar_noon`` - ``sunset`` - ``dusk_civil`` - ``dusk_nautical`` - ``dusk_astronomical`` Arguments: event (str): Solar event that triggers this task. See note for available values. lat (float): The latitude of the observer. lon (float): The longitude of the observer. nowfun (Callable): Function returning the current date and time as a class:`~datetime.datetime`. app (Celery): Celery app instance. """_all_events={'dawn_astronomical','dawn_nautical','dawn_civil','sunrise','solar_noon','sunset','dusk_civil','dusk_nautical','dusk_astronomical',}_horizons={'dawn_astronomical':'-18','dawn_nautical':'-12','dawn_civil':'-6','sunrise':'-0:34','solar_noon':'0','sunset':'-0:34','dusk_civil':'-6','dusk_nautical':'-12','dusk_astronomical':'18',}_methods={'dawn_astronomical':'next_rising','dawn_nautical':'next_rising','dawn_civil':'next_rising','sunrise':'next_rising','solar_noon':'next_transit','sunset':'next_setting','dusk_civil':'next_setting','dusk_nautical':'next_setting','dusk_astronomical':'next_setting',}_use_center_l={'dawn_astronomical':True,'dawn_nautical':True,'dawn_civil':True,'sunrise':False,'solar_noon':False,'sunset':False,'dusk_civil':True,'dusk_nautical':True,'dusk_astronomical':True,}def__init__(self,event:str,lat:int|float,lon:int|float,**kwargs:Any)->None:self.ephem=__import__('ephem')self.event=eventself.lat=latself.lon=lonsuper().__init__(**kwargs)ifeventnotinself._all_events:raiseValueError(SOLAR_INVALID_EVENT.format(event=event,all_events=', '.join(sorted(self._all_events)),))iflat<-90orlat>90:raiseValueError(SOLAR_INVALID_LATITUDE.format(lat=lat))iflon<-180orlon>180:raiseValueError(SOLAR_INVALID_LONGITUDE.format(lon=lon))cal=self.ephem.Observer()cal.lat=str(lat)cal.lon=str(lon)cal.elev=0cal.horizon=self._horizons[event]cal.pressure=0self.cal=calself.method=self._methods[event]self.use_center=self._use_center_l[event]def__reduce__(self)->tuple[type,tuple[str,int|float,int|float]]:returnself.__class__,(self.event,self.lat,self.lon)def__repr__(self)->str:return'<solar: {} at latitude {}, longitude: {}>'.format(self.event,self.lat,self.lon,)
[文档]defremaining_estimate(self,last_run_at:datetime)->timedelta:"""Return estimate of next time to run. Returns: ~datetime.timedelta: when the periodic task should run next, or if it shouldn't run today (e.g., the sun does not rise today), returns the time when the next check should take place. """last_run_at=self.maybe_make_aware(last_run_at)last_run_at_utc=localize(last_run_at,timezone.utc)self.cal.date=last_run_at_utctry:ifself.use_center:next_utc=getattr(self.cal,self.method)(self.ephem.Sun(),start=last_run_at_utc,use_center=self.use_center)else:next_utc=getattr(self.cal,self.method)(self.ephem.Sun(),start=last_run_at_utc)exceptself.ephem.CircumpolarError:# pragma: no cover# Sun won't rise/set today. Check again tomorrow# (specifically, after the next anti-transit).next_utc=(self.cal.next_antitransit(self.ephem.Sun())+timedelta(minutes=1))next=self.maybe_make_aware(next_utc.datetime())now=self.maybe_make_aware(self.now())delta=next-nowreturndelta
[文档]defis_due(self,last_run_at:datetime)->tuple[bool,datetime]:"""Return tuple of ``(is_due, next_time_to_run)``. Note: next time to run is in seconds. See Also: :meth:`celery.schedules.schedule.is_due` for more information. """rem_delta=self.remaining_estimate(last_run_at)rem=max(rem_delta.total_seconds(),0)due=rem==0ifdue:rem_delta=self.remaining_estimate(self.now())rem=max(rem_delta.total_seconds(),0)returnschedstate(due,rem)