`TaskCron` model reference¶

sheppy.models.TaskCron ¶

Bases: BaseModel

A cron definition that creates tasks on a schedule.

ATTRIBUTE	DESCRIPTION
`id`	Unique identifier for the cron definition. TYPE: `UUID`
`expression`	Cron expression defining the schedule, e.g. "/5 * * " for every 5 minutes. TYPE:* `str`
`spec`	Task specification TYPE: `Spec`
`config`	Task configuration TYPE: `Config`

Note

You should not create TaskCron instances directly. Instead, use the add_cron method of the Queue class to create a cron definition.
args and kwargs in spec must be JSON serializable.

Example

from sheppy import Queue, task

q = Queue(...)

@task
def say_hello(to: str) -> str:
    s = f"Hello, {to}!"
    print(s)
    return s

# add_cron returns bool indicating success
success = await q.add_cron(say_hello("World"), "*/5 * * * *")
assert success is True

# retrieve all cron jobs
crons = await q.get_crons()
for cron in crons:
    print(cron.id)  # UUID of the cron definition
    print(cron.expression)  # "*/5 * * * *"
    print(cron.spec.func)  # "my_module:say_hello"
    print(cron.spec.args)  # ["World"]

id ¶

id: UUID = Field(default_factory=uuid4)

UUID: Unique identifier for the cron definition.

expression ¶

expression: CronExpression

str: Cron expression defining the schedule, e.g. "*/5 * * * *" for every 5 minutes.

spec ¶

spec: Spec

Task specification

config ¶

config: Config

Task configuration

deterministic_id ¶

deterministic_id: UUID

Deterministic UUID to prevent duplicated cron definitions.

This property generates a deterministic UUID for the cron definition based on its spec, config, and expression. This ensures that identical cron definitions always have the same UUID, preventing duplicates.

RETURNS	DESCRIPTION
`UUID`	A deterministic UUID based on the cron definition's spec, config, and expression. TYPE: `UUID`

Example

from sheppy import task, Queue
from sheppy.task_factory import TaskFactory

@task
def say_hello(to: str) -> None:
    print(f"Hello, {to}!")

q = Queue(...)
success = await q.add_cron(say_hello("World"), "*/5 * * * *")
assert success is True

success = await q.add_cron(say_hello("World"), "*/5 * * * *")
assert success is False  # duplicate cron definition prevented

# second example
cron1 = TaskFactory.create_cron_from_task(say_hello("World"), "*/5 * * * *")
cron2 = TaskFactory.create_cron_from_task(say_hello("World"), "*/5 * * * *")
assert cron1.deterministic_id == cron2.deterministic_id
assert cron1.id != cron2.id  # different random UUIDs

next_run ¶

next_run(start: datetime | None = None) -> datetime

Get the next scheduled run time based on the cron expression.

PARAMETER	DESCRIPTION
`start`	The starting point to calculate the next run time. If None, the current UTC time is used. TYPE: `datetime \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`datetime`	The next scheduled run time. TYPE: `datetime`

Source code in src/sheppy/models.py

def next_run(self, start: datetime | None = None) -> datetime:
    """Get the next scheduled run time based on the cron expression.

    Args:
        start (datetime|None): The starting point to calculate the next run time. If None, the current UTC time is used.

    Returns:
        datetime: The next scheduled run time.
    """
    if not start:
        start = datetime.now(timezone.utc)
    return croniter(self.expression, start).get_next(datetime)

create_task ¶

create_task(start: datetime) -> Task

Create a Task instance for the next scheduled run. Used by workers to create tasks based on the cron schedule.

The task ID is deterministic based on the cron definition and the scheduled time to prevent duplicates.

PARAMETER	DESCRIPTION
`start`	The scheduled time for the task. TYPE: `datetime`

RETURNS	DESCRIPTION
`Task`	A new Task instance scheduled to run at the specified time. TYPE: `Task`