In the doc. a "while True" would be more pythonic than the current "while 1" IMO. Maybe "import time" could be added too.

Very nice project, I'm already using it :)

Hello Daniel, I would really like to use schedule, but for me it only returns TypeError: 'Job' object is not callable:

import schedule

def job():
    print("foo")

schedule.every(10).minutes().do(job)

I'm on Xubuntu 14.04, Python 2.7.6 and installed schedule via pip. Can you tell me what is going wrong here? Maybe it's just a simple mistake. Thank you very much for your help!

I would love to see schedule extended to work as a decorator. I envision syntax along the lines of

@schedule.every().hour.do
def job():
    print("I'm working...")

However, this doesn't work.

When thinking through this, I found that this does work now:

import schedule
import time

dec = schedule.every(2).seconds.do

@dec
def job():
    print("I'm working...")

Parentheses aren't allowed in decorators. (I think this is a result of Guido's gut feeling.) Syntax along the lines of this would be permitted: @schedule.every.seconds.do.

Maybe it'd be insane, but we could implement something like @schedule.every.ten.seconds.do. There's probably a clever way to make that work for a large number of cases. Though, it doesn't resolve the issue of hard coding times.

Some discussion would be helpful before anyone tries implementing this.

I came across this project on HN and am putting together my mental team of projects for a software literate company - and wanted to understand a bit more.

Cron suffers from being fixed to one machine - whereas a company or organisation almost be default has more than one server and wants to load balance in some form

The first part (distributed job definitions) seems simple - text files can be as distributed as you like and brought together

The next stage is after the jobs have been scheduled, to have them sent out and run on more than one server - to have a grid, treat it as a grid and scheduler for it

How amenable is the schedule package to that step (or is it already there and I can stop worrying!)

This might be out of scope for schedule, but I think it would be very handy.

In a similar vein to what APScheduler has, we should support at least some kind of persistent job store.

Let's leave RAM as the default, but add an SQLite store as an option.

Reasonings:

• SQLite API is built into Python 2/3 and is rock-solid, so no additional dependencies.
• Keeps persistent state across restarts, as mentioned in #5.
• Ability to manage jobs from outside of the main schedule process, which avoids interrupting the process for that.
• Related to the above, ability to see when the next run is scheduled for (perhaps there could be a schedule CLI tool?).
• Centralizes the jobs so in theory we could daemonize the schedule process and have it manage all jobs on a single machine.

Thoughts? I might implement this myself soon, just wanted to see if there's any interest for it.

One particular hurdle I'll have to look into is how APScheduler is dealing with persisting jobs. I'm not familiar with pickling/storing functions.

Now to schedule new job uses the following syntax:

schedule.every(10).minutes.do(job)

Maybe the more consistent to use

schedule.every.10.minutes.do(job)

Any thoughts?

Hi! I'm trying to use this package to run a task once every week. However, I have noticed that when I run my scheduler script (currently set to running the task every 2 minutes for debugging) in the background the job does not happen. Does the script need to be always running in the foreground for the task to be executed, or was this package intended to provide fire-and-forget functionality?

To further elaborate: suppose I run the scheduler script in the background, close the terminal, shut down my laptop and open it later can I still expect the weekly task to execute when Sunday comes around (just like cron can be relied on to behave so)?

I have mentioned this previously, I run schedule each day with result:
2015-10-02 18:00:00 - schedule - INFO - Running job Every 1 day at 18:00:00 do delete_kwh() (last run: [never], next run: 2015-10-02 18:00:00)

The problem is the next run information it says 2015-10-02 but it just ran that job so it should say the next run is tomorrow like next run: 2015-10-03 18:00:00)

Currently adding a job like :

schedule.every(5).minutes.do(my_job)

Is there any option to do the specific job on a specific date like:

schedule.on(year/month/date).at(time).do(my_job)

If it is already available can you please share me an example.

I'm interested in scheduling an event at a certain time every day, for x amount of day (or on certain dates) is there any way to do this

Hi! how can i use schedule as a cron job with flask microframework? This is the answer i found on stackoverflow

# -*- coding: utf-8 -*-

"""
This script is a simple example you can use to schedule task with flask 
microframework and schedule (https://github.com/dbader/schedule).
I've found it on on stackoverflow!
"""

import time
import schedule

from flask import Flask, request
from threading import Thread

app = Flask(__name__)

start_time = time.time()

def run_every_10_seconds():
    print("Running periodic task!")
    print "Elapsed time: " + str(time.time() - start_time)

def run_schedule():
    while 1:
        schedule.run_pending()
        time.sleep(1)   

@app.route('/', methods=['GET'])
def index():
    return '<html>test</html>'

if __name__ == '__main__':
    schedule.every(10).seconds.do(run_every_10_seconds)
    t = Thread(target=run_schedule)
    t.start()
    print "Start time: " + str(start_time)
    app.run(debug=True, host='0.0.0.0', port=5000, use_reloader=False)

schedule.every(10).minutes.do(job)
AttributeError: 'module' object has no attribute 'every'

My full code

import schedule
import time

def job():
print("Helo Wolrd ")

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)

while True:
schedule.run_pending()
time.sleep(1)

then I can control agent working status like logging

thx

Sorry, accidentally opened this in the wrong project >.<

How do I pass an argument to the function?
Or shall I use a global object?
Cheers.

Right now we're seeing:

2014-11-28 17:31:53,074 [PID 12] INFO - Running job Every 90 seconds do threaded(<function quoter at 0x7f739e3e00c8>) (last run: 2014-11-28 17:30:22, next run: 2014-11-28 17:31:5$

It'd be nice if this would be slightly more concise. Eg.

2014-11-28 17:31:53,074 [PID 12] INFO - Running every 90 seconds 'threaded(<function quoter at 0x7f739e3e00c8>)' (last run: 17:30:22, next run: 17:31:50)

Hi, I have some question to ask.

import threading
import time

import schedule


def job(name, j):
    print("I'm running on thread %s" % threading.current_thread() + name)
    print(6 + j)


def run_threaded(job_func, arg):
    threading.Thread(target=job_func, args=arg).start()


args = ("123", 6)

schedule.every(5).seconds.do(run_threaded, (job, args))
schedule.every(5).seconds.do(run_threaded, (job, args))

while 1:
    schedule.run_pending()
    time.sleep(1)

1. TypeError: run_threaded() takes exactly 2 arguments (1 given)
2. How to run a job ? for example 10:00-13:00 , 15:00-16:00
3. How to know when the job finish and cancel it from schedule ?Or it will cancel itself ?

Schedule is a great tool .Thanks to its contributors.I hope it can has a better documentation.Because the example in README and FAQ is very simple.

schedule.every().hour.do(job)
only take (1) as argument, Please correcet me if I am wrong?

Do I need some hack to schedule it for two hours?

Hi,

I need help with this code snippet. Below, I'm instantiating a schedule object within a class.
And from within the class I'm running my loop which I hope to run the scheduled job. It seems the "at" method won't accept the job I give it throwing a TypeError. Whereas running it every minute works fine.
In both cases assert statement does not fire. Can someone provide insight?

Thanks,
Dennis

import schedule
import time

class TestSchedule():

    def __init__(self):
        self.schedule = schedule

    def setup(self):
        ### WORKS ###
        # self.schedule.every(1).minutes.do(self.job)
        assert callable(self.job)
        ### DOESN'T WORK ###
        self.schedule.every().day().at("7:54").do(self.job)

    def job(self):
        print 'Job start at {0}'.format(time.strftime('%H:%M'))

    def loop(self):
        while 1:
            self.schedule.run_pending()
            print 'Sleeping for 10 sec'
            time.sleep(10)

if __name__ == '__main__':
    ts = TestSchedule()
    ts.setup()
    print 'Start schedule at {0}'.format(time.strftime('%H:%M'))
    ts.loop()

I added one of my functions to the basic starter example. To start, the test function prints every 5 seconds as expected. Then my function runs but when it completes, the whole process stops and the test function is no longer running every 5 seconds.

Could there be something specific to my function that would cause the while True condition to break?

import schedule
import time
import datetime
import my_module

def test():
    print '{} This is a test'.format(datetime.datetime.now())

# Test run
schedule.every(5).seconds.do(test)
schedule.every().tuesday.at('14:16').do(my_module.function)

while True:
    schedule.run_pending()
    time.sleep(1)

Hello,

When I tried to install schedule via pip install schedule, i got the following error :

Traceback (most recent call last): File "/usr/bin/pip", line 9, in <module> load_entry_point('pip==1.5.6', 'console_scripts', 'pip')() File "/usr/lib/python2.7/dist-packages/pkg_resources/__init__.py", line 558, in load_entry_point return get_distribution(dist).load_entry_point(group, name) File "/usr/lib/python2.7/dist-packages/pkg_resources/__init__.py", line 2682, in load_entry_point return ep.load() File "/usr/lib/python2.7/dist-packages/pkg_resources/__init__.py", line 2355, in load return self.resolve() File "/usr/lib/python2.7/dist-packages/pkg_resources/__init__.py", line 2361, in resolve module = __import__(self.module_name, fromlist=['__name__'], level=0) File "/usr/lib/python2.7/dist-packages/pip/__init__.py", line 74, in <module> from pip.vcs import git, mercurial, subversion, bazaar # noqa File "/usr/lib/python2.7/dist-packages/pip/vcs/mercurial.py", line 9, in <module> from pip.download import path_to_url File "/usr/lib/python2.7/dist-packages/pip/download.py", line 25, in <module> from requests.compat import IncompleteRead ImportError: cannot import name IncompleteRead
What can I do ?
Thx

This section of the documentation mentions source in a fork of scheduler: https://github.com/dbader/schedule/blob/master/FAQ.rst#how-to-continuously-run-the-scheduler-without-blocking-the-main-thread https://github.com/mrhwick/schedule/blob/master/schedule/__init__.py

So why not include this in the library? I installed from https://github.com/mrhwick/schedule/ to get the run_continuously() function, but then realized the fork does not include other things mentioned in the documentation, such as CancelJob ( https://github.com/dbader/schedule/blob/master/FAQ.rst#how-can-i-run-a-job-only-once )

Currently the module only comes with a README and docstrings within the code. We should make documentation more explicit by providing the following:

• Full HTML documentation generated from the source, hosted on read-the-docs.org
  • read-the-docs tutorial
• An improved rationale why this module exists
• Example code for common use cases
• A FAQ section that deals with a few points raised in the reddit /r/python thread

Valid value should also be there for second like for minutes and hours

Can not get this to work. error: No module named every?

Any ideas?

Is there a way to schedule a job to run in a specified amount of time? e.g., I would like to schedule a job to run 5 minutes from now.

There is one task out of several that is not running or getting scheduled for its next run incorrectly.
The task 'catalog_export' typically runs for 90-120 minutes so the start is occurs on one day at it finishes on the next day. We run it on the staging environment at 3am and it has never failed to run correctly, so I am wondering if the start date being different from the end date could be a factor.

Before enabling schedule logging it failed twice. The pattern was that it ran successfully on the initial run, failed to run on the next night, ran sucessfully on the following (third) night, and then failed to run again on the fourth night.

The code says:
schedule.every().day.at("23:25").do(catalog_export)
And it appears in the log:

2016-04-11 23:25:00,334 = schedule - INFO - Running job Every 1 day at 23:25:00 do catalog_export() (last run: [never], next run: 2016-04-11 23:25:00)

Here is one that worked without incident and was scheduled before catalog_export and ran the next day without incident.

2016-04-11 23:15:00,397 = schedule - INFO - Running job Every 1 day at 23:15:00 do customerAlias_load() (last run: [never], next run: 2016-04-11 23:15:00)

2016-04-12 23:15:00,048 = schedule - INFO - Running job Every 1 day at 23:15:00 do customerAlias_load() (last run: 2016-04-11 23:16:55, next run: 2016-04-12 23:15:00)

But there is nothing in the log for catalog_export on 2016-04-12, 23:25 is just silent.

2016-04-12 23:16:51,927 = cat2es.customerAlias_load - INFO - finished at: 23:16:51
2016-04-12 23:57:24,667 = schedule - INFO - Running job Every 60 minutes do heartbeat() (last run: 2016-04-12 22:57:24, next run: 2016-04-12 23:57:24)

There are two logs attached.
In the log for 20160411:
between lines 22-23: catalog_export fails to run
line 34: schedule log enabled
line 96: catalog_export starts, and is rescheduled

in the log for 20160412:
between lines 20-31: catalog_export fails to run

cat2es_20160412_log.txt
cat2es_20160411_log.txt

In my uses I've been wanting to run a job only once at a specific time, then cancel it.
This program is so intuitively worded that I keep thinking it's supported.
Something like schedule.today.at('10:30').do(job)

I've been working around it by scheduling it every hour/day/week at the time I desire, then cancelling the job right after it executes. It's a tad tedious though because I have to manually keep a reference to the job ID to later cancel that one job, and it looks clunky of course.

Hi!
many thanks for useful library!
I have issue - i need to allow user to input scheduler according to his needs. Is there any way I could allow user to input scheduling command and input 'job' myself.
many thanks!

What do you think about adding support for scheduler.every().month.do(…) or scheduler.every().year.do(…)? I have a job that needs to run on the first of the month. (I'm currently running it every day, and the job bails if it's not the first of the month.)

• Document #22 (weekday support)
• Give credit for #22
• Bump version to 0.3.0 (bump minor because we're extending the API)
• Write docs
• QA
• Bundle up and push to PyPI

The code snippet in the FAQ's for scheduling a job just once does not actually seem to work. I have just tried the following and got the error that the attribute today does not exist (i also checked the code and can't see today as part of scheduler or job class):

>>> import schedule
>>> def job():
...     print "its time"
...     return schedule.CancelJob
... 
>>> schedule.today.at('22:30').do(job)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'module' object has no attribute 'today'

Then to double check that scheduler is working for me i did the following:

>>> schedule.every().day.at('22:30').do(job)
Every 1 day at 22:30:00 do job() (last run: [never], next run: 2015-07-13 22:30:00)

I am wanting to schedule 1 off jobs for the future (as well as every day jobs) and wondering how this could be achieved.

Thanks,
James

On run schedule.run_pending() and exist 50 jobs and each job leads 20m, generate dead lock!

When next_run() is called on an empty job queue, a ValueError is thrown in min(self.jobs):
ValueError: min() arg is an empty sequence

This can happen, for instance, while checking the remaining idle time with idle_seconds().

Proposal: check the job queue in next_run() before calculating min.

Hi,

Thanks for making this, it is a great tool. I am wondering whether anyone has or is planning to build out functionality for complex intervals such as "last Saturday of every month"?

Thanks,
Alex

Hi! I've been using schedule for about a month or so now, and I think it's a great library. I was wondering if I could contribute to its development in any way? Any new features, documentation or tests to add?

e.g. every().monday.do(job) and every().monday.at("10:30").do(job)

(This was suggested on reddit by kweer)

We could also go crazy with allowing longer intervals, for example every().second.monday.do(job) but I think this would be out of schedule's scope. Reliably handling infrequent jobs with large intervals requires persistent state. This is better left to tools specialized for this job, such as cron.

Reddit user fdemmer suggested directly using datetime and timedelta objects for maximum flexibility.

In crontab it's:
*/5 18-23 * * * /home/me/job

Do you know how to realize it in schedule?
Many thanks!

At the moment it is possible to schedule a task to run every hour however it schedules the task every hour from the moment it is added to the cue. There is no convenient way to schedule a task every "whole" hour (You'd have to add the job 24 times on the hour..)

Suggestion for improvement: enable "whole hour" cues.

Currently adding a job like :

    schedule.every(5).minutes.do(my_job)

will run the job after 5 min from now. add a func which lets the user to specify if the job should be run at the time of adding it. Example :

    schedule.every(5).minutes.do(my_job).now(True)

I am trying to execute 50 items every 10 seconds, but from the my logs it says it executes every item in 10 second schedule serially, is there a work around?

Hi,

FYI: there is a similar work called APScheduler (http://pythonhosted.org/APScheduler/), which is a heavier solution and more similar to cron.

I have a question too. With "schedule" I want to execute a code at Xh00, Xh15, Xh30, and Xh45, where X ranges from 0 to 23, i.e. execute a code every 15 minutes and start at Xh00 sharp. This is different from "execute something every 15 minutes", because this could start at Xh03, then it continues at Xh18, etc.

Best,

Laszlo

What is the syntax to use the run_continuously function? I tried run_continuously(name) and run_continuously(schedule.every().day.at("00:30").do(name)) but both didnt work. Also, where do i put the function?

Hi and thank for your work: greatfull

Can you explain me how to kill, or modify a Schedule task?

Thank you

Hi,

I am using schedule as a cron-like library. Installation on my OS X (python3 installed via brew) has just being install, but I can't install on my production server (debian).

I've check my setups, I have 3.4.0 on my debian and 3.4.1 on my OS X.

On python 3.4.0I have

Downloading/unpacking schedule
  Downloading schedule-0.3.0.tar.gz
  Running setup.py (path:/tmp/pip_build_root/schedule/setup.py) egg_info for package schedule
    Traceback (most recent call last):
      File "<string>", line 20, in <module>
      File "/tmp/pip_build_root/schedule/setup.py", line 20, in <module>
        open('README.rst').read() + '\n\n' +
      File "/usr/local/lib/python3.4/encodings/ascii.py", line 26, in decode
        return codecs.ascii_decode(input, self.errors)[0]
    UnicodeDecodeError: 'ascii' codec can't decode byte 0xe2 in position 1964: ordinal not in range(128)
    Complete output from command python setup.py egg_info:
    Traceback (most recent call last):

  File "<string>", line 20, in <module>

  File "/tmp/pip_build_root/schedule/setup.py", line 20, in <module>

    open('README.rst').read() + '\n\n' +

  File "/usr/local/lib/python3.4/encodings/ascii.py", line 26, in decode

    return codecs.ascii_decode(input, self.errors)[0]

UnicodeDecodeError: 'ascii' codec can't decode byte 0xe2 in position 1964: ordinal not in range(128)

----------------------------------------
Cleaning up...
Command "python setup.py egg_info" failed with error code 1 in /tmp/pip_build_root/schedule
Storing debug log for failure in /root/.pip/pip.log

Regards,

I like the fluent interface of this library but the main thread must run the GUI event loop so it'd be great if there was a way of conveniently scheduling jobs on a separate thread.

Hi,

I'm trying to schedule a set of jobs. The jobs are supposed to be taken from a config file but for this test I only need to simulate it.

The sample code below setups a list of jobs (I'm using "work" as my name for job) prior to setting up the scheduler. When I run the scheduler and pass in the function "pp" to "do" method the code below executes as expected. However, when I pass to "do" a function within the works list, whether self.sqr or inline definition of sqr or a lambda function, the code does not execute correctly.

What am I doing wrong? I want the function to be defined in the works list (that gets executed by "do") because I would like to define my scheduler to have no knowledge of the kind of work it will perform.

Thanks for any help,
Dennis

import schedule
import time
import datetime

class TestSchedule():

    def __init__(self):
        self.schedule = schedule
        self.works = []
        self.setup_works()
        self.setup_schedule()

    def sqr(self, x):
        return x * x

    def setup_works(self):
        #def sqr(x):
        #    return x * x

        for num in range(24):
            self.works.append({
                'name': num,
                'func': self.sqr,
            })

    def run(self, interval=30, blocking=True):
        while True:
            if not blocking:
                break
            else:
                print 'Before run_pending()  -----------------------'
                self.schedule.run_pending()
                print '{0}: Sleeping ... ({1} sec)'.format(datetime.datetime.now(), interval)
                time.sleep(interval)

    def print_works(self):
        for work in self.works:
            print(work['name']),
        print

    def print_exec_works(self):
        for work in self.works:
            print(work['func'](work['name'])),
        print

    def setup_schedule(self):
        def pp(s):
            print(s),

        for work in self.works:
            sched = self.parse_work_schedule(work)
            if sched:
                #sched(pp, work['name'])
                sched(work['func'], work['name'])
                print 'Scheduled: {0}'.format(sched)
            else:
                print 'Function pointer {0} returned None value.'.format(sched)

    def parse_work_schedule(self, work):
        options = {
            'all': self.schedule.every().minute.do,
        }
        return options['all']


def main():
    t = TestSchedule()
    t.print_works()
    t.print_exec_works()
    t.run()

if __name__ == '__main__':
    main()

Hi,

I don't know if this belongs here so sorry if it's in the wrong place! Loving the scheduler though.

Quick question, I would like to use variables to remotely change the times for the schedule. Something like

time1 = "08:00"
schedule.every().day.at("%s").do(job) % (time1)

However, I just can't get it to work, which ever way round I try the string configuration and explanation marks.

Any help would be greatly received.

Thanks

T

I believe this line

https://github.com/dbader/schedule/blob/master/schedule/__init__.py#L325

is missing an

and self.interval == 1

because when you set for example:

scheduler.every(2).days.at('12:00').do(job)

executing it after 12:00 will schedule next_run for the next 2 days, and executing it before 12:00 will schedule to the next day, breaking the 2-day rule

to see it happen just change

with mock_datetime(2010, 1, 6, 13, 16):

to

with mock_datetime(2010, 1, 6, 10, 16):

here: https://github.com/dbader/schedule/blob/master/test_schedule.py#L192

Hi, I set my job to run every several seconds like: "schedule.every(t).seconds.do(func)".

Then I find that the job wait to run until t seconds. So how can I make it to run immediately? Thanks.