dbader / schedule Goto Github PK
View Code? Open in Web Editor NEWPython job scheduling for humans.
Home Page: https://schedule.readthedocs.io/
License: MIT License
Python job scheduling for humans.
Home Page: https://schedule.readthedocs.io/
License: MIT License
Valid value should also be there for second like for minutes and hours
How do I pass an argument to the function?
Or shall I use a global object?
Cheers.
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.0
I 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,
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.)
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)
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 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)
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)
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.
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!)
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)
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 and thank for your work: greatfull
Can you explain me how to kill, or modify a Schedule task?
Thank you
On run schedule.run_pending()
and exist 50 jobs and each job leads 20m, generate dead lock!
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
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?
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
Currently the module only comes with a README and docstrings within the code. We should make documentation more explicit by providing the following:
Sorry, accidentally opened this in the wrong project >.<
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()
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:
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.
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, 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.
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)
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)
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
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
then I can control agent working status like logging
thx
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
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!
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.
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?
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.
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
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)
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
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 )
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?
In crontab it's:
*/5 18-23 * * * /home/me/job
Do you know how to realize it in schedule?
Many thanks!
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 :)
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 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.
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.
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.
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,
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()
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)?
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.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.