JobContext¶

JobContext is a specialized object that encapsulates the entire execution environment of a job. It serves as the single source of truth for handlers and middleware, providing access to job state, global configuration, and system resources.

Context Attributes¶

JobContext provides complete insight into the execution lifecycle through the following attributes:

• app: Jobify — The Jobify application instance.
• job: Job[Any] — Live information about the task (ID, status, metadata).
• state: State — The global app.state, shared across all jobs (e.g., DB connections).
• runnable: Runnable[Any] — Internal execution strategy and bound arguments.
• request_state: RequestState — Temporary state existing only for the duration of this job.
• route_options: RouteOptions — Configuration from the @app.task(...) decorator.
• jobify_config: JobifyConfiguration — Global application settings.
• schedule_builder: ScheduleBuilder[Any] — Access to internal scheduling state.

Context Injection¶

While middleware has direct access to JobContext, your task functions can access it via dependency injection. Declare a parameter with the appropriate type hint and assign it the INJECT default value.

from jobify import INJECT, Jobify, JobContext

app = Jobify()

@app.task
def my_job(ctx: JobContext = INJECT) -> None:
    print(f"Executing job {ctx.job.id}")
    # Access ctx.state, ctx.route_options, etc.

from jobify import INJECT, Job, State, Jobify

app = Jobify()

@app.task
def clean_task(current_job: Job[None] = INJECT, state: State = INJECT) -> None:
    print(f"Job ID: {current_job.id}")
    db = state["db"]

OuterContext¶

OuterContext is the scheduling-phase counterpart to JobContext. It is used exclusively by outer middleware to inspect or modify a job before it is registered or persisted.

Context Attributes¶

OuterContext provides a comprehensive view of the scheduling attempt:

• app: Jobify — The Jobify application instance.
• job: Job[Any] — The job instance being scheduled.
• state: State — The global application state.
• trigger: Triggers — The trigger mechanism (e.g., Cron, Push).
• runnable: Runnable[Any] — The runnable component being scheduled.
• arguments: dict[str, Any] — The arguments bound to the job function.
• func_spec: FuncSpec[Any] — Inspection details of the job function.
• is_force: bool — Whether the job is forced to run.
• is_persist: bool — Whether the job should be persisted to storage.
• is_replace: bool — Whether the job replaces an existing one.
• route_options: RouteOptions — Configuration options for the route.
• jobify_config: JobifyConfiguration — Global Jobify configuration.
• request_state: RequestState — State specific to the current request.
• persist_job_hook: Callable — Callback to persist the job.
• schedule_hook: Callable — Callback to schedule the job.
• schedule_builder: ScheduleBuilder[Any] — Access to internal scheduling state.

Key Differences¶

Using OuterContext¶

Outer middleware receives OuterContext to perform pre-scheduling logic. This is ideal for validating job arguments or injecting metadata before persistence.

import asyncio
from jobify import Jobify, OuterContext
from jobify.middleware import BaseOuterMiddleware, CallNextOuter

class ValidationMiddleware(BaseOuterMiddleware):
    async def __call__(
        self,
        call_next: CallNextOuter,
        ctx: OuterContext,
    ) -> asyncio.Handle:
        # Inspect arguments before scheduling
        if ctx.arguments.get("admin") and not ctx.arguments.get("is_authorized"):
            msg = f"Unauthorized scheduling attempt for job: {ctx.job.id}"
            raise PermissionError(msg)

        # Proceed to scheduling
        return await call_next(ctx)

app = Jobify(outer_middleware=[ValidationMiddleware()])