Add intervals for scheduled task #498
Conversation
|
I saw that several tests failed on older versions of Pydantic and Python, so I pushed fixes. |
| from tests.utils import AsyncQueueBroker | ||
|
|
||
|
|
||
| @pytest.mark.anyio |
There was a problem hiding this comment.
I enabled anyio_mode = "auto" flag for pytest. So tests doesn't need to explicitly say that they are async with pytest.mark.anyio marker.
You can just remove it and tests should work as they were)
…er_intervals # Conflicts: # docs/guide/cli.md # taskiq/cli/scheduler/run.py # tests/cli/scheduler/test_task_delays.py
|
I have resolved the merge conflicts and fixed the failing tests. |
| raise ValueError("Either cron, interval, or datetime must be present.") | ||
|
|
||
| # Validate interval constraints | ||
| if self.interval is not None: |
There was a problem hiding this comment.
Seems like this logic should be in a separate validator for interval field. Also we can use strict=True and ge=1 arguments for pydantic.Field to check that it's really a positive integer.
I think it should look like that and work the same as your implementation:
class ScheduledTask(BaseModel):
.... # other field here
interval: Union[int, timedelta, None] = Field(ge=1, strict=True)
@pydantic.field_validator("interval", mode="before")
@classmethod
def __check_interval(cls, value: Any) -> int | None:
if isinstance(value, timedelta):
return value.total_seconds()
return valueThere was a problem hiding this comment.
Unfortunately, this approach won't work because the strict=True parameter doesn't function with Union annotations, and le=1 is not designed for timedelta classes.
However, I've reviewed the validation for the interval parameter and decided to extract it into a separate function to avoid code duplication between the v1 and v2 schemas.
| "If not specified, scheduler will run once a minute." | ||
| ), | ||
| ) | ||
| parser.add_argument( |
There was a problem hiding this comment.
I don't understand why we need --loop-inteval if we alerady have --update-interval argument in line 96. Can you explain the difference between this two?
There was a problem hiding this comment.
The --loop-interval parameter specifies how often to check tasks for execution needs, while --update-interval specifies how often to synchronize the list of scheduled tasks.
I separated them for two reasons:
- Polling sources every second might be too resource-intensive for some types due to network overhead (Redis, PostgreSQL, etc.). This is why the
--update-intervalparameter cannot regulate the loop interval. - One second is quite frequent; some users might want to increase this interval. This is why the
--loop-intervalparameter is needed.
3 participants
Hi!
Following up on PR #399, which I left unfinished half a year ago:
In this PR, I want to add intervals for scheduled tasks.
List of changes:
intervalfield toScheduledTask.SchedulerLoopclass—an abstraction over the scheduler loop.loop_intervalargument toSchedulerArgs.skip_first_runmechanism intoSchedulerLoop.run().teskiq.api.scheduler.run_scheduler_tasksfunction now usesSchedulerLoop.schedule_by_intervalmethod in theAsyncTaskiqDecoratedTaskandAsyncKickerclasses.The main idea is to iterate through all tasks every
loop_intervaland check if they need to be executed.Example:
Logs: