Skip to content

Latest commit

Β 

History

History
64 lines (43 loc) Β· 2.17 KB

README.md

File metadata and controls

64 lines (43 loc) Β· 2.17 KB

Implementation of dependency injection for apscheduler

PyPI versioncodecovDownloads

Motivation:

  • apscheduler-di addresses the issue of apscheduler not natively supporting Dependency Injection, making it challenging for developers to pass complex objects to jobs without encountering issues.

Features:

  • Supports type hints (PEP 561)
  • Extend apscheduler and provide handy aliases for events(such as on_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

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_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()