#NEWLINE# def scheduler( broker = None ) : #NEWLINE# #INDENT# '\n    Creates a task from a schedule at the scheduled time and schedules next run\n    ' #NEWLINE# if ( not broker ) : #NEWLINE# #INDENT# broker = get_broker ( ) #NEWLINE# #UNINDENT# db . close_old_connections ( ) #NEWLINE# try : #NEWLINE# #INDENT# for s in Schedule . objects . exclude ( repeats = 0 ) . filter ( next_run__lt = timezone . now ( ) ) : #NEWLINE# #INDENT# args = ( ) #NEWLINE# kwargs = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# if s . kwargs : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# kwargs = eval ( 'dict({})' . format ( s . kwargs ) ) #NEWLINE# #UNINDENT# except SyntaxError : #NEWLINE# #INDENT# kwargs = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# #UNINDENT# #UNINDENT# if s . args : #NEWLINE# #INDENT# args = ast . literal_eval ( s . args ) #NEWLINE# if ( type ( args ) != tuple ) : #NEWLINE# #INDENT# args = ( args , ) #NEWLINE# #UNINDENT# #UNINDENT# q_options = kwargs . get ( 'q_options' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# if s . hook : #NEWLINE# #INDENT# q_options [ 'hook' ] = s . hook #NEWLINE# #UNINDENT# if ( not ( s . schedule_type == s . ONCE ) ) : #NEWLINE# #INDENT# next_run = arrow . get ( s . next_run ) #NEWLINE# while True : #NEWLINE# #INDENT# if ( s . schedule_type == s . MINUTES ) : #NEWLINE# #INDENT# next_run = next_run . replace ( minutes = ( + ( s . minutes or 1 ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . HOURLY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( hours = ( + 1 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . DAILY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( days = ( + 1 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . WEEKLY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( weeks = ( + 1 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . MONTHLY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( months = ( + 1 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . QUARTERLY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( months = ( + 3 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( s . schedule_type == s . YEARLY ) : #NEWLINE# #INDENT# next_run = next_run . replace ( years = ( + 1 ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# if ( Conf . CATCH_UP or ( next_run > arrow . utcnow ( ) ) ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# s . next_run = next_run . datetime #NEWLINE# s . repeats += ( - 1 ) #NEWLINE# #UNINDENT# q_options [ 'broker' ] = broker #NEWLINE# q_options [ 'group' ] = q_options . get ( 'group' , ( s . name or broker . id ) ) #NEWLINE# kwargs [ 'q_options' ] = q_options #NEWLINE# s . task = tasks . async ( s . func , * args , ** kwargs ) #NEWLINE# if ( not s . task ) : #NEWLINE# #INDENT# logger . error ( _ ( '{} failed to create a task from schedule [{}]' ) . format ( current_process ( ) . name , ( s . name or s . id ) ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# logger . info ( _ ( '{} created a task from schedule [{}]' ) . format ( current_process ( ) . name , ( s . name or s . id ) ) ) #NEWLINE# #UNINDENT# if ( s . schedule_type == s . ONCE ) : #NEWLINE# #INDENT# if ( s . repeats < 0 ) : #NEWLINE# #INDENT# s . delete ( ) #NEWLINE# continue #NEWLINE# #UNINDENT# s . repeats = 0 #NEWLINE# #UNINDENT# s . save ( ) #NEWLINE# #UNINDENT# #UNINDENT# except Exception as e : #NEWLINE# #INDENT# logger . error ( e )