apscheduler-dishka 1.1.0

Apscheduler integration for dishka

APScheduler integration for Dishka

This package provides integration of Dishka dependency injection framework and APScheduler, an advanced Python scheduler.

Installation

pip install apscheduler-dishka

Features

• automatic REQUEST scope management for scheduled jobs
• automatic injection of dependencies into job functions
• support for both sync and async schedulers
• support for auto_inject mode to automatically wrap all jobs

How to use

1. Import

from apscheduler_dishka import (
    inject,
    setup_dishka,
)
from dishka import make_container, Provider, provide, Scope, FromDishka

2. Create provider

class YourProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def create_interactor(self) -> Interactor:
         ...

3. Mark those of your job function parameters which are to be injected with FromDishka[] and decorate them using @inject

@inject
def job(
    data: str,
    interactor: FromDishka[Interactor],
):
    ...

4. Create container

container = make_container(YourProvider())

5. Setup dishka integration

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler_dishka import setup_dishka

scheduler = BlockingScheduler()
setup_dishka(container=container, scheduler=scheduler)

6. Add jobs to scheduler

scheduler.add_job(
    job,
    trigger="interval",
    seconds=5,
    args=["args"],
)

Auto-inject mode

If you don't want to manually decorate each job with @inject, you can use auto_inject mode:

setup_dishka(
    container=container,
    scheduler=scheduler,
    auto_inject=True,
)

In this mode, all jobs will be automatically wrapped with dependency injection:

def job(
    data: str,
    interactor: FromDishka[Interactor],
):
    ...

scheduler.add_job(
    job,
    trigger="interval",
    seconds=5,
    args=["args"],
)

Async support

The library supports async schedulers:

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from dishka import make_async_container
from apscheduler_dishka import setup_dishka

scheduler = AsyncIOScheduler()
container = make_async_container(YourProvider())
setup_dishka(container=container, scheduler=scheduler)