Scheduling Jobs¶

You can schedule tasks in two main ways: using recurring cron expressions or dynamically at runtime.

Cron Expressions¶

Jobify uses the crontab library to parse and schedule jobs from cron expressions. This provides a flexible and powerful way to define recurring tasks.

Syntax¶

A cron expression is a string of fields that describe a schedule. The crontab library supports standard 5-field cron expressions, but also expressions with 6 or 7 fields for second-level precision and year specification.

The general representation is:

<seconds> <minutes> <hours> <day_of_month> <month> <day_of_week> <year>

Expression Formats¶

The crontab library supports multiple expression formats, which are interpreted as follows:

• 5 fields: MINUTES HOURS DAY_OF_MONTH MONTH DAY_OF_WEEK

  • Example: * * * * * - runs every minute.

• 6 fields: MINUTES HOURS DAY_OF_MONTH MONTH DAY_OF_WEEK YEAR

  • Example: 0 9 * * 1 2024 - runs at 9:00 AM every Monday in 2024.

• 7 fields: SECONDS MINUTES HOURS DAY_OF_MONTH MONTH DAY_OF_WEEK YEAR

  • This is the format you must use for second-level precision.
  • Example: */15 * * * * * * - runs every 15 seconds.

Fields¶

Special Characters¶

• * (Asterisk): Selects all possible values for a field. For example, * in the minutes field means "every minute".
• , (Comma): Used to specify a list of values. For example, "1, 5, 10" in the "minutes" field means "at minutes 1, 5 and 10".
• - (Dash): Specifies a range of values. For example, 1-5 in the day_of_week field means "Monday to Friday."
• / (Slash): Specifies a step value. For example, /15 in the seconds field means "every 15 seconds."
• ? (Question Mark): Used in day_of_month or day_of_week to indicate "no specific value", which is useful when you want to specify one of these fields but not the other.
• L (Last): has different meanings depending on the context. In day_of_month it refers to the last day of the month, while in day_of_week it means the last day of the week, which is Saturday.

Predefined Expressions (Aliases)¶

You can also use one of the following predefined cron expressions:

For more detailed information, please refer to the official crontab library documentation.

The Cron Object¶

You can use the Cron class to have more control over the scheduling of the job (e.g., limiting the number of runs or handling misfires).

from jobify import Cron, MisfirePolicy
from datetime import datetime

Cron(
    "0 18 * * 1-5",
    max_runs=100,
    max_failures=5,
    misfire_policy=MisfirePolicy.ALL,
    start_date=datetime(2027, 1, 1), # The task will start on January 1, 2027.
)

The Cron class has the following properties:

• expression (str): The cron expression as a string.
• max_runs (int, default: INFINITY (-1)): The maximum number of times a job can run. The run count is persisted in the database, so it survives application restarts. When the limit is reached, the job is automatically removed from the schedule.
• max_failures (int, default: 10): The maximum number of consecutive failed attempts allowed before a job is permanently stopped and no longer scheduled. This value must be greater than or equal to 1.
• misfire_policy (MisfirePolicy | GracePolicy, default: MisfirePolicy.ONCE): Determines how to deal with missed schedules (for example, if the application is unavailable).
  • MisfirePolicy.ALL: Run all missed executions immediately.
  • MisfirePolicy.SKIP: If there were missed executions, then skip them.
  • MisfirePolicy.ONCE: If there were missed executions, run only once.
  • MisfirePolicy.GRACE(timedelta): If the missed schedule is within the specified grace period, please start it immediately.
• start_date (datetime | None, default: None): The scheduled start time is used to anchor the first execution of the job to a specific datetime.

Dynamic Scheduling¶

In addition to cron jobs that are defined at the application level, users can also schedule tasks to run at a specific time or after a delay. They can even dynamically create new cron schedules. This is useful for one-time tasks or tasks that are triggered by the application's logic.

To dynamically schedule a task, you need to create a ScheduleBuilder. You can do this by calling the .schedule() method on the task function. The arguments that you pass to .schedule() will be used when the task runs.

push (Immediate Execution)¶

The .push() method is the fastest way to offload a task to the background. It schedules the task to run as soon as possible, similar to the .delay(0), but more concise.

Unlike the .schedule() method, which requires you to create the schedule first, the .push() method accepts your function arguments directly.

# Immediately schedules the job for execution.
await my_task.push(*args, **kwargs)

example:

import asyncio
from typing import Any

from jobify import Jobify

app = Jobify()


@app.task(durable=False, retry=3)
def process_data(data: dict[str, Any]) -> None:
    print(f"Processing: {data}")


async def main() -> None:
    async with app:
        # Offload execution immediately
        job = await process_data.push({"id": 1, "value": "test"})

        # You can still wait the result for it if needed
        await job.wait()


asyncio.run(main())

cron¶

To dynamically schedule a recurring task using a cron expression, use the .cron() method.

# Schedules a recurring task.
schedule(*args, **kwargs).cron(
    cron: str | Cron, # The cron expression or `Cron` object.
    *,
    job_id: str, # Required unique identifier for the job.
    replace: bool = False, # If True, updates the existing job.
    force: bool = False, # Force scheduling even if the job already exists with the same parameters.
)

example:

import asyncio
from jobify import Jobify

app = Jobify()

@app.task
def cleanup_logs() -> None:
    print("Cleaning up logs...")

@app.task(cron="@daily")
def daily_report() -> None:
    print("Generating daily report...")

async def main() -> None:
    async with app:
        # Schedule cleanup every 5 minutes
        job = await cleanup_logs.schedule().cron(
            cron="*/5 * * * *",
            job_id="cleanup_task_dynamic",
        )
        # Keep the app running...
        await app.wait_all()

asyncio.run(main())

delay¶

To run a task after a specified delay, use the delay() method of the builder.

# Schedules the task to run after a specified number of seconds.
schedule(*args, **kwargs).delay(
    seconds: float, # The delay in seconds.
    *,
    job_id: str | None = None, # Optional unique identifier for the job.
    now: datetime | None = None, # Optional reference datetime.
    replace: bool = False, # If True, updates the existing job.
    force: bool = False, # Force scheduling even if the job already exists with the same parameters.
)

import asyncio
from jobify import Jobify

app = Jobify()

@app.task
def send_email(to: str, subject: str) -> None:
    print(f"Sending email to {to} about {subject}")

async def main() -> None:
    async with app:
        # Schedule the email to be sent in 60 seconds
        job = await send_email.schedule(
            to="user@example.com",
            subject="Hello",
        ).delay(seconds=60)
        await job.wait()

asyncio.run(main())

at¶

To run a task at a specific datetime, use the .at() method.

# Schedules the task to run at the specified `datetime`.
schedule(*args, **kwargs).at(
    at: datetime, # The execution time.
    *,
    job_id: str | None = None, # Optional unique identifier for the job.
    replace: bool = False, # If True, updates the existing job.
    force: bool = False, # Force scheduling even if the job already exists with the same parameters.
)

example:

import asyncio
from datetime import datetime, timedelta
from jobify import Jobify

app = Jobify()

@app.task
def generate_report(report_id: int) -> None:
    print(f"Generating report {report_id}")

async def main() -> None:
    async with app:
        # Schedule the report to be generated 10 minutes from now
        run_time = datetime.now(app.configs.tz) + timedelta(minutes=10)
        job = await generate_report.schedule(report_id=123).at(at=run_time)
        # Keep the app running to allow the job to execute
        await job.wait()

asyncio.run(main())