(schedule: Any, timezone: str | None = None)
| 61 | |
| 62 | |
| 63 | def _task_schedule_from_input(schedule: Any, timezone: str | None = None) -> TaskSchedule: |
| 64 | if isinstance(schedule, str): |
| 65 | parts = schedule.split() |
| 66 | schedule_data: dict[str, Any] = { |
| 67 | "minute": parts[0] if len(parts) > 0 else "*", |
| 68 | "hour": parts[1] if len(parts) > 1 else "*", |
| 69 | "day": parts[2] if len(parts) > 2 else "*", |
| 70 | "month": parts[3] if len(parts) > 3 else "*", |
| 71 | "weekday": parts[4] if len(parts) > 4 else "*", |
| 72 | } |
| 73 | elif isinstance(schedule, dict): |
| 74 | schedule_data = dict(schedule) |
| 75 | else: |
| 76 | schedule_data = {} |
| 77 | |
| 78 | task_schedule_kwargs = { |
| 79 | "minute": str(schedule_data.get("minute", "*")), |
| 80 | "hour": str(schedule_data.get("hour", "*")), |
| 81 | "day": str(schedule_data.get("day", "*")), |
| 82 | "month": str(schedule_data.get("month", "*")), |
| 83 | "weekday": str(schedule_data.get("weekday", "*")), |
| 84 | } |
| 85 | normalized_timezone = _normalize_timezone(timezone if timezone is not None else schedule_data.get("timezone")) |
| 86 | if normalized_timezone: |
| 87 | task_schedule_kwargs["timezone"] = normalized_timezone |
| 88 | |
| 89 | return TaskSchedule(**task_schedule_kwargs) |
| 90 | |
| 91 | |
| 92 | def _validate_task_schedule(task_schedule: TaskSchedule) -> str: |
no test coverage detected