API Reference¶

@dataclass(slots=True, kw_only=True, order=True)
class Cron:
    """Configuration for cron-based job scheduling.

    Attributes:
        expression: The crontab-formatted expression.
        max_runs: Maximum number of times the job can be triggered.
            Defaults to infinity.
        max_failures: Maximum number of consecutive failures before disabling the job.
            Defaults to 10.
        misfire_policy: Policy to handle missed job executions.
            Defaults to MisfirePolicy.ONCE.
        start_date: Optional datetime when the cron job becomes active.
        args: Positional arguments to pass to the job.
        kwargs: Keyword arguments to pass to the job.

    """

    expression: str = field(kw_only=False)
    max_runs: int = INFINITY
    max_failures: int = 10
    misfire_policy: MisfirePolicy | GracePolicy = MisfirePolicy.ONCE
    start_date: datetime | None = None
    args: tuple[Any, ...] = ()
    kwargs: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if self.max_failures < 1:
            msg = "max_cron_failures must be >= 1. Use 1 for 'stop on first error'."
            raise ValueError(msg)

class CronContext(Generic[ReturnT]):
    """Holds configuration and state for a cron-based job.

    Attributes:
        cron: The cron configuration.
        cron_parser: The parser used to calculate next run times.
        failure_count: Number of consecutive failures.
        job: The associated job instance.
        offset: The base datetime from which the next run is calculated.
        run_count: Number of times this job has been executed.

    """

    __slots__: tuple[str, ...] = (
        "cron",
        "cron_parser",
        "failure_count",
        "job",
        "offset",
        "run_count",
    )

    def __init__(  # noqa: PLR0913
        self,
        *,
        cron: Cron,
        cron_parser: CronParser,
        failure_count: int = 0,
        job: Job[ReturnT],
        offset: datetime,
        run_count: int,
    ) -> None:
        """Initialize the CronContext.

        Args:
            cron: The cron configuration.
            cron_parser: The parser used to calculate next run times.
            failure_count: Initial failure count.
            job: The associated job instance.
            offset: The base datetime.
            run_count: Initial run count.

        """
        self.cron = cron
        self.cron_parser = cron_parser
        self.failure_count = failure_count
        self.job = job
        self.offset = offset
        self.run_count = run_count

    def is_run_exceeded_by_limit(self) -> bool:
        """Check if the maximum number of runs has been exceeded.

        Returns:
            True if the run limit is reached, False otherwise.

        """
        if self.cron.max_runs == INFINITY:
            return False
        return self.run_count >= self.cron.max_runs

    def is_failure_allowed_by_limit(self) -> bool:
        """Check if the job can still fail based on the allowed limit.

        Returns:
            True if the failure limit has not been reached, False otherwise.

        """
        return self.failure_count < self.cron.max_failures

class GracePolicy(NamedTuple):
    """Grace period policy for handling misfired jobs.

    Attributes:
        value: The duration of the grace period.

    """

    value: timedelta

    @override
    def __str__(self) -> str:
        return f"<{type(self).__name__} value={self.value}>"

class Job(Generic[ReturnT]):
    """Represents a scheduled job.

    Attributes:
        id: Unique identifier for the job.
        status: Current status of the job.
        exec_at: The scheduled execution time.
        exception: The exception raised, if any.

    """

    __slots__: tuple[str, ...] = (
        "_cron_context",
        "_event",
        "_handle",
        "_result",
        "_storage",
        "_unregister_hook",
        "exception",
        "exec_at",
        "id",
        "status",
    )

    def __init__(
        self,
        *,
        job_id: str,
        storage: Storage,
        exec_at: datetime,
        unregister_hook: Callable[[str], None],
        job_status: JobStatus = JobStatus.PENDING,
    ) -> None:
        """Initialize the Job.

        Args:
            job_id: Unique identifier for the job.
            storage: Storage backend.
            exec_at: The scheduled execution time.
            unregister_hook: Callback to unregister the job.
            job_status: Initial status of the job.

        """
        self._unregister_hook = unregister_hook
        self._event = asyncio.Event()
        self._result: ReturnT = UNSET
        self._storage = storage
        self._handle: asyncio.Handle | None = None
        self._cron_context: CronContext[ReturnT] | None = None
        self.id = job_id
        self.status = job_status
        self.exception: Exception | None = None
        self.exec_at = exec_at

        self._event.set()

    @property
    def cron_expression(self) -> str | None:
        """Return the cron expression if this is a cron job, else None."""
        if self._cron_context is not None:
            return self._cron_context.cron.expression
        return None

    @override
    def __repr__(self) -> str:
        if self.cron_expression is not None:
            cron_info = f", cron={self.cron_expression!r}"
        else:
            cron_info = ""
        return (
            f"<{type(self).__name__} "
            f"id={self.id!r}, "
            f"status={self.status.value!r}, "
            f"exec_at={self.exec_at.isoformat()!r}"
            f"{cron_info}>"
        )

    def __await__(self) -> Generator[object, None, ReturnT]:
        async def _await() -> ReturnT:
            await self.wait()
            return self.result()

        return _await().__await__()

    def bind_handle(self, handle: asyncio.Handle) -> None:
        """Bind an asyncio handle to the job.

        Args:
            handle: The handle to bind.

        """
        self._handle = handle

    def bind_cron_context(self, ctx: CronContext[ReturnT]) -> None:
        """Bind a cron context to the job.

        Args:
            ctx: The cron context to bind.

        """
        self._cron_context = ctx

    def result(self) -> ReturnT:
        """Return the result of the job.

        Returns:
            The job result.

        Raises:
            JobFailedError: If the job failed.
            JobNotCompletedError: If the job is not yet completed.

        """
        if self.status is JobStatus.SUCCESS or self._result is not UNSET:
            return self._result
        if self.status is JobStatus.FAILED:
            raise JobFailedError(
                self.id,
                reason=str(self.exception),
            ) from self.exception
        raise JobNotCompletedError

    def set_result(self, val: ReturnT, *, status: JobStatus) -> None:
        """Set the result of the job.

        Args:
            val: The job result.
            status: The status to set.

        """
        self._result = val
        self.status = status

    def set_exception(self, exc: Exception, *, status: JobStatus) -> None:
        """Set the exception for the job.

        Args:
            exc: The exception to set.
            status: The status to set.

        """
        self.exception = exc
        self.status = status

    def update(self, *, exec_at: datetime, status: JobStatus) -> None:
        """Update the job schedule and status.

        Args:
            exec_at: The new scheduled time.
            status: The new status.

        """
        self._event = asyncio.Event()
        self.status = status
        self.exec_at = exec_at

    def is_done(self) -> bool:
        """Check if the job is done.

        Returns:
            True if the job is done, False otherwise.

        """
        return self._event.is_set()

    def is_cron(self) -> bool:
        """Check if this is a cron job.

        Returns:
            True if it's a cron job, False otherwise.

        """
        return self._cron_context is not None

    def is_reschedulable(self) -> bool:
        """Check if the job can be rescheduled.

        Returns:
            True if reschedulable, False otherwise.

        """
        return self.status not in (
            JobStatus.PERMANENTLY_FAILED,
            JobStatus.CANCELLED,
        )

    async def wait(self) -> None:
        """Wait until the job is done.

        If the job is already completed, this method returns immediately.
        Safe for concurrent use by multiple coroutines.
        """
        await self._event.wait()

    async def cancel(self) -> None:
        """Cancel the job."""
        self.status = JobStatus.CANCELLED
        self._cancel()
        await self._storage.delete_schedule(self.id)

    def _cancel(self) -> None:
        """Handle cancellation internally."""
        self._event.set()
        self._unregister_hook(self.id)
        if self._handle is not None:
            self._handle.cancel()

class JobContext(NamedTuple):
    """Context object injected into jobs at runtime.

    Provides access to job execution context, allowing jobs to inspect
    their configuration, state, and scheduling builder.

    Attributes:
        job: The job instance.
        state: The global application state.
        runnable: The runnable component.
        request_state: State specific to the current request.
        route_options: Configuration options for the route.
        jobify_config: Global Jobify configuration.
        schedule_builder: Builder used to construct the schedule.

    """

    job: Job[Any]
    state: State
    runnable: Runnable[Any]
    request_state: RequestState
    route_options: RouteOptions
    jobify_config: JobifyConfiguration
    schedule_builder: ScheduleBuilder[Any]

class NodeRouter(Router):
    """Router for organizing jobs and sub-routers.

    `NodeRouter` is a container for routes and sub-routers, providing
    a way to structure large applications by grouping related jobs together.

    Args:
        state: Optional initial state for the router.
        prefix: Optional prefix for route names.
        lifespan: Optional lifespan handler for the router.
        middleware: Middleware to apply to jobs in this router.
        outer_middleware: Middleware to apply to scheduling process.
        exception_handlers: Exception handlers for jobs.
        route_class: Class to use for creating new routes.

    """

    def __init__(  # noqa: PLR0913
        self,
        *,
        state: State | None = None,
        prefix: str | None = None,
        lifespan: Lifespan[NodeRouter_co] | None = None,
        middleware: Sequence[BaseMiddleware] | None = None,
        outer_middleware: Sequence[BaseOuterMiddleware] | None = None,
        exception_handlers: MappingExceptionHandlers | None = None,
        route_class: type[NodeRoute[..., Any]] = NodeRoute,
    ) -> None:
        super().__init__(prefix=prefix)
        self._registrator: NodeRegistrator = NodeRegistrator(
            state=state,
            lifespan=lifespan,
            middleware=middleware,
            outer_middleware=outer_middleware,
            exception_handlers=exception_handlers,
            route_class=route_class,
        )
        self.state: State = self._registrator.state
        self.exception_handlers: ExceptionHandlers = (
            self._registrator._exception_handlers
        )

    @property
    @override
    def task(self) -> NodeRegistrator:
        return self._registrator

    @property
    @override
    def routes(self) -> Iterator[NodeRoute[..., Any]]:
        yield from self.task._routes.values()

    @property
    @override
    def sub_routers(self) -> list[NodeRouter]:
        return cast("list[NodeRouter]", self._sub_routers)

@unique
class JobStatus(str, Enum):
    """The status of a job.

    Attributes:
        PENDING: Job is waiting to be processed.
        SCHEDULED: Job is scheduled for future execution.
        RUNNING: Job is currently executing.
        CANCELLED: Job was cancelled.
        SUCCESS: Job completed successfully.
        FAILED: Job failed execution.
        TIMEOUT: Job timed out.
        PERMANENTLY_FAILED: Job failed and exhausted all retry attempts.

    """

    PENDING = "pending"
    SCHEDULED = "scheduled"
    RUNNING = "running"
    CANCELLED = "cancelled"
    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"
    PERMANENTLY_FAILED = "permanently_failed"

class Jobify(RootRouter):
    """Jobify is the main app for scheduling and managing background jobs.

    It provides a flexible and extensible framework for defining, running,
    and persisting jobs, supporting various executors, middleware, and
    serialization options.

    Attributes:
        configs: Configuration object for the application.
        plugins: Set of registered plugins.

    """

    def __init__(  # noqa: PLR0913
        self,
        *,
        tz: ZoneInfo | None = None,
        state: State | None = None,
        dumper: Dumper | None = None,
        loader: Loader | None = None,
        storage: Storage | Literal[False] = UNSET,
        lifespan: Lifespan[AppT] | None = None,
        serializer: Serializer | None = None,
        middleware: Sequence[BaseMiddleware] | None = None,
        outer_middleware: Sequence[BaseOuterMiddleware] | None = None,
        cron_factory: CronFactory = create_crontab,
        loop_factory: LoopFactory = asyncio.get_running_loop,
        exception_handlers: MappingExceptionHandlers | None = None,
        threadpool_executor: ThreadPoolExecutor | None = None,
        processpool_executor: ProcessPoolExecutor | None = None,
        route_class: type[RootRoute[..., Any]] = RootRoute,
        plugins: Sequence[Plugin] = (),
        uuid_generator: UUIDGenerator = uuid.uuid4,
    ) -> None:
        """Initialize a `Jobify` instance.

        Args:
            tz: Timezone info, defaults to UTC.
            state: Optional initial application state.
            dumper: Dumper for serialization.
            loader: Loader for deserialization.
            storage: Storage backend, defaults to SQLite if not provided.
            lifespan: Optional lifespan handler.
            serializer: Serializer for job messages.
            middleware: List of middleware components.
            outer_middleware: List of outer middleware components.
            cron_factory: Factory function for cron parsing.
            loop_factory: Factory function for the event loop.
            exception_handlers: Mapping of exception types to handlers.
            threadpool_executor: Executor for thread-based jobs.
            processpool_executor: Executor for process-based jobs.
            route_class: Class to use for root routes.
            plugins: List of plugins to register.
            uuid_generator: uuid generator factory.

        """
        getloop = cache_result(loop_factory)
        tz = tz or ZoneInfo("UTC")

        if storage is False:
            storage = DummyStorage()
        elif storage is UNSET:
            storage = SQLiteStorage()

        if isinstance(storage, SQLiteStorage):
            storage.getloop = getloop
            storage.threadpool = threadpool_executor
            storage.tz = tz

        system_types = (
            Message,
            Cron,
            PushArguments,
            AtArguments,
            CronArguments,
            MisfirePolicy,
            GracePolicy,
        )
        if serializer is None:
            serializer = (
                ExtendedJSONSerializer(system_types)
                if dumper is None and loader is None
                else JSONSerializer()
            )
        elif isinstance(serializer, ExtendedJSONSerializer):
            serializer.add_system_types(system_types)

        if dumper is None:
            dumper = DummyDumper()
        if loader is None:
            loader = DummyLoader()

        self.configs: JobifyConfiguration = JobifyConfiguration(
            tz=tz,
            dumper=dumper,
            loader=loader,
            storage=storage,
            getloop=getloop,
            serializer=serializer,
            worker_pools=WorkerPools(
                _processpool=processpool_executor,
                threadpool=threadpool_executor,
            ),
            cron_factory=cron_factory,
            uuid_generator=uuid_generator,
        )
        idle_event = asyncio.Event()
        idle_event.set()
        super().__init__(
            state=state,
            lifespan=lifespan,
            middleware=middleware,
            outer_middleware=outer_middleware,
            task_tracker=TaskTracker(
                pending_jobs={},
                pending_tasks={},
                idle_event=idle_event,
            ),
            jobify_config=self.configs,
            exception_handlers=exception_handlers,
            route_class=route_class,
        )
        self._captured_signals: list[int] = []
        self.plugins: set[Plugin] = set(plugins)

    def add_plugin(self, plug: Plugin, /) -> None:
        """Register a plugin instance for application lifecycle hooks."""
        self.plugins.add(plug)

    def find_job(self, id_: str, /) -> Job[ReturnT] | None:
        """Find an active job by its ID.

        Args:
            id_: Unique identifier of the job.

        Returns:
            The `Job` instance if it's currently pending or running,
            otherwise `None`.

        """
        return self.task._task_tracker.pending_jobs.get(id_)

    def get_active_jobs(self) -> list[Job[Any]]:
        """Return a list of all currently active jobs."""
        return list(self.task._task_tracker.pending_jobs.values())

    async def __aenter__(self) -> Self:
        """Enter the Jobify context manager.

        Returns:
            The initialized Jobify instance ready for use.

        Raises:
            Exception: Any exception raised by `startup()` method.

        """
        await self.startup()
        return self

    async def startup(self) -> None:
        """Initialize the Jobify application.

        This method:
        1. Marks the application as started
        2. Propagates startup events to all routers and their registrators
        3. Schedules any pending cron jobs

        Raises:
            RuntimeError: If application startup fails due to configuration issues or
                router initialization errors.

        """

        def chain_middlewares() -> Iterator[BaseMiddleware]:
            for router in self.chain_tail:
                yield from router.task._middleware

        for mid in chain_middlewares():
            if isinstance(mid, Plugin):
                self.add_plugin(mid)

        for plug in self.plugins:
            await plug.startup(self)

        self.propagate_real_routes()

        self.configs.app_started = True

        await self.configs.storage.startup()
        await self._propagate_startup(self)
        await self._restore_schedules()

        logger.info("Jobify startup complete. Ready to schedule jobs.")

    async def _restore_schedules(self) -> None:
        """Restore schedules.

        Phase 1: Synchronize Declarative Cron Jobs (Code and Database)
            - If there are changes or new jobs, run the full process
              (middleware and database persistence).
            - If no changes, silently restore jobs in memory.
        Phase 2: Restore Imperative Jobs (Database Only)
            - Restore valid jobs and delete obsolete ones from the database.
        """
        crons_def: CronsDefinition = self.task.state.pop(CRONS_DEF_KEY, {})
        schedules = await self.configs.storage.get_schedules()

        to_delete: list[str] = []
        processed_jobs: set[str] = set()
        db_map = {sch.job_id: sch for sch in schedules}

        # --- PHASE 1: Declarative Crons ---
        for job_id, (rout, cron_def) in crons_def.items():
            processed_jobs.add(job_id)
            builder = rout.schedule(*cron_def.args, **cron_def.kwargs)

            if (sch_in_db := db_map.get(job_id)) is not None:
                restored = self._restore_job_from_storage(sch_in_db)
                if (
                    restored is not None
                    and (msg := restored[1])
                    and isinstance(msg.trigger, CronArguments)
                ):
                    self._start_restored_job_in_memory(
                        restored[0],
                        msg.trigger,
                        sch_in_db.next_run_at,
                    )
                    if msg.trigger.cron == cron_def:
                        continue

            await builder.cron(cron_def, job_id=job_id, replace=True, force=True)

        # --- PHASE 2: Dynamic/Imperative Jobs & Cleanup ---
        for job_id, sch in db_map.items():
            if job_id in processed_jobs:
                continue

            if job_id.endswith(PATCH_CRON_DEF_ID):
                to_delete.append(job_id)
                continue

            if (restored := self._restore_job_from_storage(sch)) is not None:
                builder, msg, _ = restored
                self._start_restored_job_in_memory(
                    builder,
                    msg.trigger,
                    sch.next_run_at,
                )
                continue

            to_delete.append(job_id)

        if to_delete:
            logger.info("Cleaning up %d obsolete/broken jobs", len(to_delete))
            await self.configs.storage.delete_schedule_many(to_delete)

    def _restore_job_from_storage(
        self,
        sch: ScheduledJob,
    ) -> tuple[ScheduleBuilder[Any], Message, inspect.BoundArguments] | None:
        """Safely deserializes and binds a job from storage.

        Returns:
            Tuple (Builder, Message, BoundArguments) or None if failed.

        """
        route = self.task._routes.get(sch.name)
        if not route or route.options.get("durable") is False:
            return None
        try:
            raw_msg = self.configs.serializer.loadb(sch.message)
            msg = self.configs.loader.load(raw_msg, Message)

            for name, raw_arg in msg.arguments.items():
                param_type = route.func_spec.type_params[name]
                msg.arguments[name] = self.configs.loader.load(
                    raw_arg,
                    param_type,
                )
            bound = route.func_spec.signature.bind(**msg.arguments)
        except (KeyError, TypeError, ValueError) as exc:
            logger.warning(
                "Failed to revive job %s (%s). Reason: %s",
                sch.job_id,
                sch.name,
                exc,
            )
            return None
        else:
            builder = route.create_builder(bound)
            return builder, msg, bound

    def _start_restored_job_in_memory(
        self,
        builder: ScheduleBuilder[Any],
        trigger: Triggers,
        db_next_run_at: datetime,
    ) -> None:
        """Start the internal timer for a restored job.

        IMPORTANT: This bypasses middleware and database persistence.
        It handles misfire policy to ensure we do not execute old jobs
        incorrectly.
        """
        match trigger:
            case PushArguments():
                builder._push(trigger.job_id, exec_at=db_next_run_at)
            case AtArguments():
                builder._at(trigger.at, trigger.job_id)
            case CronArguments():
                parser = self.configs.cron_factory(trigger.cron.expression)
                next_run_at = handle_misfire_policy(
                    cron_parser=parser,
                    next_run_at=db_next_run_at,
                    real_now=builder.now(),
                    policy=trigger.cron.misfire_policy,
                )
                builder._cron(
                    cron=trigger.cron,
                    job_id=trigger.job_id,
                    offset=trigger.offset,
                    next_run_at=next_run_at,
                    cron_parser=parser,
                    run_count=trigger.run_count,
                )

    async def __aexit__(
        self,
        _exc_type: type[BaseException] | None = None,
        _exc_val: BaseException | None = None,
        _exc_tb: TracebackType | None = None,
    ) -> None:
        """Exit the Jobify context manager.

        Note:
            This method ensures proper shutdown regardless of whether an
            exception occurred in the managed context. The exception parameters
            are ignored as shutdown should proceed even if the context failed.

        """
        await self.shutdown()

    async def shutdown(self) -> None:
        """Gracefully shut down the Jobify application.

        This method performs a structured shutdown:
        1. Marks the application as stopped (`app_started = False`).
        2. Propagates shutdown events to all routers/components.
        3. Cancels all scheduled future jobs in the registry
           (`_jobs_registry`).
        4. Closes the jobify configuration (e.g., stopping the internal
           scheduler).
        5. Cancels all currently running tasks (in `_tasks_registry`), waits
           for their completion, and explicitly clears the task registry.

        Note:
            The method uses `return_exceptions=True` when gathering cancelled
            tasks to prevent shutdown from being interrupted by task exception.

        """
        for plug in self.plugins:
            await plug.shutdown()

        self.configs.app_started = False

        if jobs := tuple(self.task._task_tracker.pending_jobs.values()):
            for job in jobs:
                job._cancel()

        if tasks := self.task._task_tracker.pending_tasks.values():  # pragma: no cover
            for task in tuple(tasks):
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)

        self.configs.worker_pools.close()
        await self._propagate_shutdown()
        await self.configs.storage.shutdown()
        self._raise_captured_signals()

        logger.info("Jobify shutdown complete.")

    def _raise_captured_signals(self) -> None:
        # If we did gracefully shut down due to a signal, try to
        # trigger the expected behaviour now; multiple signals would be
        # done LIFO, see https://stackoverflow.com/questions/48434964
        for captured_signal in reversed(self._captured_signals):
            signal.raise_signal(captured_signal)

    async def wait_all(self, timeout: float | None = None) -> None:
        """Wait for all currently scheduled jobs to complete.

        This method waits until all jobs currently registered have finished
        executing (with statuses of SUCCESS, FAILED, or TIMEOUT). This is
        useful in situations where it's important to ensure that background
        tasks have completed before moving on.

        The method sets an internal event when both conditions are met:
        1. No jobs remain in the jobs registry (`_jobs_registry`)

        Args:
            timeout (optional): The maximum time in seconds to wait for the
                jobs to complete. If not specified, the default value of `None`
                will be used, which means the job will wait indefinitely. If a
                timeout is specified and it is reached, the method will raise
                an `asyncio.TimeoutError`.

        """
        idle_event = self.task._task_tracker.idle_event
        with self._capture_signals():
            await asyncio.wait_for(idle_event.wait(), timeout=timeout)

    @contextmanager
    def _capture_signals(self) -> Generator[None]:
        handled_signals = [
            signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
            signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.
        ]
        if sys.platform == "win32":  # pragma: no cover
            handled_signals.append(
                signal.SIGBREAK  # Windows signal 21. Sent by Ctrl+Break.
            )

        # Signals can only be listened to from the main thread.
        if threading.current_thread() is not threading.main_thread():
            yield
            return
        # always use signal.signal, even if loop.add_signal_handler is
        # available this allows to restore previous signal handlers later on
        original_handlers = {
            sig: signal.signal(sig, self._handle_exit) for sig in handled_signals
        }
        try:
            yield
        finally:
            for sig, handler in original_handlers.items():
                signal.signal(sig, handler)

    def _handle_exit(self, sig: int, _: FrameType | None) -> None:
        self._captured_signals.append(sig)
        loop = self.configs.getloop()
        idle_event = self.task._task_tracker.idle_event
        loop.call_soon_threadsafe(idle_event.set)