<s> from __future__ import absolute_import <EOL> import logging <EOL> from multiprocessing . util import Finalize <EOL> from anyjson import loads , dumps <EOL> from celery import current_app <EOL> from celery import schedules <EOL> from celery . beat import Scheduler , ScheduleEntry <EOL> from celery . utils . encoding import safe_str , safe_repr <EOL> from celery . utils . log import get_logger <EOL> from celery . utils . timeutils import is_naive <EOL> from django . db import transaction <EOL> from django . core . exceptions import ObjectDoesNotExist <EOL> from . db import commit_on_success <EOL> from . models import ( PeriodicTask , PeriodicTasks , <EOL> CrontabSchedule , IntervalSchedule ) <EOL> from . utils import DATABASE_ERRORS , make_aware <EOL> from . compat import itervalues <EOL> DEFAULT_MAX_INTERVAL = <NUM_LIT:5> <EOL> ADD_ENTRY_ERROR = """<STR_LIT>""" <EOL> logger = get_logger ( __name__ ) <EOL> debug , info , error = logger . debug , logger . info , logger . error <EOL> class ModelEntry ( ScheduleEntry ) : <EOL> model_schedules = ( ( schedules . crontab , CrontabSchedule , '<STR_LIT>' ) , <EOL> ( schedules . schedule , IntervalSchedule , '<STR_LIT>' ) ) <EOL> save_fields = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self , model ) : <EOL> self . app = current_app . _get_current_object ( ) <EOL> self . name = model . name <EOL> self . task = model . task <EOL> try : <EOL> self . schedule = model . schedule <EOL> except model . DoesNotExist : <EOL> logger . error ( '<STR_LIT>' ) <EOL> logger . warning ( '<STR_LIT>' , self . name ) <EOL> self . _disable ( model ) <EOL> try : <EOL> self . args = loads ( model . args or '<STR_LIT>' ) <EOL> self . kwargs = loads ( model . kwargs or '<STR_LIT:{}>' ) <EOL> except ValueError : <EOL> logging . error ( '<STR_LIT>' , self . name , <EOL> exc_info = <NUM_LIT:1> ) <EOL> logging . warning ( '<STR_LIT>' , self . name ) <EOL> self . _disable ( model ) <EOL> self . options = { '<STR_LIT>' : model . queue , <EOL> '<STR_LIT>' : model . exchange , <EOL> '<STR_LIT>' : model . routing_key , <EOL> '<STR_LIT>' : model . expires } <EOL> self . total_run_count = model . total_run_count <EOL> self . model = model <EOL> if not model . last_run_at : <EOL> model . last_run_at = self . _default_now ( ) <EOL> orig = self . last_run_at = model . last_run_at <EOL> if not is_naive ( self . last_run_at ) : <EOL> self . last_run_at = self . last_run_at . replace ( tzinfo = None ) <EOL> assert orig . hour == self . last_run_at . hour <EOL> def _disable ( self , model ) : <EOL> model . no_changes = True <EOL> model . enabled = False <EOL> model . save ( ) <EOL> def is_due ( self ) : <EOL> if not self . model . enabled : <EOL> return False , <NUM_LIT> <EOL> return self . schedule . is_due ( self . last_run_at ) <EOL> def _default_now ( self ) : <EOL> return self . app . now ( ) <EOL> def __next__ ( self ) : <EOL> self . model . last_run_at = self . app . now ( ) <EOL> self . model . total_run_count += <NUM_LIT:1> <EOL> self . model . no_changes = True <EOL> return self . __class__ ( self . model ) <EOL> next = __next__ <EOL> def save ( self ) : <EOL> obj = self . model . _default_manager . get ( pk = self . model . pk ) <EOL> for field in self . save_fields : <EOL> setattr ( obj , field , getattr ( self . model , field ) ) <EOL> obj . last_run_at = make_aware ( obj . last_run_at ) <EOL> obj . save ( ) <EOL> @ classmethod <EOL> def to_model_schedule ( cls , schedule ) : <EOL> for schedule_type , model_type , model_field in cls . model_schedules : <EOL> schedule = schedules . maybe_schedule ( schedule ) <EOL> if isinstance ( schedule , schedule_type ) : <EOL> model_schedule = model_type . from_schedule ( schedule ) <EOL> model_schedule . save ( ) <EOL> return model_schedule , model_field <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( schedule ) ) <EOL> @ classmethod <EOL> def from_entry ( cls , name , skip_fields = ( '<STR_LIT>' , '<STR_LIT>' ) , ** entry ) : <EOL> options = entry . get ( '<STR_LIT>' ) or { } <EOL> fields = dict ( entry ) <EOL> for skip_field in skip_fields : <EOL> fields . pop ( skip_field , None ) <EOL> schedule = fields . pop ( '<STR_LIT>' ) <EOL> model_schedule , model_field = cls . to_model_schedule ( schedule ) <EOL> fields [ model_field ] = model_schedule <EOL> fields [ '<STR_LIT:args>' ] = dumps ( fields . get ( '<STR_LIT:args>' ) or [ ] ) <EOL> fields [ '<STR_LIT>' ] = dumps ( fields . get ( '<STR_LIT>' ) or { } ) <EOL> fields [ '<STR_LIT>' ] = options . get ( '<STR_LIT>' ) <EOL> fields [ '<STR_LIT>' ] = options . get ( '<STR_LIT>' ) <EOL> fields [ '<STR_LIT>' ] = options . get ( '<STR_LIT>' ) <EOL> return cls ( PeriodicTask . _default_manager . update_or_create ( <EOL> name = name , defaults = fields , <EOL> ) ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( <EOL> safe_str ( self . name ) , self . task , safe_repr ( self . args ) , <EOL> safe_repr ( self . kwargs ) , self . schedule , <EOL> ) <EOL> class DatabaseScheduler ( Scheduler ) : <EOL> Entry = ModelEntry <EOL> Model = PeriodicTask <EOL> Changes = PeriodicTasks <EOL> _schedule = None <EOL> _last_timestamp = None <EOL> _initial_read = False <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . _dirty = set ( ) <EOL> self . _finalize = Finalize ( self , self . sync , exitpriority = <NUM_LIT:5> ) <EOL> Scheduler . __init__ ( self , * args , ** kwargs ) <EOL> self . max_interval = ( <EOL> kwargs . get ( '<STR_LIT>' ) or <EOL> self . app . conf . CELERYBEAT_MAX_LOOP_INTERVAL or <EOL> DEFAULT_MAX_INTERVAL ) <EOL> def setup_schedule ( self ) : <EOL> self . install_default_entries ( self . schedule ) <EOL> self . update_from_dict ( self . app . conf . CELERYBEAT_SCHEDULE ) <EOL> def all_as_schedule ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> s = { } <EOL> for model in self . Model . objects . enabled ( ) : <EOL> try : <EOL> s [ model . name ] = self . Entry ( model ) <EOL> except ValueError : <EOL> pass <EOL> return s <EOL> def schedule_changed ( self ) : <EOL> try : <EOL> try : <EOL> transaction . commit ( ) <EOL> except transaction . TransactionManagementError : <EOL> pass <EOL> last , ts = self . _last_timestamp , self . Changes . last_change ( ) <EOL> except DATABASE_ERRORS as exc : <EOL> error ( '<STR_LIT>' , exc , exc_info = <NUM_LIT:1> ) <EOL> return False <EOL> try : <EOL> if ts and ts > ( last if last else ts ) : <EOL> return True <EOL> finally : <EOL> self . _last_timestamp = ts <EOL> return False <EOL> def reserve ( self , entry ) : <EOL> new_entry = Scheduler . reserve ( self , entry ) <EOL> self . _dirty . add ( new_entry . name ) <EOL> return new_entry <EOL> def sync ( self ) : <EOL> info ( '<STR_LIT>' ) <EOL> _tried = set ( ) <EOL> try : <EOL> with commit_on_success ( ) : <EOL> while self . _dirty : <EOL> try : <EOL> name = self . _dirty . pop ( ) <EOL> _tried . add ( name ) <EOL> self . schedule [ name ] . save ( ) <EOL> except ( KeyError , ObjectDoesNotExist ) : <EOL> pass <EOL> except DATABASE_ERRORS as exc : <EOL> self . _dirty |= _tried <EOL> error ( '<STR_LIT>' , exc , exc_info = <NUM_LIT:1> ) <EOL> def update_from_dict ( self , dict_ ) : <EOL> s = { } <EOL> for name , entry in dict_ . items ( ) : <EOL> try : <EOL> s [ name ] = self . Entry . from_entry ( name , ** entry ) <EOL> except Exception as exc : <EOL> error ( ADD_ENTRY_ERROR , name , exc , entry ) <EOL> self . schedule . update ( s ) <EOL> def install_default_entries ( self , data ) : <EOL> entries = { } <EOL> if self . app . conf . CELERY_TASK_RESULT_EXPIRES : <EOL> entries . setdefault ( <EOL> '<STR_LIT>' , { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : schedules . crontab ( '<STR_LIT:0>' , '<STR_LIT:4>' , '<STR_LIT:*>' ) , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:12> * <NUM_LIT> } , <EOL> } , <EOL> ) <EOL> self . update_from_dict ( entries ) <EOL> @ property <EOL> def schedule ( self ) : <EOL> update = False <EOL> if not self . _initial_read : <EOL> debug ( '<STR_LIT>' ) <EOL> update = True <EOL> self . _initial_read = True <EOL> elif self . schedule_changed ( ) : <EOL> info ( '<STR_LIT>' ) <EOL> update = True <EOL> if update : <EOL> self . sync ( ) <EOL> self . _schedule = self . all_as_schedule ( ) <EOL> if logger . isEnabledFor ( logging . DEBUG ) : <EOL> debug ( '<STR_LIT>' , '<STR_LIT:\n>' . join ( <EOL> repr ( entry ) for entry in itervalues ( self . _schedule ) ) , <EOL> ) <EOL> return self . _schedule </s>
