core.cronjobs

Provides a way to specify cronjobs which should be called at an exact time.

Example:

# A job that runs at 08:00 exactly
@App.cronjob(hour=8, minute=0, timezone='Europe/Zurich')
def cleanup_stuff(request):
    pass

# A job that runs every 15 minutes
@App.cronjob(hour='*', minute='*/15', timezone='UTC')
def do_interesting_things(request):
    pass

Functions registered like this will be called through a regular request. That means they are just like any other view and have all the power associated with it.

Warnings

The function is called once for each application id! So when the time comes for your function to be called, you can expect many calls on a busy site. Each application id also gets its own thread, which is not terribly efficient. It is expected that this behaviour will change in the future with an improved scheduling mechanism.

Also note that the scheduler will offset your function automatically by up to 30 seconds, to mitigate against the trampling herd problem.

As a result you want to use this feature sparingly. If possible do clean up on user action (either manually, or when the user changes something somewhat related). For example, we clean up old reservation records whenever a new reservation is received.

This also means that it is better to have a cronjob that runs once a day on a specific time (say at 13:37 each day), than a cronjob that runs exactly on 00:00 when a lot of other things might be scheduled.

Also the more a cronjob is called the costlier it is for the site (a job that runs every minute is naturally more heavy on resources than one run every 15 minutes).

Finally note that cronjobs for any given application id are only run once the first request to the application with that id has been made. The reason for this is the fact that the background thread needs a real request to be initialized.

In other words, if nobody visits the website the cronjob runs on, then the cronjobs won’t run.

Module Contents

Classes

Job	A single cron job.
ApplicationBoundCronjobs	A daemon thread which runs the cronjobs it is given in the background.

Functions

parse_cron(→ Iterable[int])	Minimal cron style interval parser. Currently only supports this:
get_job(→ Job[Executor] | None)	The internal path to the cronjob. The id can't be guessed.
run_job(→ None)	Executes the job.
register_cronjob(→ None)

Attributes

Scheduled
_JobFunc
CRONJOB_MAX_DURATION
CRONJOB_FORMAT

core.cronjobs.Scheduled: typing_extensions.TypeAlias

core.cronjobs._JobFunc

core.cronjobs.CRONJOB_MAX_DURATION = 30

core.cronjobs.CRONJOB_FORMAT

core.cronjobs.parse_cron(value: str | int, type: Literal[hour, minute]) → Iterable[int]

Minimal cron style interval parser. Currently only supports this:

*   -> Run every hour, minute
*/2 -> Run every other hour, minute
2   -> Run on the given hour, minute

Returns a tuple, iterator or generator.

class core.cronjobs.Job(function: _JobFunc, hour: int | str, minute: int | str, timezone: sedate.types.TzInfoOrName, once: bool = False, url: str | None = None)

Bases: Generic[_JobFunc]

A single cron job.

property title: str

property id: str

Internal id signed by the application. Used to access the job through an url which must be unguessable, but the same over many processes.

The signature is only possible if self.app is present, which can only be set after the instance has already been created.

See as_request_call().

__slots__ = ('app', 'name', 'function', 'hour', 'minute', 'timezone', 'offset', 'once', 'url')

app: onegov.core.framework.Framework

name: str

function: _JobFunc

hour: int | str

minute: int | str

timezone: sedate.types.TzInfo

offset: float

once: bool

url: str | None

runtimes(today: datetime.date) → Iterator[datetime]

Generates the runtimes of this job on the given day, excluding runtimes in the past.

next_runtime(today: datetime.date | None = None) → datetime.datetime

Returns the time (epoch) when this job should be run next, not taking into account scheduling concerns (like when it has run last).

If no runtime is found for today, a runtime is searched tomorrow. If that doesn’t work, no next runtime exists. This would be an error, since we do not currently support things like weekly cronjobs.

as_request_call(request: core.request.CoreRequest) → Job[Scheduled]

Returns a new job which does the same as the old job, but it does so by calling an url which will execute the original job.

class core.cronjobs.ApplicationBoundCronjobs(request: core.request.CoreRequest, jobs: Iterable[Job[Executor]])

Bases: threading.Thread

A daemon thread which runs the cronjobs it is given in the background.

The cronjobs are not actually run in the same thread. Instead a request to the actual cronjob is made. This way the thread is pretty much limited to basic IO work (GET request) and the actual cronjob is called with a normal request.

Basically there is no difference between calling the url of the cronjob at a scheduled time and using this cronjob.

This also avoids any and all locking problems as we don’t have to write OneGov applications in a thread safe way. The actual work is always done on the main thread.

Each application id is meant to have its own thread. To avoid extra work we make sure that only one thread per application exists on each server. This is accomplished by holding a local lock during the lifetime of the thread.

WARNING This doesn’t work on distributed systems. That is if the onegov processes are distributed over many servers the lock isn’t shared.

This can be achieved, but it’s difficult to get right, so for now we do not implement it, as we do not have distributed systems yet.

run() → None

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

run_locked() → None

schedule(job: Job[Scheduled]) → None

process_job(job: Job[Scheduled]) → None

core.cronjobs.get_job(app: onegov.core.framework.Framework, id: str) → Job[Executor] | None

The internal path to the cronjob. The id can’t be guessed.

core.cronjobs.run_job(self: Job[Executor], request: core.request.CoreRequest) → None

Executes the job.

core.cronjobs.register_cronjob(registry: object, function: Executor, hour: int | str, minute: int | str, timezone: sedate.types.TzInfoOrName, once: bool = False) → None