Прежде чем я спрошу, Задачи Cron и планировщик заданий будут моими последними параметрами, этот script будет использоваться в Windows и Linux, и я бы предпочел использовать кодированный способ выполнения этого чем оставить это конечному пользователю.
Есть ли библиотека для Python, которую я могу использовать для планирования задач? Мне нужно будет запускать функцию один раз в час, однако со временем, если я запускаю script один раз в час и использую .sleep, "один раз в час" будет работать в другую часть часа с предыдущего дня из-за задержка, присущая выполнению/запуску script и/или функции.
Что такое лучший способ планирования функции в определенное время суток (более одного раза) без использования задания Cron или планирования его с помощью планировщика заданий?
Или, если это невозможно, я также хотел бы получить ваш вход.
import datetime
import time
from apscheduler.scheduler import Scheduler
# Start the scheduler
sched = Scheduler()
sched.daemonic = False
sched.start()
def job_function():
print("Hello World")
print(datetime.datetime.now())
time.sleep(20)
# Schedules job_function to be run once each minute
sched.add_cron_job(job_function, minute='0-59')
из
>Hello World
>2014-03-28 09:44:00.016.492
>Hello World
>2014-03-28 09:45:00.0.14110
(Из ответа Анимеша Пандея ниже)
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
@sched.scheduled_job('interval', seconds=10)
def timed_job():
print('This job is run every 10 seconds.')
@sched.scheduled_job('cron', day_of_week='mon-fri', hour=10)
def scheduled_job():
print('This job is run every weekday at 10am.')
sched.configure(options_from_ini_file)
sched.start()
Возможно, это может помочь: Advanced Python Scheduler
Вот небольшой фрагмент кода из их документации:
from apscheduler.schedulers.blocking import BlockingScheduler
def some_job():
print "Decorated job"
scheduler = BlockingScheduler()
scheduler.add_job(some_job, 'interval', hours=1)
scheduler.start()
Чтобы запустить что-то каждые 10 минут после часа.
from datetime import datetime, timedelta
while 1:
print 'Run something..'
dt = datetime.now() + timedelta(hours=1)
dt = dt.replace(minute=10)
while datetime.now() < dt:
time.sleep(1)
Для apscheduler <3.0 см. Неизвестный ответ.
Для apscheduler > 3.0
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
@sched.scheduled_job('interval', seconds=10)
def timed_job():
print('This job is run every 10 seconds.')
@sched.scheduled_job('cron', day_of_week='mon-fri', hour=10)
def scheduled_job():
print('This job is run every weekday at 10am.')
sched.configure(options_from_ini_file)
sched.start()
документация к apscheduler.
Это для apscheduler-3.3.1 на Python 3.6.2.
"""
Following configurations are set for the scheduler:
- a MongoDBJobStore named "mongo"
- an SQLAlchemyJobStore named "default" (using SQLite)
- a ThreadPoolExecutor named "default", with a worker count of 20
- a ProcessPoolExecutor named "processpool", with a worker count of 5
- UTC as the schedulers timezone
- coalescing turned off for new jobs by default
- a default maximum instance limit of 3 for new jobs
"""
from pytz import utc
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
"""
Method 1:
"""
jobstores = {
'mongo': {'type': 'mongodb'},
'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
'default': {'type': 'threadpool', 'max_workers': 20},
'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
'coalesce': False,
'max_instances': 3
}
"""
Method 2 (ini format):
"""
gconfig = {
'apscheduler.jobstores.mongo': {
'type': 'mongodb'
},
'apscheduler.jobstores.default': {
'type': 'sqlalchemy',
'url': 'sqlite:///jobs.sqlite'
},
'apscheduler.executors.default': {
'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
'max_workers': '20'
},
'apscheduler.executors.processpool': {
'type': 'processpool',
'max_workers': '5'
},
'apscheduler.job_defaults.coalesce': 'false',
'apscheduler.job_defaults.max_instances': '3',
'apscheduler.timezone': 'UTC',
}
sched_method1 = BlockingScheduler() # uses overrides from Method1
sched_method2 = BlockingScheduler() # uses same overrides from Method2 but in an ini format
@sched_method1.scheduled_job('interval', seconds=10)
def timed_job():
print('This job is run every 10 seconds.')
@sched_method2.scheduled_job('cron', day_of_week='mon-fri', hour=10)
def scheduled_job():
print('This job is run every weekday at 10am.')
sched_method1.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
sched_method1.start()
sched_method2.configure(gconfig=gconfig)
sched_method2.start()
В версии, опубликованной sunshinekitty под названием "Версия < 3.0", вам может потребоваться указать apscheduler 2.1.2. Я случайно имел версию 3 на моей установке 2.7, поэтому я пошел:
pip uninstall apscheduler
pip install apscheduler==2.1.2
После этого он работал правильно. Надеюсь, что это поможет.
Стандартная библиотека Python предоставляет sched и threading для этой задачи. Но это означает, что ваш планировщик script будет работать все время, а не оставлять его выполнение в ОС, что может быть или не быть тем, что вы хотите.
Один из вариантов - написать оболочку C/С++, которая регулярно выполняет python script. Ваш конечный пользователь будет запускать исполняемый файл C/С++, который будет работать в фоновом режиме и периодически выполнять python script. Это может быть не лучшее решение и может не работать, если вы не знаете C/С++ или хотите сохранить этот 100% -ный питон. Но это похоже на самый удобный подход, поскольку люди привыкли щелкать по исполняемым файлам. Все это предполагает, что python установлен на вашем компьютере конечного пользователя.
Другой вариант - использовать cron job/Task Scheduler, но установить его в установщик как script, чтобы ваш конечный пользователь этого не делал.
#For scheduling task execution
import schedule
import time
def job():
print("I'm working...")
schedule.every(1).minutes.do(job)
#schedule.every().hour.do(job)
#schedule.every().day.at("10:30").do(job)
#schedule.every(5).to(10).minutes.do(job)
#schedule.every().monday.do(job)
#schedule.every().wednesday.at("13:15").do(job)
#schedule.every().minute.at(":17").do(job)
while True:
schedule.run_pending()
time.sleep(1)