Tasks

Background task scheduling for running periodic operations.

Overview

The tasks module provides background task execution with interval-based and cron-based scheduling. Tasks run while the server is running and are automatically started after startup handlers and stopped before shutdown handlers.

Task Decorator

Register tasks using the @app.task() decorator on your Xitzin application:

from xitzin import Xitzin

app = Xitzin()

@app.task(interval="1h")
async def hourly_cleanup():
    await cleanup_old_records()

@app.task(cron="0 2 * * *")
def daily_backup():
    backup_database()

See the Application reference for the Xitzin.task() decorator API.

BackgroundTask

Data class representing a registered background task.

BackgroundTask dataclass

BackgroundTask(
    handler: Callable[[], Any],
    interval: float | None,
    cron: str | None,
    name: str,
)

Configuration for a background task.

Helper Functions

parse_interval

parse_interval

parse_interval(interval: str | int | float) -> float

Parse interval string or int to seconds.

Parameters:

Name	Type	Description	Default
interval	str | int | float	Either an integer/float (seconds) or string like "1h", "30m", "1d"	required

Returns:

Type	Description
float	Interval in seconds

Raises:

Type	Description
ValueError	If format is invalid

Source code in src/xitzin/tasks.py

def parse_interval(interval: str | int | float) -> float:
    """Parse interval string or int to seconds.

    Args:
        interval: Either an integer/float (seconds) or string like "1h", "30m", "1d"

    Returns:
        Interval in seconds

    Raises:
        ValueError: If format is invalid
    """
    if isinstance(interval, (int, float)):
        if interval <= 0:
            raise ValueError("Interval must be positive")
        return float(interval)

    # Parse duration strings
    pattern = r"^(\d+)([smhd])$"
    match = re.match(pattern, interval.lower().strip())
    if not match:
        raise ValueError(
            f"Invalid interval format: {interval!r}. "
            "Use integer seconds or format like '1h', '30m', '1d'"
        )

    value, unit = match.groups()
    value = int(value)

    multipliers = {
        "s": 1,
        "m": 60,
        "h": 3600,
        "d": 86400,
    }

    return float(value * multipliers[unit])

run_interval_task

run_interval_task async

run_interval_task(task: BackgroundTask) -> None

Run a task on a fixed interval.

Parameters:

Name	Type	Description	Default
task	BackgroundTask	The task to run	required

Source code in src/xitzin/tasks.py

async def run_interval_task(task: BackgroundTask) -> None:
    """Run a task on a fixed interval.

    Args:
        task: The task to run
    """
    task_logger = logger.bind(task=task.name)
    task_logger.info("task_started", interval=task.interval)

    while True:
        try:
            # Wait first (standard behavior)
            await asyncio.sleep(task.interval)  # type: ignore[arg-type]

            # Execute handler
            await _execute_handler(task.handler)
            task_logger.debug("task_executed")

        except asyncio.CancelledError:
            task_logger.info("task_cancelled")
            raise
        except Exception as e:
            task_logger.error(
                "task_failed",
                error=str(e),
                error_type=type(e).__name__,
            )

run_cron_task

run_cron_task async

run_cron_task(task: BackgroundTask) -> None

Run a task on a cron schedule.

Parameters:

Name	Type	Description	Default
task	BackgroundTask	The task to run	required

Source code in src/xitzin/tasks.py

async def run_cron_task(task: BackgroundTask) -> None:
    """Run a task on a cron schedule.

    Args:
        task: The task to run
    """
    from croniter import croniter

    task_logger = logger.bind(task=task.name)
    task_logger.info("task_started", cron=task.cron)

    try:
        cron_iter = croniter(task.cron, datetime.now(timezone.utc))
    except Exception as e:
        task_logger.error(
            "task_cron_invalid",
            cron=task.cron,
            error=str(e),
            error_type=type(e).__name__,
        )
        raise

    while True:
        try:
            # Calculate next run time
            next_run = cron_iter.get_next(datetime)
            now = datetime.now(timezone.utc)

            # Handle timezone-naive datetime from croniter
            if next_run.tzinfo is None:
                next_run = next_run.replace(tzinfo=timezone.utc)

            sleep_seconds = (next_run - now).total_seconds()

            if sleep_seconds > 0:
                task_logger.debug("task_waiting", next_run=next_run.isoformat())
                await asyncio.sleep(sleep_seconds)

            # Execute handler
            await _execute_handler(task.handler)
            task_logger.debug("task_executed")

        except asyncio.CancelledError:
            task_logger.info("task_cancelled")
            raise
        except Exception as e:
            task_logger.error(
                "task_failed",
                error=str(e),
                error_type=type(e).__name__,
            )

Exceptions

TaskConfigurationError

TaskConfigurationError

Bases: Exception

Raised when a background task is misconfigured.

This typically indicates mutually exclusive parameters were provided, or a required optional dependency is missing.

Example

@app.task() # Error: neither interval nor cron provided def my_task(): pass

@app.task(interval="1h", cron="* * * * *") # Error: both provided def my_task(): pass

Installation

Cron-based tasks require the croniter package:

pip install 'xitzin[tasks]'

Interval-based tasks work without any additional dependencies.