|
1 | 1 | import json |
2 | | -from typing import Optional, List |
| 2 | +from typing import Optional, List, Callable |
3 | 3 | import logging |
4 | 4 |
|
5 | 5 | from apscheduler.jobstores.base import ConflictingIdError |
@@ -125,51 +125,53 @@ def create_schedule(report_name): |
125 | 125 | return jsonify({"status": "Failed", "content": str(e)}) |
126 | 126 |
|
127 | 127 |
|
128 | | -def convert_day_of_week(day_of_week: str) -> str: |
| 128 | +def _convert_day_of_week(day_of_week: str, convert_func: Callable) -> str: |
129 | 129 | """ |
130 | | - UNIX standard maps days-of-week to ints as: |
131 | | - SUN: 0 |
132 | | - MON: 1 |
133 | | - TUE: 2.... |
134 | | -
|
135 | | - APScheduler uses: |
136 | | - MON: 0 |
137 | | - TUE: 1 |
138 | | - WED: 2..... |
139 | | -
|
140 | | - Given we are providing a UNIX-style crontab, convert ints accordingly. Don't shift char-based descriptors, |
141 | | - ie., 'MON-FRI'. |
| 130 | + Given we are providing a crontab converts the int-based day specification according to the function passed. |
| 131 | + Does not shift char-based descriptors i.e. 'MON-FRI'. |
142 | 132 | Parameters |
143 | 133 | ---------- |
144 | 134 | day_of_week - str - "FRI", "MON-FRI", "1"(UNIX Monday) |
145 | | -
|
146 | | - Returns |
147 | | - ------- |
148 | | - day_of_week formatted to APScheduler standard |
| 135 | + convert_func - function to use for conversion |
149 | 136 | """ |
150 | 137 |
|
151 | 138 | def shift(mychar): |
152 | 139 | if mychar.isnumeric(): |
153 | 140 | myint = int(mychar) |
154 | | - return str((myint + 6) % 7) |
| 141 | + return str(convert_func(myint)) |
155 | 142 | else: |
156 | 143 | return mychar |
157 | 144 |
|
158 | 145 | return "".join([shift(char) for char in day_of_week]) |
159 | 146 |
|
160 | 147 |
|
| 148 | +def crontab_to_apscheduler_day_of_week(day_of_week: str) -> str: |
| 149 | + """ |
| 150 | + Converts UNIX standard days-of-week (SUN=0, MON=1, ...) to APScheduler ones (MON=0, TUE=1, ...) |
| 151 | + """ |
| 152 | + return _convert_day_of_week(day_of_week, lambda dow: (dow + 6) % 7) |
| 153 | + |
| 154 | + |
| 155 | +def apscheduler_to_crontab_day_of_week(day_of_week: str) -> str: |
| 156 | + """ |
| 157 | + Converts APScheduler days-of-week (MON=0, TUE=1, ...) to UNIX standard ones (SUN=0, MON=1, ...) |
| 158 | + """ |
| 159 | + return _convert_day_of_week(day_of_week, lambda dow: (dow - 6) % 7) |
| 160 | + |
| 161 | + |
161 | 162 | def validate_crontab(crontab: str, issues: List[str]) -> cron.CronTrigger: |
162 | 163 | parts = crontab.split() |
163 | 164 | if len(parts) != 5: |
164 | 165 | issues.append("The crontab key must be passed with a string using 5 crontab parts") |
165 | 166 | else: |
166 | | - parts[4] = convert_day_of_week(parts[4]) |
| 167 | + parts[4] = crontab_to_apscheduler_day_of_week(parts[4]) |
167 | 168 | return cron.CronTrigger(minute=parts[0], hour=parts[1], day=parts[2], month=parts[3], day_of_week=parts[4]) |
168 | 169 |
|
169 | 170 |
|
170 | 171 | def trigger_to_crontab(trigger: cron.CronTrigger) -> str: |
171 | 172 | fields = {f.name: str(f) for f in trigger.fields} |
172 | | - return f"{fields['minute']} {fields['hour']} {fields['day']} {fields['month']} {fields['day_of_week']}" |
| 173 | + day_of_week = apscheduler_to_crontab_day_of_week(fields["day_of_week"]) |
| 174 | + return f"{fields['minute']} {fields['hour']} {fields['day']} {fields['month']} {day_of_week}" |
173 | 175 |
|
174 | 176 |
|
175 | 177 | def _job_to_json(job): |
|
0 commit comments