### Basic Jobify Quick Start

Source: https://github.com/theseriff/jobify/blob/main/README.md

A simple example demonstrating how to set up Jobify, define a task, and run it. This snippet shows the fundamental usage of the Jobify application and task definition.

```python
import asyncio

from jobify import Jobify, Job

app = Jobify()


@app.task
async def hello(name: str) -> None:
    print(f"Hello, {name}")


async def main() -> None:
    async with app:
        job: Job = await hello.push("Alex")
        await job.wait()


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Initialize Development Environment

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

Run this command after installing prerequisites to synchronize dependencies and install pre-commit hooks.

```bash
just init
```

--------------------------------

### Install Jobify

Source: https://github.com/theseriff/jobify/blob/main/README.md

Install the Jobify library using pip. This is the first step to using Jobify in your Python project.

```bash
pip install jobify
```

--------------------------------

### Install Dependencies with uv

Source: https://github.com/theseriff/jobify/blob/main/examples/real_world/README.md

Installs project dependencies using the uv package manager. Ensure uv is installed and configured for your project.

```bash
uv sync
```

--------------------------------

### Configure Task with Multiple Settings

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Example demonstrating how to configure a task with name, cron schedule, retries, timeout, durability, run mode, metadata, and exception handlers.

```python
from jobify import Cron, Jobify, MisfirePolicy, RunMode

app = Jobify()


@app.task(
    name="reports:daily",
    cron=Cron(
        "0 8 * * 1-5",
        max_runs=100,
        max_failures=5,
        misfire_policy=MisfirePolicy.SKIP,
    ),
    retry=3,
    timeout=300,
    durable=True,
    run_mode=RunMode.PROCESS,
    metadata={"priority": 100, "team": "finance"},
    exception_handlers={TimeoutError: lambda exc, ctx: "timed_out"},
)
def daily_report() -> None:
    ...
```

--------------------------------

### Quick Example of Jobify Usage

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Demonstrates scheduling a task, checking its initial status, waiting for completion, and retrieving the result. Requires importing Jobify and JobStatus.

```python
import asyncio

from jobify import Jobify, JobStatus

app = Jobify()


@app.task
def add(x: int, y: int) -> int:
    return x + y


async def main() -> None:
    async with app:
        job = await add.schedule(1, 2).delay(1)
        print(job.id, job.status)

        await job.wait()
        if job.status is JobStatus.SUCCESS:
            print(job.result())  # 3


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Extended Jobify Example with Cron, Delay, and Absolute Time Scheduling

Source: https://github.com/theseriff/jobify/blob/main/README.md

This example showcases advanced scheduling options in Jobify, including cron jobs, delayed execution, and scheduling tasks at specific absolute times. It also demonstrates setting a custom timezone for the application.

```python
import asyncio
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from jobify import Jobify, Job

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


@app.task(cron="* * * * * * *")  # every second
async def my_cron() -> None:
    print("cron tick")


@app.task
def my_job(name: str) -> None:
    now = datetime.now(tz=UTC)
    print(f"Hello, {name}! at {now!r}")


async def main() -> None:
    async with app:
        await my_job.push("Alex")

        run_next_day = datetime.now(tz=UTC) + timedelta(days=1)
        job_at: Job = await my_job.schedule("Connor").at(run_next_day)
        job_delay: Job = await my_job.schedule("Sara").delay(seconds=20)
        job_cron: Job = await my_cron.schedule().cron("* * * * *", job_id="dynamic_cron_id")

        await job_at.wait()
        await job_delay.wait()
        await job_cron.wait()


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Custom Plugin for App Lifecycle

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Register custom plugins to manage application startup and shutdown events. Plugins can perform setup and cleanup tasks.

```python
from jobify import Jobify, Plugin


class MyPlugin(Plugin):
    async def startup(self, app: Jobify) -> None:
        print("plugin startup")

    async def shutdown(self) -> None:
        print("plugin shutdown")


app = Jobify(plugins=[MyPlugin()])
```

--------------------------------

### Run the Jobify Application

Source: https://github.com/theseriff/jobify/blob/main/examples/real_world/README.md

Executes the jobify application using the uv run command. This command starts the application's scheduled tasks.

```bash
uv run real_world
```

--------------------------------

### Setup LIFO Queue

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/queue.md

Configures Jobify with QueueMiddleware using asyncio.LifoQueue for Last-In, First-Out (LIFO) job execution. Sets a maximum buffer size of 200 and allows up to 30 concurrent workers.

```python
import asyncio

from jobify import Jobify
from jobify.middleware import QueueMiddleware

app = Jobify()
app.add_middleware(
    QueueMiddleware(queue=asyncio.LifoQueue(maxsize=200), workers=30)
)
```

--------------------------------

### Setup FIFO Queue

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/queue.md

Configures Jobify with QueueMiddleware using asyncio.Queue for First-In, First-Out (FIFO) job execution. Sets a maximum buffer size of 200 and allows up to 20 concurrent workers.

```python
import asyncio

from jobify import Jobify
from jobify.middleware import QueueMiddleware

app = Jobify()
app.add_middleware(
    QueueMiddleware(queue=asyncio.Queue(maxsize=200), workers=20)
)
```

--------------------------------

### Jobify Lifespan with Custom State

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Demonstrates using a custom lifespan context manager to initialize application state, such as database connections or caches, before the application starts.

```python
import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import TypedDict

from jobify import Jobify


class AppState(TypedDict):
    db: object
    cache: dict[str, str]


@asynccontextmanager
async def lifespan(_: Jobify) -> AsyncIterator[AppState]:
    db = object()
    yield {"db": db, "cache": {}}


async def main() -> None:
    async with Jobify(lifespan=lifespan) as app:
        assert "db" in app.state


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Setup Default PriorityQueue

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/queue.md

Configures Jobify with QueueMiddleware using asyncio.PriorityQueue for prioritized job execution. Sets a maximum buffer size of 1024 and allows up to 10 concurrent workers.

```python
import asyncio

from jobify import Jobify
from jobify.middleware import QueueMiddleware

app = Jobify()
app.add_middleware(
    QueueMiddleware(queue=asyncio.PriorityQueue(maxsize=1024), workers=10)
)

@app.task(metadata={"priority": 100})
async def critical_job() -> None:
    ...

@app.task(metadata={"priority": 10})
async def regular_job() -> None:
    ...
```

--------------------------------

### Hierarchical Exception Handling Example

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Demonstrates how task-level handlers take precedence over router and global handlers when an exception occurs. The most specific handler is executed.

```python
# Global
app = Jobify(exception_handlers={TypeError: handle_global})

# Router
router = JobRouter(prefix="reports", exception_handlers={TypeError: handle_router})

# Task
@router.task(exception_handlers={TypeError: handle_task})
async def process_report():
    raise TypeError("Specific error")
```

--------------------------------

### Priority Job Execution Example

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/queue.md

Demonstrates how higher priority jobs are executed before lower priority jobs, even if enqueued later. Uses a PriorityQueue with a max size of 256 and a single worker.

```python
import asyncio

from jobify import Jobify
from jobify.middleware import QueueMiddleware

app = Jobify()
app.add_middleware(QueueMiddleware(queue=asyncio.PriorityQueue(maxsize=256), workers=1))


@app.task(metadata={"priority": 100})
async def charge_payment(order_id: str) -> str:
    return f"charged:{order_id}"


@app.task(metadata={"priority": 10})
async def send_analytics(event_id: str) -> str:
    return f"tracked:{event_id}"


async def main() -> None:
    async with app:
        low = await send_analytics.push("evt-1")
        high = await charge_payment.push("ord-1")
        await asyncio.gather(low.wait(), high.wait())
```

--------------------------------

### Cron Object Configuration

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Configure a cron schedule with advanced options like max runs, max failures, misfire policy, start date, and arguments.

```python
from datetime import datetime

from jobify import Cron, MisfirePolicy

cron = Cron(
    "0 18 * * 1-5",
    max_runs=100,
    max_failures=5,
    misfire_policy=MisfirePolicy.ALL,
    start_date=datetime(2027, 1, 1),
    args=("sales",),
    kwargs={"file_format": "csv"},
)
```

--------------------------------

### Define Tasks with JobRouter

Source: https://github.com/theseriff/jobify/blob/main/docs/router.md

Define tasks within a JobRouter, specifying a prefix for hierarchical naming. This example shows how to create a router for email-related tasks.

```python
from jobify import JobRouter

router = JobRouter(prefix="email")

@router.task
async def send_welcome(user_id: int) -> None:
    print(f"Sending welcome to {user_id}")
```

--------------------------------

### Define an Exception Handler Function

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

A handler is a callable that accepts the exception instance and the JobContext. This example shows a basic handler that logs the error.

```python
async def my_handler(exc: Exception, context: JobContext) -> None:
    print(f"Job {context.job.id} failed: {exc}")
```

--------------------------------

### Router-level Lifespan Management

Source: https://github.com/theseriff/jobify/blob/main/docs/router.md

Manages resources specific to a module using router lifespans, such as establishing and closing specialized database connections. The example defines an async context manager for a reports router.

```python
@asynccontextmanager
async def reports_lifespan(router: JobRouter) -> AsyncIterator[None]:
    router.state["db"] = await connect_to_reports_db()
    yield
    await router.state["db"].close()

reports_router = JobRouter(prefix="reports", lifespan=reports_lifespan)
```

--------------------------------

### Job Execution Logging Middleware

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Implement custom logic for job execution, such as logging start and end times. Ensure `call_next` is invoked to continue the middleware chain.

```python
import asyncio
import logging
from typing import Any

from jobify import JobContext, Jobify
from jobify.middleware import BaseMiddleware, CallNext

logging.basicConfig(level=logging.INFO)


class LoggingMiddleware(BaseMiddleware):
    async def __call__(self, call_next: CallNext, context: JobContext) -> Any:
        logging.info("Job %s started", context.job.id)
        try:
            return await call_next(context)
        finally:
            logging.info("Job %s finished", context.job.id)


app = Jobify(middleware=[LoggingMiddleware()])
```

--------------------------------

### Register JobRouter with Jobify Application

Source: https://github.com/theseriff/jobify/blob/main/docs/router.md

Include a defined JobRouter into the main Jobify application. This makes the tasks within the router active and schedulable. The example demonstrates including an email router and scheduling a task.

```python
from jobify import Jobify
from tasks.email import router as email_router
from tasks.email import send_welcome

app = Jobify()
app.include_router(email_router)

async def main():
    async with app:
        # Tasks are now active and schedulable
        await send_welcome.schedule(user_id=42).delay(0)
```

--------------------------------

### Jobify Constructor with Full Configuration

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Demonstrates initializing Jobify with a comprehensive set of arguments, including time zone, storage, serializer, middleware, and executor pools.

```python
import asyncio
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from zoneinfo import ZoneInfo

from jobify import Jobify, Plugin
from jobify.crontab import create_crontab
from jobify.middleware import QueueMiddleware
from jobify.router import RootRoute
from jobify.storage import SQLiteStorage
from jobify.serializers import JSONSerializer

app = Jobify(
    tz=ZoneInfo("UTC"),
    state=None,
    dumper=None,
    loader=None,
    storage=SQLiteStorage(),
    lifespan=None,
    serializer=JSONSerializer(),
    middleware=[QueueMiddleware()],
    outer_middleware=[],
    cron_factory=create_crontab,
    loop_factory=asyncio.get_running_loop,
    exception_handlers={},
    threadpool_executor=ThreadPoolExecutor(max_workers=8),
    processpool_executor=ProcessPoolExecutor(max_workers=4),
    route_class=RootRoute,
    plugins=(),
)
```

--------------------------------

### Jobify with Custom Storage Backend

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Shows how to integrate a custom storage backend by subclassing the `Storage` class and passing an instance to the Jobify constructor.

```python
from jobify import Jobify
from jobify.storage import Storage

class MyStorage(Storage):
    ...

app = Jobify(storage=MyStorage())
```

--------------------------------

### Basic Jobify Task Scheduling

Source: https://github.com/theseriff/jobify/blob/main/docs/index.md

Shows how to define cron and regular jobs, push jobs, and schedule jobs for specific times or delays. Requires importing asyncio, datetime, and Jobify.

```python
import asyncio
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from jobify import Jobify, Job

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


@app.task(cron="* * * * * * *")  # every second
async def my_cron() -> None:
    print("cron tick")


@app.task
def my_job(name: str) -> None:
    now = datetime.now(tz=UTC)
    print(f"Hello, {name}! at {now!r}")


async def main() -> None:
    async with app:
        await my_job.push("Alex")

        run_next_day = datetime.now(tz=UTC) + timedelta(days=1)
        job_at: Job = await my_job.schedule("Connor").at(run_next_day)
        job_delay: Job = await my_job.schedule("Sara").delay(20)
        job_cron: Job = await my_cron.schedule().cron("* * * * *", job_id="dynamic_cron_id")

        await job_at.wait()
        await job_delay.wait()
        await job_cron.wait()


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Run Pre-commit Hooks on All Files

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

It is recommended to run all pre-commit hooks on all project files before committing changes.

```bash
just pre-commit-all
```

--------------------------------

### Jobify with Default SQLite Storage

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Initializes Jobify using the default SQLite storage for persistence. This is the recommended default behavior.

```python
from jobify import Jobify

app = Jobify()  # SQLiteStorage() is used automatically
```

--------------------------------

### Dynamic Scheduling Builder

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Initiate dynamic scheduling by creating a builder from a task with optional arguments and keyword arguments.

```python
builder = my_task.schedule(*args, **kwargs)
```

--------------------------------

### Perform Static Analysis

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

Run Mypy, Basedpyright, Bandit, Semgrep, and Zizmor for comprehensive static code analysis.

```bash
just static-analysis
```

--------------------------------

### Configure Task Run Modes

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Specify the execution model for a task using 'run_mode'. Options include RunMode.MAIN for async, RunMode.THREAD for sync I/O, and RunMode.PROCESS for CPU-bound tasks.

```python
from jobify import RunMode


@app.task(run_mode=RunMode.MAIN)
async def async_job() -> None:
    ...


@app.task(run_mode=RunMode.THREAD)
def sync_io_job() -> None:
    ...


@app.task(run_mode=RunMode.PROCESS)
def cpu_heavy_job() -> None:
    ...
```

--------------------------------

### Concise Await Style for Job Handling

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

A recommended pattern for awaiting a job within a try-except block to gracefully handle potential exceptions during job execution.

```python
try:
    data = await job
except Exception as exc:
    print("job failed:", exc)
```

--------------------------------

### Batching Jobs with Awaitable Shorthand

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Demonstrates using the awaitable shorthand for jobs within `asyncio.gather` to concurrently wait for multiple jobs and collect their results.

```python
jobs = [await add.schedule(i, i).delay(0) for i in range(3)]
results = await asyncio.gather(*jobs)
```

--------------------------------

### Run All Tests

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

Execute the comprehensive pytest suite to verify all features and regressions.

```bash
just test-all
```

--------------------------------

### at

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Schedule a task for execution at an exact `datetime`. Supports `job_id`, `replace`, and `force` parameters for managing existing schedules.

```APIDOC
## at

Schedule execution at an exact `datetime`.

### Description
Schedules a task to run at a specific point in time. Allows specifying a `job_id`, and controlling replacement and middleware execution.

### Parameters

#### Request Body

- **at** (datetime) - Required - The exact datetime at which the job should run.
- **job_id** (string | None) - Optional - A unique identifier for the job. Defaults to `None`.
- **replace** (boolean) - Optional - If `True`, updates an existing job with the same `job_id`. Defaults to `False`.
- **force** (boolean) - Optional - If `True`, forces outer middleware execution even if the schedule is unchanged. Defaults to `False`.

### Example
```python
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

run_at = datetime.now(tz=ZoneInfo("UTC")) + timedelta(minutes=10)
job = await my_task.schedule(*args, **kwargs).at(
    at=run_at,
    job_id="report_123",
    replace=False,
    force=False,
)
```
```

--------------------------------

### Manage Jobify Lifecycle with FastAPI Lifespan

Source: https://github.com/theseriff/jobify/blob/main/docs/integrations.md

Use the async context manager for automatic startup and shutdown of Jobify within your FastAPI application's lifespan. This is the recommended approach for managing the Jobify instance.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from jobify import Jobify

jobify_app = Jobify()

@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
    async with jobify_app:
        yield

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Run Linter and Formatter

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

Execute Ruff for code formatting and checking, along with Codespell, to ensure code quality.

```bash
just linter
```

--------------------------------

### Run Tests with Coverage

Source: https://github.com/theseriff/jobify/blob/main/docs/contributing.md

Execute the pytest suite and generate a coverage report to identify areas needing more tests.

```bash
just test-coverage-all
```

--------------------------------

### Nesting Routers and Naming Logic

Source: https://github.com/theseriff/jobify/blob/main/docs/router.md

Illustrates how routers can be nested to create complex hierarchies. The final task ID follows a pattern of `[parent_prefix].[sub_prefix]:[task_name]`, aiding in debugging and monitoring.

```python
# reports/router.py (prefix="reports")
# reports/daily.py  (prefix="daily", nested in reports)

@daily_router.task(name="generate")
def gen(): ...

# Resulting name: "reports.daily:generate"
```

--------------------------------

### Injecting Full JobContext

Source: https://github.com/theseriff/jobify/blob/main/docs/context.md

Use this when your task function needs access to multiple properties of the execution environment. Ensure Jobify and JobContext are imported.

```python
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.
```

--------------------------------

### Manually Control Jobify Lifecycle in FastAPI

Source: https://github.com/theseriff/jobify/blob/main/docs/integrations.md

Explicitly call startup and shutdown methods for granular control over the Jobify instance lifecycle within your FastAPI application. This is an alternative to the context manager approach.

```python
from fastapi import FastAPI
from jobify import Jobify

jobify_app = Jobify()

@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
    await jobify_app.startup()
    yield
    await jobify_app.shutdown()

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Configure Cron Schedule with Advanced Options

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Register a recurring schedule using the 'cron' parameter with advanced options like max_runs, misfire_policy, start_date, and arguments.

```python
from datetime import datetime

from jobify import Cron, MisfirePolicy


@app.task(
    cron=Cron(
        "0 18 * * 1-5",
        max_runs=100,
        max_failures=5,
        misfire_policy=MisfirePolicy.ALL,
        start_date=datetime(2027, 1, 1),
        args=("financial",),
        kwargs={"recipient": "admin@example.com"},
    )
)
def daily_report(report_type: str, recipient: str) -> None:
    ...
```

--------------------------------

### push

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Enqueue a task for immediate execution as soon as possible. Jobs enqueued with `push()` are persisted if the task is `durable=True`.

```APIDOC
## push (Immediate Execution)

Fastest way to enqueue a task for execution as soon as possible.

### Description
Enqueues a task for immediate execution. Jobs are persisted if the task is defined as durable.

### Example
```python
job = await my_task.push(*args, **kwargs)
await job.wait()
```

### Notes
- `push()` jobs are persisted if the task is `durable=True`.
- If the task is `durable=False`, jobs are not restored after a restart.
```

--------------------------------

### Configure Simple and Advanced Retries

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Set the number of retry attempts using 'retry=int' for simple retries, or use 'SmartRetry' for advanced backoff strategies and filtering.

```python
from jobify import SmartRetry

@app.task(retry=3) # Simple: 3 retries after first failure
def fragile_io() -> None:
    ...

@app.task(retry=SmartRetry(retries=5, initial_delay=2.0)) # Advanced
def custom_retry() -> None:
    ...
```

--------------------------------

### Enqueue Task for Immediate Execution

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

The `push` method enqueues a task for execution as soon as possible. Jobs enqueued with `push()` are persisted if the task is `durable=True`, but not restored after a restart if `durable=False`.

```python
job = await my_task.push(*args, **kwargs)
await job.wait()
```

--------------------------------

### Define Priority Levels for Tasks

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/queue.md

Illustrates defining distinct numeric priority bands for tasks to ensure consistent and explicit priority rules across the codebase. Includes critical, user flow, and background task priorities.

```python
PRIORITY_CRITICAL = 100
PRIORITY_USER_FLOW = 50
PRIORITY_BACKGROUND = 5

@app.task(metadata={"priority": PRIORITY_CRITICAL})
async def webhook_retry() -> None: ...

@app.task(metadata={"priority": PRIORITY_USER_FLOW})
async def user_notification() -> None: ...

@app.task(metadata={"priority": PRIORITY_BACKGROUND})
async def metrics_rollup() -> None: ...
```

--------------------------------

### Configure Global Exception Handlers

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Apply exception handlers to all tasks in the Jobify application by defining them in the Jobify constructor.

```python
app = Jobify(
    exception_handlers={
        TypeError: global_type_error_handler
    }
)
```

--------------------------------

### Concurrent Job Waits using asyncio.gather

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

A recommended pattern for concurrently waiting for multiple jobs to complete using `asyncio.gather` and `job.wait()`.

```python
jobs = [
    await add.schedule(1, 1).delay(0),
    await add.schedule(2, 2).delay(0),
]
await asyncio.gather(*(job.wait() for job in jobs))
```

--------------------------------

### Basic Task Retry

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/retry.md

Configure a task to retry a specified number of times after initial failure by passing an integer to the 'retry' parameter of the @app.task decorator.

```python
from jobify import Jobify

app = Jobify()

@app.task(retry=3)
def my_task():
    # Will be tried 4 times total: 1 original + 3 retries
    pass
```

--------------------------------

### Advanced Task Retry with SmartRetry

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/retry.md

Utilize the SmartRetry object for detailed control over retry behavior, including the number of retries, initial delay, maximum delay, backoff factor, jitter, and specific exceptions to include or exclude.

```python
from jobify import Jobify, SmartRetry

app = Jobify()

@app.task(
    retry=SmartRetry(
        retries=5,
        initial_delay=1.0,
        max_delay=300.0,
        backoff_factor=2.0,
        jitter=True,
        include_exceptions=(IOError, ConnectionError),
        exclude_exceptions=(ValueError,)
    )
)
def sensitive_task():
    # Task logic here
    pass
```

--------------------------------

### Configure Router Exception Handlers

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Define exception handlers for all tasks within a specific JobRouter. These handlers are applied before global handlers.

```python
router = JobRouter(
    exception_handlers={
        ValueError: router_value_error_handler
    }
)
```

--------------------------------

### Injecting Specifics from JobContext

Source: https://github.com/theseriff/jobify/blob/main/docs/context.md

Declare only the specific attributes needed by your task function for cleaner and more explicit dependencies. Ensure necessary types like Job, State, and Jobify are imported.

```python
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"]
```

--------------------------------

### Jobify with Disabled Persistence

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Configures Jobify to run without persistence, meaning jobs are not stored and will be lost on restart. This is useful for in-memory-only operations.

```python
from jobify import Jobify

app = Jobify(storage=False)  # in-memory only
```

--------------------------------

### Scheduling Operations Logging Middleware

Source: https://github.com/theseriff/jobify/blob/main/docs/app_settings.md

Customize behavior during job scheduling, like logging scheduling details. This middleware runs on operations like `.push()`, `.schedule()`, etc.

```python
import asyncio

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


class ScheduleLoggerMiddleware(BaseOuterMiddleware):
    async def __call__(self, call_next: CallNextOuter, context: OuterContext) -> asyncio.Handle:
        print(f"Scheduling {context.job.id} with trigger={context.trigger}")
        return await call_next(context)


app = Jobify(outer_middleware=[ScheduleLoggerMiddleware()])
```

--------------------------------

### Set Task Execution Timeout

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Configure the execution timeout in seconds using the 'timeout' parameter. Tasks exceeding this duration will be marked as timed out.

```python
@app.task(timeout=30.0)
async def slow_task() -> None:
    ...
```

--------------------------------

### Awaitable Shorthand for Job Completion and Result

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

A concise way to wait for a job and retrieve its result, equivalent to calling `await job.wait()` followed by `job.result()`.

```python
result = await job
```

--------------------------------

### cron

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Create or update a recurring dynamic cron job. The `job_id` is required. Setting `replace=True` updates an existing schedule, while `force=True` re-runs outer middleware even if the schedule is unchanged.

```APIDOC
## cron

Create/update a recurring dynamic cron job.

### Description
Schedules a recurring job using cron syntax. Requires a `job_id`. Supports replacing existing jobs and forcing middleware execution.

### Parameters

#### Request Body

- **cron** (string) - Required - Cron expression for scheduling.
- **job_id** (string) - Required - Unique identifier for the job.
- **replace** (boolean) - Optional - If `True`, updates an existing job with the same `job_id`. Defaults to `False`.
- **force** (boolean) - Optional - If `True`, forces outer middleware execution even if the schedule is unchanged. Defaults to `False`.

### Example
```python
await my_task.schedule(*args, **kwargs).cron(
    cron="*/5 * * * *",
    job_id="cleanup_dynamic",  # required
    replace=False,
    force=False,
)
```

### Notes
- Reusing an existing `job_id` with `replace=False` raises `DuplicateJobError`.
- `force=True` ensures outer middleware runs even when the schedule configuration is identical.
```

--------------------------------

### Recover Job with Handler Return Value

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Mark a job as SUCCESS and provide a result by returning a value from the exception handler. If no value is returned, the job is also marked as SUCCESS.

```python
async def recovery_handler(exc, ctx) -> str:
    return "fallback_value"  # Job status: SUCCESS
```

--------------------------------

### Abort Retries with Excluded Exceptions

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Configure `SmartRetry` to immediately fail a job without retries by specifying exceptions in `exclude_exceptions`. This is used for unrecoverable errors.

```python
from jobify import SmartRetry

# Define a custom fatal exception
class FatalError(Exception): ...

@app.task(
    retry=SmartRetry(retries=3, exclude_exceptions=(FatalError,))
)
async def my_task():
    raise FatalError("Unrecoverable") # Fails immediately, no retries
```

--------------------------------

### Schedule Task Execution at a Specific Datetime

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

The `at` method schedules a task to run at an exact `datetime`. You can specify `job_id`, `replace`, and `force` parameters. By default, reusing an existing `job_id` raises `DuplicateJobError` unless `replace=True`.

```python
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

run_at = datetime.now(tz=ZoneInfo("UTC")) + timedelta(minutes=10)
job = await my_task.schedule(*args, **kwargs).at(
    at=run_at,
    job_id="report_123",
    replace=False,
    force=False,
)
```

--------------------------------

### Aborting Retries with Excluded Exceptions

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/retry.md

Configure a task to immediately abort retries upon encountering specific fatal exceptions by using the 'exclude_exceptions' parameter within SmartRetry. This prevents unnecessary retries for errors that cannot be resolved by repetition.

```python
from jobify import Jobify, SmartRetry

class FatalError(Exception): ...

app = Jobify()

@app.task(
    retry=SmartRetry(retries=3, exclude_exceptions=(FatalError,))
)
def my_task():
    if not database_connected():
        raise FatalError("Cannot proceed") # No retries will be attempted
```

--------------------------------

### delay

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Schedule a task for execution after a specified delay in seconds. Supports optional `job_id` and `now` reference point, with `replace` parameter to update existing jobs.

```APIDOC
## delay

Schedule execution after a delay in seconds.

### Description
Schedules a task to run after a specified number of seconds. Allows for optional job ID and a reference point for time calculation. Existing jobs can be replaced.

### Parameters

#### Request Body

- **seconds** (float) - Required - The delay in seconds before execution.
- **job_id** (string | None) - Optional - A unique identifier for the job. Defaults to `None`.
- **now** (datetime | None) - Optional - A reference datetime, defaulting to the current time in UTC, used for calculating the delay. Defaults to `None`.
- **replace** (boolean) - Optional - If `True`, updates an existing job with the same `job_id`. Defaults to `False`.

### Example
```python
from datetime import datetime
from zoneinfo import ZoneInfo

job = await my_task.schedule(*args, **kwargs).delay(
    seconds=60,
    job_id="email_60s",
    now=datetime.now(tz=ZoneInfo("UTC")),  # optional reference point
    replace=False,
)
```

### Notes
- The `force` parameter is not available for `delay()`. Use `.at(..., force=True)` for force semantics when rescheduling at an absolute time.
```

--------------------------------

### Schedule Task Execution After a Delay

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Use the `delay` method to schedule a task for execution after a specified number of seconds. An optional `job_id` can be provided, and `replace=False` is the default behavior to prevent overwriting existing jobs. For absolute-time rescheduling with force semantics, use `.at(..., force=True)`.

```python
from datetime import datetime
from zoneinfo import ZoneInfo

job = await my_task.schedule(*args, **kwargs).delay(
    seconds=60,
    job_id="email_60s",
    now=datetime.now(tz=ZoneInfo("UTC")),  # optional reference point
    replace=False,
)
```

--------------------------------

### Waiting for Job Completion

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Waits for a job to reach a terminal state. This method does not raise exceptions directly; check job status afterward.

```python
await job.wait()
print(job.status)
```

--------------------------------

### Define Task-Specific Exception Handlers

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Map specific exceptions to handler functions using 'exception_handlers' for localized error recovery. This takes precedence over global handlers.

```python
from jobify import JobContext


def on_value_error(exc: Exception, context: JobContext) -> str:
    return "fallback"


@app.task(exception_handlers={ValueError: on_value_error})
def task_with_recovery() -> None:
    ...
```

--------------------------------

### Configure Task-Specific Exception Handlers

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

Assign exception handlers directly to a task using the `exception_handlers` argument in the task decorator. These handlers take precedence.

```python
@app.task(
    exception_handlers={
        TimeoutError: task_timeout_handler
    }
)
async def my_task(): ...
```

--------------------------------

### Set Explicit Task Name

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Use the 'name' parameter to assign a stable, explicit name to a task, which is useful for stable naming across refactors or deployments.

```python
@app.task(name="billing:close_invoices")
def close_invoices() -> None:
    ...
```

--------------------------------

### Setting Task Metadata

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Use the metadata parameter in the @app.task decorator to assign custom attributes to a task. This is useful for controlling task execution order with middleware like QueueMiddleware.

```python
from typing import Any
from jobify import app

@app.task(metadata={"priority": 100, "tenant": "acme"})
async def critical_pipeline() -> None:
    ... 
```

--------------------------------

### Schedule Recurring Cron Job

Source: https://github.com/theseriff/jobify/blob/main/docs/schedule.md

Use the `cron` method to create or update a recurring dynamic cron job. `job_id` is required. Setting `replace=False` with an existing `job_id` raises `DuplicateJobError`, while `replace=True` updates the existing schedule. `force=True` can be used to re-run outer middleware even if the schedule is unchanged.

```python
await my_task.schedule(*args, **kwargs).cron(
    cron="*/5 * * * *",
    job_id="cleanup_dynamic",  # required
    replace=False,
    force=False,
)
```

--------------------------------

### Disable Durability for High-Frequency Cron Jobs

Source: https://github.com/theseriff/jobify/blob/main/docs/task_settings.md

Set 'durable=False' for very high-frequency cron jobs to optimize by reducing storage churn and write load. Default is True.

```python
@app.task(cron="* * * * * * *", durable=False)
async def heartbeat() -> None:
    ...
```

--------------------------------

### Trigger Retries with Exception Handler

Source: https://github.com/theseriff/jobify/blob/main/docs/advanced_usage/exception_handlers.md

To enable retries for a task, re-raise the caught exception within the handler. This allows the RetryMiddleware to process the retry logic.

```python
async def my_handler(exc: Exception, context: JobContext):
    log.error("Retrying...")
    raise exc  # Re-raise to trigger retry logic
```

--------------------------------

### Checking if a Job is Done

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Returns a boolean indicating whether the job has reached a terminal state (e.g., SUCCESS, FAILED, CANCELLED).

```python
if job.is_done():
    print(job.status)
```

--------------------------------

### Retrieving Job Result

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Reads the result of a completed job. Raises JobFailedError for failed jobs or JobNotCompletedError if the job is not ready.

```python
await job.wait()
value = job.result()
```

--------------------------------

### Checking if a Job is a Cron Job

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Returns a boolean indicating if the job handle is associated with a cron schedule execution.

```python
if job.is_cron():
    print(job.cron_expression)
```

--------------------------------

### Cancelling a Scheduled Job

Source: https://github.com/theseriff/jobify/blob/main/docs/job.md

Cancels and unschedules a job that has been scheduled but not yet executed. This includes cleanup for durable schedules.

```python
job = await add.schedule(10, 20).delay(300)
await job.cancel()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.