apscheduler-di
addresses the issue ofapscheduler
not natively supporting Dependency Injection, making it challenging for developers to pass complex objects to jobs without encountering issues.
- Supports type hints (PEP 561)
- Extend
apscheduler
and provide handy aliases for events(such ason_startup
,on_shutdown
and etc) - Offers a way to implement the Dependency Inversion principle from SOLID design principles.
"Under the hood" apscheduler-di
implements Decorator pattern and wraps up the
work of native BaseScheduler
using rodi lib
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_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, some_argument: int):
print(tack)
def main():
scheduler = ContextSchedulerDecorator(BlockingScheduler(jobstores=job_stores))
scheduler.ctx.add_instance(Tack(), Tack)
scheduler.add_executor('processpool')
scheduler.add_job(tick, 'interval', seconds=3, kwargs={"some_argument": 5})
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()