Reference

pgcron

Core interface for configuring pgcron jobs.

pgcron.Delete

Delete(queryset: QuerySet[Any])

Bases: Expression

A DELETE statement to be executed by pgcron.

Examples:

Delete all users over 100 years old:

pgcron.Delete(User.objects.filter(age__gt=100))

Source code in pgcron/expressions.py

def __init__(self, queryset: models.QuerySet[Any]) -> None:
    self.queryset = queryset

pgcron.SQLExpression

Bases: str, Expression

A SQL statement to be executed by pgcron.

pgcron.Update

Update(queryset: QuerySet[Any], **kwargs: Any)

Bases: Expression

An UPDATE statement to be executed by pgcron.

Examples:

Update all users over 100 years old:

pgcron.Update(User.objects.filter(age__gt=100), age=100)

Source code in pgcron/expressions.py

def __init__(self, queryset: models.QuerySet[Any], **kwargs: Any) -> None:
    self.queryset = queryset
    self.kwargs = kwargs

pgcron.crontab

crontab(
    minute: str = "*",
    hour: str = "*",
    day_of_month: str = "*",
    month_of_year: str = "*",
    day_of_week: str = "*",
)

Bases: Schedule

A crontab for a pgcron job.

Source code in pgcron/schedule.py

def __init__(
    self,
    minute: str = "*",
    hour: str = "*",
    day_of_month: str = "*",
    month_of_year: str = "*",
    day_of_week: str = "*",
) -> None:
    self.minute = minute
    self.hour = hour
    self.day_of_week = day_of_week
    self.day_of_month = day_of_month
    self.month_of_year = month_of_year

from_str classmethod

from_str(expr: str) -> crontab

Create a crontab from a string.

Source code in pgcron/schedule.py

@classmethod
def from_str(cls, expr: str) -> crontab:
    """Create a crontab from a string."""
    minute, hour, day_of_month, month_of_year, day_of_week = expr.split(" ")
    return cls(minute, hour, day_of_month, month_of_year, day_of_week)

to_pgcron_expr

to_pgcron_expr() -> str

Return the schedule as a pgcron expression.

Source code in pgcron/schedule.py

def to_pgcron_expr(self) -> str:
    """Return the schedule as a pgcron expression."""
    return (
        f"{self.minute} {self.hour} {self.day_of_month} {self.month_of_year} {self.day_of_week}"
    )

pgcron.seconds

seconds(n: int)

Bases: Schedule

Run a job every n seconds.

Source code in pgcron/schedule.py

def __init__(self, n: int) -> None:
    if n < 1:
        raise ValueError("Interval must at least 1 second.")
    if n > 59:
        raise ValueError("Interval must be less than 60 seconds.")

    self.n = n

to_pgcron_expr

to_pgcron_expr() -> str

Return the schedule as a pgcron expression.

Source code in pgcron/schedule.py

def to_pgcron_expr(self) -> str:
    """Return the schedule as a pgcron expression."""
    return f"{self.n} seconds"

pgcron.job

job(
    schedule: Schedule | str,
    *,
    name: str | None = None,
    database: str | None = None
) -> Callable[[Callable[[], Expression]], Job]

Register a pgcron job that runs a SQL expression.

Parameters:

Name	Type	Description	Default
schedule	Schedule | str	The schedule object or crontab expression for the job.	required
name	str | None	The name of the job, if None, the name will be the function name.	None
database	str | None	The database to run the job on. Defaults to the database set in PGCRON_DATABASE in settings, which defaults to Django's default database.	None

Returns:

Type	Description
Callable[[Callable[[], Expression]], Job]	A decorator that registers the job with pgcron, and returns a Job object
Callable[[Callable[[], Expression]], Job]	that can be used to run the job synchronously if needed.

Examples:

Job that deletes users over 100 years old every minute:

import pgcron


@pgcron.job("* * * * *")
def delete_old_users() -> pgcron.SQLExpression:
    return pgcron.SQLExpression("DELETE FROM users WHERE age > 100")

Job that vacuums the database every day at 9am:

import pgcron


@pgcron.job("0 9 * * *")
def vacuum_db() -> pgcron.SQLExpression:
    return pgcron.SQLExpression("VACUUM")

Job that calls a stored procedure every 5 seconds:

import pgcron


@pgcron.job(pgcron.seconds(5))
def call_stored_procedure() -> pgcron.SQLExpression:
    return pgcron.SQLExpression("CALL my_stored_procedure()")

Source code in pgcron/core.py

def job(
    schedule: pgcron.schedule.Schedule | str,
    *,
    name: str | None = None,
    database: str | None = None,
) -> Callable[[Callable[[], expressions.Expression]], Job]:
    """Register a pgcron job that runs a SQL expression.

    Args:
        schedule: The schedule object or crontab expression for the job.
        name: The name of the job, if None, the name will be the function name.
        database: The database to run the job on.
            Defaults to the database set in `PGCRON_DATABASE` in settings,
            which defaults to Django's default database.

    Returns:
        A decorator that registers the job with pgcron, and returns a `Job` object
        that can be used to run the job synchronously if needed.

    Examples:
        Job that deletes users over 100 years old every minute:
        ```python
        import pgcron


        @pgcron.job("* * * * *")
        def delete_old_users() -> pgcron.SQLExpression:
            return pgcron.SQLExpression("DELETE FROM users WHERE age > 100")
        ```

        Job that vacuums the database every day at 9am:
        ```python
        import pgcron


        @pgcron.job("0 9 * * *")
        def vacuum_db() -> pgcron.SQLExpression:
            return pgcron.SQLExpression("VACUUM")
        ```

        Job that calls a stored procedure every 5 seconds:
        ```python
        import pgcron


        @pgcron.job(pgcron.seconds(5))
        def call_stored_procedure() -> pgcron.SQLExpression:
            return pgcron.SQLExpression("CALL my_stored_procedure()")
        ```
    """

    if isinstance(schedule, str):
        schedule = pgcron.schedule.crontab.from_str(schedule)

    def decorator(func: Callable[[], expressions.Expression]) -> Job:
        app_label = func.__module__.split(".")[0]
        sql = func()
        job_name = name or func.__name__
        db_alias = database or _config.get_database()
        _registry.register(
            name=job_name,
            app_name=app_label,
            expression=sql,
            schedule=schedule.to_pgcron_expr(),
            database=db_alias,
        )
        return Job(name=job_name, schedule=schedule, expression=sql, database=db_alias)

    return decorator