at_multiple_of and force_missed_run
- Dominant language
- Python
- Stars
- 12.3k
- Forks
- 999
- PR merge metrics
- No merged PRs in 30d
Description
Hello I made this Override over your Classes, I Share it in case you want to implement it:
```python
class CustomScheduler(Scheduler):
def __init__(self):
super(CustomScheduler, self).__init__()
def at_multiple_of(self, interval):
"""
Schedule a new periodic job.
:param interval: A quantity of a certain time unit
:return: An unconfigured :class:`Job `
"""
job = CustomJob(interval, self)
job.run_at_multiple_of = True
return job
class CustomJob(Job):
def __init__(self, interval, scheduler=None):
super(CustomJob, self).__init__(interval, scheduler)
self.run_at_multiple_of = False
self._force_missed_run = False
self.future_run = None
@property
def force_missed_run(self):
self._force_missed_run = True
return self
@property
def modulo(self):
if self.unit == "minutes":
return self.next_run.minute % self.interval
if self.unit == "hours":
return self.next_run.hour % self.interval
if self.unit == "days":
return self.next_run.day % self.interval
def _schedule_next_run(self):
super(CustomJob, self)._schedule_next_run()
if self.latest is None and self.run_at_multiple_of:
if self.modulo != 0:
self.next_run -= datetime.timedelta(**{self.unit: self.modulo})
if self.future_run and self.future_run < self.next_run and self._force_missed_run:
self.next_run, self.future_run = self.future_run, self.next_run
else:
self.future_run = self.next_run + self.period
```
I implemented 2 features:
- **at_multiple_of**
:warning: **If the unit is not divisible by your multiple:**
ex : 60 minutes are not divisible by 7, it will be scheduled at:
00:00, 00:07, 00:14, 00:21, 00:28, 00:35, 00:42, 00:49, 00:56 and **01:00** and not **01:03**
exemple with **every**:
```python
schedule.every(5).minutes.at(":01").do(do_something)
```
results:
```
Launched at 13:33:52
Start at 13:38:01
Start at 13:43:01
Start at 13:48:01
```
exemple with **at_multiple_of**:
```python
schedule.at_multiple_of(5).minutes.at(":01").do(do_something_at_multiple)
```
results:
```
Launched at 13:33:52
Start at 13:35:01
Start at 13:40:01
Start at 13:45:01
```
- **force_missed_run**
Re execute a job that missed a run (when a job took longer than the periode)
exemple :
```python
schedule.every().minutes.at(":01").do(do_something)
```
results:
```
Launched at 13:33:52
Start at 13:34:01
Done at 13:35:10
Start at 13:36:01
Done at 13:37:10
Start at 13:38:01
```
```python
schedule.every().minutes.at(":01").force_missed_run.do(do_something)
```
results:
```
Launched at 13:33:52
Start at 13:34:01
Done at 13:34:10
Start at 13:34:10
Done at 13:35:21
Start at 13:35:21
```
PS: The Architecture of this Library is awesome, It's simple but really effective and user-friendly
Great Job!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by reviewing the existing Scheduler and Job entry points, especially how periodic jobs calculate their next run. Compare that behavior with the proposed at_multiple_of and force_missed_run examples. Done means both requested scheduling behaviors are integrated consistently and covered by appropriate project tests, though no test files are named in the issue.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- tooling
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 25/100