-
-
Notifications
You must be signed in to change notification settings - Fork 52
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Pavel Kirilin <[email protected]>
- Loading branch information
Showing
10 changed files
with
118 additions
and
51 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,11 @@ | ||
from taskiq.compat import IS_PYDANTIC2 | ||
|
||
from .cron_spec import CronSpec | ||
|
||
if IS_PYDANTIC2: | ||
from .v2 import ScheduledTask | ||
else: | ||
from .v1 import ScheduledTask # type: ignore | ||
|
||
|
||
__all__ = ["CronSpec", "ScheduledTask"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,20 @@ | ||
from datetime import timedelta | ||
from typing import Optional, Union | ||
|
||
from pydantic import BaseModel | ||
|
||
|
||
class CronSpec(BaseModel): | ||
"""Cron specification for running tasks.""" | ||
|
||
minutes: Optional[Union[str, int]] = "*" | ||
hours: Optional[Union[str, int]] = "*" | ||
days: Optional[Union[str, int]] = "*" | ||
months: Optional[Union[str, int]] = "*" | ||
weekdays: Optional[Union[str, int]] = "*" | ||
|
||
offset: Optional[Union[str, timedelta]] = None | ||
|
||
def to_cron(self) -> str: # pragma: no cover | ||
"""Converts cron spec to cron string.""" | ||
return f"{self.minutes} {self.hours} {self.days} {self.months} {self.weekdays}" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,30 @@ | ||
import uuid | ||
from datetime import datetime, timedelta | ||
from typing import Any, Dict, List, Optional, Union | ||
|
||
from pydantic import BaseModel, Field, root_validator | ||
|
||
|
||
class ScheduledTask(BaseModel): | ||
"""Abstraction over task schedule.""" | ||
|
||
task_name: str | ||
labels: Dict[str, Any] | ||
args: List[Any] | ||
kwargs: Dict[str, Any] | ||
schedule_id: str = Field(default_factory=lambda: uuid.uuid4().hex) | ||
cron: Optional[str] = None | ||
cron_offset: Optional[Union[str, timedelta]] = None | ||
time: Optional[datetime] = None | ||
|
||
@root_validator(pre=False) # type: ignore | ||
@classmethod | ||
def __check(cls, values: Dict[str, Any]) -> Dict[str, Any]: | ||
""" | ||
This method validates, that either `cron` or `time` field is present. | ||
:raises ValueError: if cron and time are none. | ||
""" | ||
if values.get("cron") is None and values.get("time") is None: | ||
raise ValueError("Either cron or datetime must be present.") | ||
return values |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,30 @@ | ||
import uuid | ||
from datetime import datetime, timedelta | ||
from typing import Any, Dict, List, Optional, Union | ||
|
||
from pydantic import BaseModel, Field, model_validator | ||
from typing_extensions import Self | ||
|
||
|
||
class ScheduledTask(BaseModel): | ||
"""Abstraction over task schedule.""" | ||
|
||
task_name: str | ||
labels: Dict[str, Any] | ||
args: List[Any] | ||
kwargs: Dict[str, Any] | ||
schedule_id: str = Field(default_factory=lambda: uuid.uuid4().hex) | ||
cron: Optional[str] = None | ||
cron_offset: Optional[Union[str, timedelta]] = None | ||
time: Optional[datetime] = None | ||
|
||
@model_validator(mode="after") | ||
def __check(self) -> Self: | ||
""" | ||
This method validates, that either `cron` or `time` field is present. | ||
:raises ValueError: if cron and time are none. | ||
""" | ||
if self.cron is None and self.time is None: | ||
raise ValueError("Either cron or datetime must be present.") | ||
return self |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,14 @@ | ||
import pytest | ||
|
||
from taskiq.scheduler.scheduled_task import ScheduledTask | ||
|
||
|
||
def test_scheduled_task_paramters() -> None: | ||
with pytest.raises(ValueError): | ||
ScheduledTask( | ||
task_name="a", | ||
labels={}, | ||
args=[], | ||
kwargs={}, | ||
schedule_id="b", | ||
) |