Coverage for /usr/local/lib/python3.12/site-packages/prefect/schedules.py: 37%
41 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 11:21 +0000
1"""
2This module contains functionality for creating schedules for deployments.
3"""
5from __future__ import annotations 1a
7import dataclasses 1a
8import datetime 1a
9from functools import partial 1a
10from typing import Any 1a
12from prefect._internal.schemas.validators import ( 1a
13 validate_cron_string,
14 validate_rrule_string,
15)
18@dataclasses.dataclass(frozen=True) 1a
19class Schedule: 1a
20 """
21 A dataclass representing a schedule.
23 Note that only one of `interval`, `cron`, or `rrule` can be defined at a time.
25 Attributes:
26 interval: A timedelta representing the frequency of the schedule.
27 cron: A valid cron string (e.g. "0 0 * * *").
28 rrule: A valid RRule string (e.g. "RRULE:FREQ=DAILY;INTERVAL=1").
29 timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
30 anchor_date: An anchor date to schedule increments against; if not provided,
31 the current timestamp will be used.
32 day_or: Control how `day` and `day_of_week` entries are handled.
33 Defaults to True, matching cron which connects those values using
34 OR. If the switch is set to False, the values are connected using AND.
35 This behaves like fcron and enables you to e.g. define a job that
36 executes each 2nd friday of a month by setting the days of month and
37 the weekday.
38 active: Whether or not the schedule is active.
39 parameters: A dictionary containing parameter overrides for the schedule.
40 slug: A unique identifier for the schedule.
41 """
43 interval: datetime.timedelta | None = None 1a
44 cron: str | None = None 1a
45 rrule: str | None = None 1a
46 timezone: str | None = None 1a
47 anchor_date: datetime.datetime = dataclasses.field( 1a
48 default_factory=partial(datetime.datetime.now, tz=datetime.timezone.utc)
49 )
50 day_or: bool = True 1a
51 active: bool = True 1a
52 parameters: dict[str, Any] = dataclasses.field(default_factory=dict) 1a
53 slug: str | None = None 1a
55 def __post_init__(self) -> None: 1a
56 defined_fields = [
57 field
58 for field in ["interval", "cron", "rrule"]
59 if getattr(self, field) is not None
60 ]
61 if len(defined_fields) > 1:
62 raise ValueError(
63 f"Only one schedule type can be defined at a time. Found: {', '.join(defined_fields)}"
64 )
66 if self.cron is not None:
67 validate_cron_string(self.cron)
68 if self.rrule is not None:
69 validate_rrule_string(self.rrule)
72def Cron( 1a
73 cron: str,
74 /,
75 timezone: str | None = None,
76 day_or: bool = True,
77 active: bool = True,
78 parameters: dict[str, Any] | None = None,
79 slug: str | None = None,
80) -> Schedule:
81 """
82 Creates a cron schedule.
84 Args:
85 cron: A valid cron string (e.g. "0 0 * * *").
86 timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
87 day_or: Control how `day` and `day_of_week` entries are handled.
88 Defaults to True, matching cron which connects those values using
89 OR. If the switch is set to False, the values are connected using AND.
90 This behaves like fcron and enables you to e.g. define a job that
91 executes each 2nd friday of a month by setting the days of month and
92 the weekday.
93 active: Whether or not the schedule is active.
94 parameters: A dictionary containing parameter overrides for the schedule.
95 slug: A unique identifier for the schedule.
97 Returns:
98 A cron schedule.
100 Examples:
101 Create a cron schedule that runs every day at 12:00 AM UTC:
102 ```python
103 from prefect.schedules import Cron
105 Cron("0 0 * * *")
106 ```
108 Create a cron schedule that runs every Monday at 8:00 AM in the America/New_York timezone:
109 ```python
110 from prefect.schedules import Cron
112 Cron("0 8 * * 1", timezone="America/New_York")
113 ```
115 """
116 if parameters is None:
117 parameters = {}
118 return Schedule(
119 cron=cron,
120 timezone=timezone,
121 day_or=day_or,
122 active=active,
123 parameters=parameters,
124 slug=slug,
125 )
128def Interval( 1a
129 interval: datetime.timedelta | float | int,
130 /,
131 anchor_date: datetime.datetime | None = None,
132 timezone: str | None = None,
133 active: bool = True,
134 parameters: dict[str, Any] | None = None,
135 slug: str | None = None,
136) -> Schedule:
137 """
138 Creates an interval schedule.
140 Args:
141 interval: The interval to use for the schedule. If an integer is provided,
142 it will be interpreted as seconds.
143 anchor_date: The anchor date to use for the schedule.
144 timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
145 active: Whether or not the schedule is active.
146 parameters: A dictionary containing parameter overrides for the schedule.
147 slug: A unique identifier for the schedule.
149 Returns:
150 An interval schedule.
152 Examples:
153 Create an interval schedule that runs every hour:
154 ```python
155 from datetime import timedelta
157 from prefect.schedules import Interval
159 Interval(timedelta(hours=1))
160 ```
162 Create an interval schedule that runs every 60 seconds starting at a specific date:
163 ```python
164 from datetime import datetime
166 from prefect.schedules import Interval
168 Interval(60, anchor_date=datetime(2024, 1, 1))
169 ```
170 """
171 if isinstance(interval, (float, int)):
172 interval = datetime.timedelta(seconds=interval)
173 if anchor_date is None:
174 anchor_date = datetime.datetime.now(tz=datetime.timezone.utc)
175 if parameters is None:
176 parameters = {}
177 return Schedule(
178 interval=interval,
179 anchor_date=anchor_date,
180 timezone=timezone,
181 active=active,
182 parameters=parameters,
183 slug=slug,
184 )
187def RRule( 1a
188 rrule: str,
189 /,
190 timezone: str | None = None,
191 active: bool = True,
192 parameters: dict[str, Any] | None = None,
193 slug: str | None = None,
194) -> Schedule:
195 """
196 Creates an RRule schedule.
198 Args:
199 rrule: A valid RRule string (e.g. "RRULE:FREQ=DAILY;INTERVAL=1").
200 timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
201 active: Whether or not the schedule is active.
202 parameters: A dictionary containing parameter overrides for the schedule.
203 slug: A unique identifier for the schedule.
205 Returns:
206 An RRule schedule.
208 Examples:
209 Create an RRule schedule that runs every day at 12:00 AM UTC:
210 ```python
211 from prefect.schedules import RRule
213 RRule("RRULE:FREQ=DAILY;INTERVAL=1")
214 ```
216 Create an RRule schedule that runs every 2nd friday of the month in the America/Chicago timezone:
217 ```python
218 from prefect.schedules import RRule
220 RRule("RRULE:FREQ=MONTHLY;INTERVAL=1;BYDAY=2FR", timezone="America/Chicago")
221 ```
222 """
223 if parameters is None:
224 parameters = {}
225 return Schedule(
226 rrule=rrule,
227 timezone=timezone,
228 active=active,
229 parameters=parameters,
230 slug=slug,
231 )