Implementation of dependency injection for apscheduler
Prerequisites:
apscheduler-disolves the problem sinceapschedulerdoesn't support Dependency Injection natively, and it's real problem for developers to pass on complicated objects to jobs without corruptions
Features:
- Supports type hints (PEP 561)
- Extend
apschedulerand provide handy aliases for events(such ason_startup,on_shutdownand etc) - Provide an opportunity to implement Dependency Inversion SOLID principle
"Under the hood" apscheduler-di just implements Decorator pattern and wraps up the work of native BaseScheduler using rodi lib
Quick example:
import os
from typing import Dict
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler_di import ContextSchedulerDecorator
# pip install redis
job_defaults: Dict[str, RedisJobStore] = {
"default": RedisJobStore(
jobs_key="dispatched_trips_jobs", run_times_key="dispatched_trips_running"
)
}
job_stores: Dict[str, RedisJobStore] = {
"default": RedisJobStore(
jobs_key="dispatched_trips_jobs", run_times_key="dispatched_trips_running"
)
}
class Tack:
def tack(self):
print("Tack!")
def tick(tack: Tack):
print(tack)
def main():
scheduler = ContextSchedulerDecorator(BlockingScheduler(jobstores=job_stores,
job_defaults=job_defaults))
scheduler.ctx.add_instance(Tack(), Tack)
scheduler.add_executor('processpool')
scheduler.add_job(tick, 'interval', seconds=3)
print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))
try:
scheduler.start()
except (KeyboardInterrupt, SystemExit):
pass
if __name__ == '__main__':
main()
12 Nov 9, 2022
92 Jul 20, 2022
15 Sep 23, 2022
2 Nov 9, 2022
9 Dec 30, 2022
67 Oct 25, 2022
2 Oct 29, 2022
9 Aug 31, 2022
32 Nov 5, 2022
37 Nov 12, 2022
1 Jan 24, 2022
3 Mar 24, 2022
14 Jun 22, 2022
4.1k Jan 09, 2023
94 Dec 13, 2022
8 Jan 03, 2023
291 Dec 28, 2022
2 Jul 05, 2022
14 Oct 18, 2022
42 Nov 09, 2022
72 Jan 01, 2023
3 Oct 04, 2022
1 Dec 21, 2021
5 Nov 17, 2022
8 Aug 26, 2022
81 Dec 27, 2022