How to automatize scheduling with additional configuration

user_6402099 0
Viewed 301

Consider that I need to send automated report e-mails to my customers frequently. Since I may have more than one customer, and one customer may demand daily and/or weekly reports, I want to keep these in a json file as report_config.json and automate it with python schedule module.

What I have tried so far:

report_config.json is looks like this:

{
    "cusomer1": [
        {
            "report_type": "twitter",
            "report_frequency": "daily",
            "receivers": [
                "example@example.com"
            ]
        },
        {
            "report_type": "twitter",
            "report_frequency": "weekly",
            "receivers": [
                "example@example.com"
            ]
        }
    ],
    "customer2": {
        "report_type": "twitter",
        "report_frequency": "daily",
        "receivers": [
                "example@example.com"
            ]
    }
}

It's like I want to keep a customer's email demands in a list as its value in the config file. Then loop through and schedule them. i.e:

if __name__ == "__main__":
    for customer, reports in report_config.items():
        for report in reports:
            # below is an example of scheduling and not working as I want to achieve right now
            schedule.every().days.at('03:00').do(preprare_report,
                                                 report['report_type'], report['report_frequency'])
            schedule.every().monday.at('03:00').do(preprare_report,
                                                   report['report_type'], report['report_frequency'])

    while True:
        schedule.run_pending()
        time.sleep(1)

But as you can see, a customer may demand emails only in daily basis. Another one may want to every monday and every day as well. But I can not automate the schedule.every().monday/days part of the code, since they are methods and not arguments.

I know I can do what I want with several if-else statements. But as the customers and their demands grow, It will become unmaintainable with hardcoding. You know, this will be painful:

if report['day'] == 'monday':
    schedule.every().monday.at('03:00')...
elif report['day'] == 'tue':
    ...

And day isn't even the only one parameter here. So, any tips about how to achieve the fully-automated system? I can give any additional information you need. Thank you.

schedule python
2 Answers

With getattr function you can make some automation on getting job instance.

I inserted some type hinting for convenience.

from typing import Literal, Optional

import schedule

UnitType = Literal["weeks", "days", "hours", "minutes", "seconds",
                   "week", "day", "hour", "minute", "second",
                   "monday", "tuesday", "wednesday", "thursday", 
                   "friday", "saturday", "sunday"]


def get_schedule_job(
        unit: UnitType,
        interval: int = 1,
        time: Optional[str] = None
) -> schedule.Job:
    schedule_job: schedule.Job = getattr(schedule.every(interval), unit)

    if time is not None:
        return schedule_job.at(time)
    return schedule_job


report_schedule_config_example = {
    "interval": 1,
    "unit": "monday",
    "time": "03:00"
}

get_schedule_job(**report_schedule_config_example).do(prepare_report)
user_134665100

You can use Celery and Celery beat for this task.

“Celery is an asynchronous task queue/job queue based on distributed message passing. It is focused on real-time operation, but supports scheduling as well."

You can use celery beat periodic tasks to dynamically schedule your jobs/tasks based on intervals or crontab.

user_96289490
Related