csbot.plugins.cron module

class csbot.plugins.cron.Cron(bot)[source]

Bases: csbot.plugin.Plugin

Time, that most mysterious of things. What is it? Is it discrete or continuous? What was before time? Does that even make sense to ask? This plugin will attempt to address some, perhaps all, of these questions.

More seriously, this plugin allows the scheduling of events. Due to computers being the constructs of fallible humans, it’s not guaranteed that a callback will be run precisely when you want it to be. Furthermore, if you schedule multiple events at the same time, don’t make any assumptions about the order in which they’ll be called.

Example of usage:

class MyPlugin(Plugin):

cron = Plugin.use(‘cron’)

def setup(self):

… self.cron.after(

“hello world”, datetime.timedelta(days=1), “callback”)
def callback(self, when):
self.log.info(u’I got called at {}’.format(when))

@Plugin.hook(‘cron.hourly’) def hourlyevent(self, e):

self.log.info(u’An hour has passed’)
tasks

Descriptor for plugin attributes that get (and cache) a value from another plugin.

See Plugin.use().

setup()[source]

Plugin setup.

  • Replace all ProvidedByPlugin attributes.
  • Fire all plugin integration methods.
  • Register all commands provided by the plugin.
teardown()[source]

Plugin teardown.

  • Unregister all commands provided by the plugin.
fire_event(now, name)[source]

Fire off a regular event.

This gets called by the scheduler at the appropriate time.

provide(plugin_name)[source]

Return the crond for the given plugin.

match_task(owner, name=None, args=None, kwargs=None)[source]

Create a MongoDB search for a task definition.

schedule(owner, name, when, interval=None, callback=None, args=None, kwargs=None)[source]

Schedule a new task.

Parameters:
  • owner – The plugin which created the task
  • name – The name of the task
  • when – The datetime to trigger the task at
  • interval – Optionally, reschedule at when + interval when triggered. Gives rise to repeating tasks.
  • callback – Call owner.callback when triggered; if None, call owner.name.
  • args – Callback positional arguments.
  • kwargs – Callback keyword arguments.

The signature of a task is (owner, name, args, kwargs), and trying to create a task with the same signature as an existing task will raise DuplicateTaskError. Any subset of the signature can be used to unschedule() all matching tasks (owner is mandatory).

unschedule(owner, name=None, args=None, kwargs=None)[source]

Unschedule a task.

Removes all existing tasks that match based on the criteria passed as arguments (see match_task()).

This could result in the scheduler having nothing to do in its next call, but this isn’t a problem as it’s not a very intensive function, so there’s no point in rescheduling it here.

schedule_event_runner()[source]

Schedule the event runner.

Set up a delayed call for event_runner() to happen no sooner than is required by the next scheduled task. If a different call already exists it is replaced.

event_runner()[source]

Run pending tasks.

Run all tasks which have a trigger time in the past, and then reschedule self to run in time for the next task.

exception csbot.plugins.cron.DuplicateTaskError[source]

Bases: Exception

Task with a given signature already exists.

This can be raised by Cron.schedule() if a plugin tries to register two events with the same name.

class csbot.plugins.cron.PluginCron(cron, plugin)[source]

Bases: object

Interface to the cron methods restricted to plugin as the task owner..

All of the scheduling functions have a signature of the form (name, time, method_name, *args, **kwargs).

This means that at the appropriate time, the method plugin.method_name will be called with the arguments (time, *args, **kwargs), where the time argument is the time it was supposed to be run by the scheduler (which may not be identical to teh actual time it is run).

These functions will raise a DuplicateNameException if you try to schedule two events with the same name.

schedule(name, when, interval=None, callback=None, args=None, kwargs=None)[source]

Pass through to Cron.schedule(), adding owner argument.

after(_delay, _name, _method_name, *args, **kwargs)[source]

Schedule an event to occur after the timedelta delay has passed.

at(_when, _name, _method_name, *args, **kwargs)[source]

Schedule an event to occur at a given time.

every(_freq, _name, _method_name, *args, **kwargs)[source]

Schedule an event to occur every time the delay passes.

unschedule(name, args=None, kwargs=None)[source]

Pass through to Cron.unschedule(), adding owner argument.

unschedule_all()[source]

Unschedule all tasks for this plugin.

This could be supported by unschedule(), but it’s nice to prevent code accidentally wiping all of a plugin’s tasks.