apscheduler-di 0.1.0

Last updated:

0 purchases

apscheduler-di 0.1.0 Image
apscheduler-di 0.1.0 Images
Add to Cart

Description:

apschedulerdi 0.1.0

Implementation of dependency injection for apscheduler

Motivation:

apscheduler-di solves the problem since apscheduler doesn'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 apscheduler and provide handy aliases for events(such as on_startup, on_shutdown and
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_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()

License:

For personal and professional use. You cannot resell or redistribute these repositories in their original state.

Customer Reviews

There are no reviews.