Jobify integration for Dishka

Though it is not required, you can use dishka-jobify integration. It features:

• REQUEST scope management using jobify request scope
• JobifyProvider for working with JobContext, Job, State, RequestState, Runnable in container

You need to specify @inject manually.

Installation

Install using pip

pip install dishka-jobify

Or with uv

uv add dishka-jobify

How to use

1. Import

from dishka_jobify import (
    FromDishka,
    inject,
    setup_dishka
)
from dishka import make_async_container, Provider, provide, Scope

2. Create provider like here below

class MyProvider(Provider):
    @provide(scope=Scope.REQUEST)
    def greeting_service(self, context: JobContext) -> GreetingService:
        return GreetingService(job_id=str(context.job.id))

    @provide(scope=Scope.APP)
    def counter_service(self) -> CounterService:
        return CounterService()

3. Mark those of your handlers parameters which are to be injected with FromDishka[]

@app.task
@inject
async def my_cron(
    greeting: FromDishka[GreetingService],
    counter: FromDishka[CounterService],
) -> None:
    count = counter.increment()
    print(f"[cron] {greeting.greet('cron')} count={count}")

4. Setup dishka integration.

UTC = ZoneInfo("UTC")
app = Jobify(tz=UTC)

provider = MyProvider()
container = make_async_container(provider, JobifyProvider())
setup_dishka(container=container, app=app)