How to use should_run_now method in localstack

Best Python code snippet using localstack_python

util.py

Source:util.py

...27    def __init__(self, name, fn, hour):28        self.name = name29        self.fn = fn30        self.hour = hour31    def should_run_now(self):32        now = self.now()33        return now.hour == self.hour34class WeekdayDailyTask(Task):35    def __init__(self, name, fn, hour):36        self.name = name37        self.fn = fn38        self.hour = hour39    def should_run_now(self):40        now = self.now()41        return now.hour == self.hour and now.weekday() < 542class HourlyTask(Task):43    def __init__(self, name, fn):44        self.name = name45        self.fn = fn46    def should_run_now(self):47        return True48class WeeklyTask(Task):49    def __init__(self, name, fn, hour, day_of_week):50        self.name = name51        self.fn = fn52        self.hour = hour53        self.day_of_week = day_of_week54    def should_run_now(self):55        now = self.now()56        return now.hour == self.hour and now.weekday() == self.day_of_week57def try_to_run_scheduler():58    with connection.cursor() as cursor:59        cursor.execute("select pg_advisory_lock(101)")60        try:61            now = timezone.now()62            if now.minute >= 15:63                return False64            if not has_scheduler_run_recently():65                update_last_run_date()66                run_tasks()67                return True68            else:69                return False70        finally:71            cursor.execute("select pg_advisory_unlock(101)")72def has_scheduler_run_recently():73    last_run_record = SchedulerLastRun.objects.first()74    if last_run_record:75        return last_run_record.last_run_date >= timezone.now() - timedelta(minutes=30)76    else:77        return False78def update_last_run_date():79    last_run_record, was_created = SchedulerLastRun.objects.get_or_create()80    last_run_record.last_run_date = timezone.now()81    last_run_record.save()82def run_tasks():83    tasks_to_run = []84    for task in ALL_TASKS:85        if task.should_run_now():86            tasks_to_run.append(task)87    for task in tasks_to_run:88        try:89            task.run()90            logger.info('task {} ran successfully'.format(task.name))91        except Exception as e:92            logger.error('error while running task {}:'.format(task.name))93            logger.exception(e)94def scheduler_loop():95    while True:96        now = timezone.now()97        if now.minute < 15:98            try_to_run_scheduler()99            time.sleep(1800)...

scheduler.py

Source:scheduler.py

...9        self.schedule = schedule10        self.job_id = short_uid()11    def run(self):12        try:13            if self.should_run_now():14                self.do_run()15        except Exception as e:16            LOG.debug('Unable to run scheduled function %s: %s' % (self.job_func, e))17    def should_run_now(self):18        schedule = CronTab(self.schedule)19        delay_secs = schedule.next()20        return delay_secs < 6021    def do_run(self):22        FuncThread(self.job_func).start()23class JobScheduler(object):24    _instance = None25    def __init__(self):26        self.jobs = []27        self.thread = None28    def add_job(self, job_func, schedule):29        job = Job(job_func, schedule)30        self.jobs.append(job)31        return job.job_id...

runner.py

Source:runner.py

...16            if module == '__init__.py' or module[-3:] != '.py':17                continue18            __import__(MODULE_BASE+module[:-3], locals(), globals())19        tasks = [subclass.instance() for subclass in BaseScheduledTask.__subclasses__()]20        tasks_to_run = [task for task in tasks if self.should_run_now(task)]21        return tasks_to_run22    def should_run_now(self, task):23        return task.get_state() in [TaskStateEnum.WAIT_NEXT, TaskStateEnum.FAILED] and\24               (task._get_db_task().next_run - datetime.utcnow()) < timedelta(seconds=TIME_GAP_IN_SECONDS)25    def _log(self, message):26        LOGGER.info(message)27if __name__ == '__main__':28    runner = ScheduledTaskRunner()29    runner._log("Start execution for scheduled tasks")30    tasks = runner.get_task_list(os.path.dirname(__file__))31    runner._log("%d tasks should be executed" % len(tasks))32    user = User.objects.get(email=FB_DATA_PULL_USER)33    for task in tasks:34        runner._log('Start executing tasks %s' % task.__name__)35        task.execute(user)36        runner._log('Task %s execution finished. Task state is %s' % (task.__name__, task.get_state()))

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.