pytb.schedule module

Run commands on certain dates with a cron-like date expression

You can use the at() and every() as a decorator for to turn that function into a scheduled thread object. Use the start_schedule() start the scheduled execution.

You can use the stop() to stop any further scheduling. The decorated function is also still directly callable to execute the task in the calling thread.

API Documentation

A simple task scheduling system to run periodic tasks

class pytb.schedule.Schedule(target: Callable[[...], Any], interval: Generator[datetime.datetime, None, None])[source]

Bases: threading.Thread

This represents a reoccuring task, exceuting target every time the schedule is due. The target is run in an extra thread by default. If you stop the schedule while the target function is running, the thread is canceled after finishing its current run.

Parameters:
  • target – The target function to execute each time the schedule is due
  • interval – A generator yielding datetime objects that determine when the schedule is due. When this generator is exhausted, the schedule stops. Datetime objects in the past are simply ignored and the next value from the generator is used to schedule the job.
next_schedule() → datetime.datetime[source]

Return the datetime object this schedule is due

run() → None[source]

Start the schedule execution in an extra thread. The target function is called, passing all arguments supplied to this call.

start_schedule(*args, **kwargs) → None[source]

Start the scheduler and pass all supplied arguments to the target function each time the schedule is due

stop() → None[source]

Stop the async execution of the schedule, cacnel all future tasks

pytb.schedule.at(minute: str = '*', hour: str = '*', day: str = '*', month: str = '*', weekday: str = '*') → Callable[[...], pytb.schedule.Schedule][source]

run the task every time the current system-time matches the cron-like expression. Check the documentation for parse_cron_spec() for the supported syntax.

pytb.schedule.every(interval: datetime.timedelta, start_at: Optional[datetime.datetime] = None) → Callable[[...], pytb.schedule.Schedule][source]

Run a task repeadetly at the given interval

Parameters:
  • interval – run the command this often the most
  • start_at – run the command for the first time only after this date has passed. If not specified, run the command immediatley
pytb.schedule.parse_cron_spec(spec: str, max_value: int, min_value: int = 0) → Sequence[int][source]

Parse a string of in a cron-like expression format to a sequence accepted numbers. The expression needs to have one of the following forms:

  • i sequence contains only the element i
  • * indicates that all values possible for this part are included
  • i,j,k specifies a list of possible values
  • i-j specifies a range of values including j
  • i-j/s additionally specifies the step-size
Parameters:
  • spec – The cron-like expression to parse
  • max_value – The maximum value allowed for this range. This is needed to specify the range using the ‘*’ wildcard
  • min_value – The minimum allowed value
Raises:

ValueError – if the spec tries to exceed the limits

Example:

>>> list(parse_cron_spec('5', max_value=7,))
[5]

>>> list(parse_cron_spec('*', max_value=7,))
[0, 1, 2, 3, 4, 5, 6, 7]

>>> list(parse_cron_spec('1-4', max_value=7,))
[1, 2, 3, 4]

>>> list(parse_cron_spec('1-4/2', max_value=7,))
[1, 3]