def scheduler(broker=None):
    if not broker:
        broker = get_broker()
    db.close_old_connections()
    try:
        for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()):
            args = ()
            kwargs = {}
            # get args, kwargs and hook
            if s.kwargs:
                try:
                    # eval should be safe here because dict()
                    kwargs = eval('dict({})'.format(s.kwargs))
                except SyntaxError:
                    kwargs = {}
            if s.args:
                args = ast.literal_eval(s.args)
                # single value won't eval to tuple, so:
                if type(args) != tuple:
                    args = (args,)
            q_options = kwargs.get('q_options', {})
            if s.hook:
                q_options['hook'] = s.hook
            # set up the next run time
            if not s.schedule_type == s.ONCE:
                next_run = arrow.get(s.next_run)
                while True:
                    if s.schedule_type == s.MINUTES:
                        next_run = next_run.replace(minutes=+(s.minutes or 1))
                    elif s.schedule_type == s.HOURLY:
                        next_run = next_run.replace(hours=+1)
                    elif s.schedule_type == s.DAILY:
                        next_run = next_run.replace(days=+1)
                    elif s.schedule_type == s.WEEKLY:
                        next_run = next_run.replace(weeks=+1)
                    elif s.schedule_type == s.MONTHLY:
                        next_run = next_run.replace(months=+1)
                    elif s.schedule_type == s.QUARTERLY:
                        next_run = next_run.replace(months=+3)
                    elif s.schedule_type == s.YEARLY:
                        next_run = next_run.replace(years=+1)
                    if Conf.CATCH_UP or next_run > arrow.utcnow():
                        break
                s.next_run = next_run.datetime
                s.repeats += -1
            # send it to the cluster
            q_options['broker'] = broker
            q_options['group'] = q_options.get('group', s.name or s.id)
            kwargs['q_options'] = q_options
            s.task = tasks.async(s.func, *args, **kwargs)
            # log it
            if not s.task:
                logger.error(
                        _('{} failed to create a task from schedule [{}]').format(current_process().name,
                                                                                  s.name or s.id))
            else:
                logger.info(
                        _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
            # default behavior is to delete a ONCE schedule
            if s.schedule_type == s.ONCE:
                if s.repeats < 0:
                    s.delete()
                    continue
                # but not if it has a positive repeats
                s.repeats = 0
            # save the schedule
            s.save()
    except Exception as e:
        logger.error(e)
Creates a task from a schedule at the scheduled time and schedules next run
dataset/ETHPy150Open Koed00/django-q/django_q/cluster.py/scheduler