Skip to content

Scheduling

Definition of the decorator that schedules tasks in GSO that are to run periodically.

CronScheduleConfig

Bases: BaseModel

Configuration for scheduling a task using crontab-like timing parameters.

Source code in gso/schedules/scheduling.py 
13
14
15
16
17
18
19
20
21
class CronScheduleConfig(BaseModel):
    """Configuration for scheduling a task using crontab-like timing parameters."""

    name: str
    minute: str = "*"
    hour: str = "*"
    day_of_week: str = "*"
    day_of_month: str = "*"
    month_of_year: str = "*"

scheduler(cron_scheduler_config)

Schedule a Celery task using crontab-like timing.

Examples:

  • minute='*/15': Run every 15 minutes.
  • hour='*/3': Run every 3 hours.
  • day_of_week='mon-fri': Run on weekdays only.
  • day_of_month='1-7,15-21': Run on the first and third weeks of the month.
  • month_of_year='*/3': Run on the first month of each quarter.

All time units can be specified with lists of numbers or crontab pattern strings for advanced scheduling. All specified time parts (minute, hour, day, etc.) must align for a task to run.

Source code in gso/schedules/scheduling.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def scheduler(cron_scheduler_config: CronScheduleConfig) -> Callable[[Callable], Callable]:
    """Schedule a Celery task using crontab-like timing.

    Examples:
        - `minute='*/15'`: Run every 15 minutes.
        - `hour='*/3'`: Run every 3 hours.
        - `day_of_week='mon-fri'`: Run on weekdays only.
        - `day_of_month='1-7,15-21'`: Run on the first and third weeks of the month.
        - `month_of_year='*/3'`: Run on the first month of each quarter.

    All time units can be specified with lists of numbers or crontab pattern strings for advanced scheduling.
    All specified time parts (minute, hour, day, etc.) must align for a task to run.
    """

    def decorator(task_func: Callable) -> Callable:
        @wraps(task_func)
        def scheduled_task(*args: Any, **kwargs: Any) -> Any:
            return task_func(*args, **kwargs)

        module = inspect.getmodule(task_func)
        if module is None:
            msg = f"Module for the task function {task_func.__name__} could not be found."
            raise ValueError(msg)

        task_path = f"{module.__name__}.{task_func.__name__}"
        current_app.conf.beat_schedule[task_func.__name__] = {
            "name": cron_scheduler_config.name,
            "task": task_path,
            "schedule": crontab(
                minute=cron_scheduler_config.minute,
                hour=cron_scheduler_config.hour,
                day_of_month=cron_scheduler_config.day_of_month,
                month_of_year=cron_scheduler_config.month_of_year,
                day_of_week=cron_scheduler_config.day_of_week,
            ),
        }

        return scheduled_task

    return decorator