bedrock.generators.generate-schedules-for-terraform

This script generates the schedules.json for Terraform configuration.

 1"""
 2This script generates the schedules.json for Terraform configuration.
 3"""
 4
 5import json
 6import os
 7import sys
 8from pathlib import Path
 9
10from bedrock._helpers.string import camelCase_to_snake_case, camelCase_to_kebab_case
11from bedrock.config._modules import _get_modules
12from bedrock.schedules.schedule_endpoint import ScheduleEndpoint
13
14BASE_PATH = None
15
16
17def _get_schedules(dir_path="schedules", module_prefix="schedules."):
18    """
19    Discover all schedule classes in the schedules directory.
20    """
21    modules = _get_modules(dir_path, module_prefix, [b"__init__.py"])
22    schedules = {}
23    for module_name, cls in modules.items():
24        if issubclass(cls, ScheduleEndpoint) and cls != ScheduleEndpoint:
25            schedules[module_name] = cls()
26    return schedules
27
28
29def make_schedules_json(schedule_instances: dict):
30    """
31    Runs through each schedule and generates a JSON object with the schedule configuration.
32
33    Result example:
34    ```json
35    {
36        "daily_cleanup": {
37            "name": "daily-cleanup",
38            "cron": "0 8 * * ? *",
39            "type": "python",
40            "runtime": "python3.13",
41            "zipName": "schedule_daily_cleanup.zip"
42        }
43    }
44    ```
45    """
46    schedules = {}
47    for module_name, schedule_instance in schedule_instances.items():
48        cron_expression = getattr(schedule_instance.__class__, "__cron__", None)
49        if cron_expression is None:
50            print(f"🟡 [WARNING] Schedule {module_name} has no @cron decorator, skipping...")
51            continue
52
53        snake_case_name = camelCase_to_snake_case(schedule_instance.__class__.__name__)
54        kebab_case_name = camelCase_to_kebab_case(schedule_instance.__class__.__name__)
55
56        schedules[snake_case_name] = {
57            "name": kebab_case_name,
58            "cron": cron_expression,
59            "type": "python",
60            "runtime": "python3.13",
61            "zipName": f"schedule_{snake_case_name}.zip"
62        }
63
64    return schedules
65
66
67def dump_dict_to_file(filename: str, d: dict, folder="./"):
68    directory = f"{folder}/".replace("//", "/")
69    Path(directory).mkdir(parents=True, exist_ok=True)
70    with open(f"{directory}{filename}", "w") as f:
71        json.dump(d, f, indent=2)
72
73
74if __name__ == '__main__':
75    BASE_PATH = sys.argv[1]
76    schedules_dir = f"{BASE_PATH}/app/schedules"
77
78    if not os.path.isdir(schedules_dir):
79        print("🟡 [INFO] No schedules directory found, creating empty schedules.json")
80        dump_dict_to_file("schedules.json", {}, f"{BASE_PATH}/tf.support.files/")
81    else:
82        print("🔵 [INFO] Generating schedules.json...")
83        schedules = _get_schedules(schedules_dir)
84        dump_dict_to_file("schedules.json", make_schedules_json(schedules), f"{BASE_PATH}/tf.support.files/")
85    print("🟢 [SUCCESS] Done!")
BASE_PATH = None
def make_schedules_json(schedule_instances: dict): 
30def make_schedules_json(schedule_instances: dict):
31    """
32    Runs through each schedule and generates a JSON object with the schedule configuration.
33
34    Result example:
35    ```json
36    {
37        "daily_cleanup": {
38            "name": "daily-cleanup",
39            "cron": "0 8 * * ? *",
40            "type": "python",
41            "runtime": "python3.13",
42            "zipName": "schedule_daily_cleanup.zip"
43        }
44    }
45    ```
46    """
47    schedules = {}
48    for module_name, schedule_instance in schedule_instances.items():
49        cron_expression = getattr(schedule_instance.__class__, "__cron__", None)
50        if cron_expression is None:
51            print(f"🟡 [WARNING] Schedule {module_name} has no @cron decorator, skipping...")
52            continue
53
54        snake_case_name = camelCase_to_snake_case(schedule_instance.__class__.__name__)
55        kebab_case_name = camelCase_to_kebab_case(schedule_instance.__class__.__name__)
56
57        schedules[snake_case_name] = {
58            "name": kebab_case_name,
59            "cron": cron_expression,
60            "type": "python",
61            "runtime": "python3.13",
62            "zipName": f"schedule_{snake_case_name}.zip"
63        }
64
65    return schedules

Runs through each schedule and generates a JSON object with the schedule configuration.

Result example:

{
    "daily_cleanup": {
        "name": "daily-cleanup",
        "cron": "0 8 * * ? *",
        "type": "python",
        "runtime": "python3.13",
        "zipName": "schedule_daily_cleanup.zip"
    }
}
def dump_dict_to_file(filename: str, d: dict, folder='./'): 
68def dump_dict_to_file(filename: str, d: dict, folder="./"):
69    directory = f"{folder}/".replace("//", "/")
70    Path(directory).mkdir(parents=True, exist_ok=True)
71    with open(f"{directory}{filename}", "w") as f:
72        json.dump(d, f, indent=2)