Schedule¶

Planificació de tasques Python per a humans. Executa funcions Python (o qualsevol altre callable) periòdicament utilitzant una sintaxi descriptiva.

Una API fàcil d'usar per programar tasques, feta per a humans. Planificador en procés per a tasques periòdiques. No calen processos addicionals! Molt lleuger i sense dependències externes. Excel·lent cobertura de proves. Provat a Python 3.7, 3.8, 3.9, 3.10 i 3.11

Exemples¶

Exemple bàsic¶

$ pip install schedule
import schedule
import time

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

schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
schedule.every().day.at("12:42", "Europe/Amsterdam").do(job)
schedule.every().minute.at(":17").do(job)

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

Executar una tasca cada x minuts¶

import schedule
import time

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

# Executa la tasca cada 3 segons/minuts/hores/dies/setmanes,
# Començant 3 segons/minuts/hores/dies/setmanes a partir d'ara
schedule.every(3).seconds.do(job)
schedule.every(3).minutes.do(job)
schedule.every(3).hours.do(job)
schedule.every(3).days.do(job)
schedule.every(3).weeks.do(job)

# Executa la tasca cada minut al segon 23
schedule.every().minute.at(":23").do(job)

# Executa la tasca cada hora al minut 42
schedule.every().hour.at(":42").do(job)

# Executa les tasques cada 5 hores, 20 minuts i 30 segons.
# Si l'hora actual és 02:00, la primera execució és a les 06:20:30
schedule.every(5).hours.at("20:30").do(job)

# Executa la tasca cada dia a una HH:MM específica i a la HH:MM:SS següent
schedule.every().day.at("10:30").do(job)
schedule.every().day.at("10:30:42").do(job)
schedule.every().day.at("12:42", "Europe/Amsterdam").do(job)

# Executa la tasca un dia específic de la setmana
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)

Usar un decorador per programar una tasca¶

Fes servir @repeat per programar una funció. Passa-li un interval usant la mateixa sintaxi que l'anterior mentre omets el .do().

from schedule import every, repeat, run_pending
import time

@repeat(every(10).minutes)
def job():
    print("I am a scheduled job")

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

El decorador @repeat no funciona en mètodes de classe no estàtics.

Passar arguments a una tasca¶

do() passa arguments addicionals a la funció de tasca.

import schedule

def greet(name):
    print('Hello', name)

schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')

from schedule import every, repeat

@repeat(every().second, "World")
@repeat(every().day, "Mars")
def hello(planet):
    print("Hello", planet)

Cancel·lar una tasca¶

Per treure una tasca del planificador, fes servir el mètode schedule.cancel_job(job).

import schedule

def some_task():
    print('Hello world')

job = schedule.every().day.at('22:30').do(some_task)
schedule.cancel_job(job)

Executar una tasca una sola vegada¶

Retorna schedule.CancelJob des d'una tasca per treure-la del planificador.

import schedule
import time

def job_that_executes_once():
    # Fes alguna feina que només cal fer una vegada...
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job_that_executes_once)

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

Obtenir totes les tasques¶

Per recuperar totes les tasques del planificador, fes servir schedule.get_jobs().

import schedule

def hello():
    print('Hello world')

schedule.every().second.do(hello)

all_jobs = schedule.get_jobs()

Cancel·lar totes les tasques¶

Per treure totes les tasques del planificador, fes servir schedule.clear().

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().second.do(greet)

schedule.clear()

Obtenir diverses tasques filtrades per etiquetes¶

Pots recuperar un grup de tasques del planificador seleccionant-les mitjançant un identificador únic.

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')

friends = schedule.get_jobs('friend')

Retornarà una llista de totes les tasques etiquetades com a friend.

Cancel·lar diverses tasques filtrades per etiquetes¶

Pots cancel·lar la programació d'un grup de tasques seleccionant-les mitjançant un identificador únic.

import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')

schedule.clear('daily-tasks')

Evitarà que totes les tasques etiquetades com a daily-tasks es tornin a executar.

Executar una tasca a intervals aleatoris¶

def my_job():
    print('Foo')

# Executa cada 5 a 10 segons.
schedule.every(5).to(10).seconds.do(my_job)

every(A).to(B).seconds executa la funció de tasca cada N segons de tal manera que A <= N <= B.

Executar una tasca fins a un moment determinat¶

import schedule
from datetime import datetime, timedelta, time

def job():
    print('Boo')

# executa la tasca fins a les 18:30 d'avui
schedule.every(1).hours.until("18:30").do(job)

# executa la tasca fins al 2030-01-01 18:33
schedule.every(1).hours.until("2030-01-01 18:33").do(job)

# Programa una tasca per executar-se durant les properes 8 hores
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# Executa my_job fins avui a les 11:33:42
schedule.every(1).hours.until(time(11, 33, 42)).do(job)

# executa la tasca fins a un datetime específic
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)

El mètode until estableix la data límit de les tasques. La tasca no s'executarà després de la data límit.

Temps fins a la propera execució¶

schedule.idle_seconds() s'usa per obtenir el nombre de segons fins que s'executi la propera tasca programada. El valor retornat és negatiu si les properes tasques programades estaven previstes per executar-se en el passat. Retorna None si no hi ha tasques programades.

import schedule
import time

def job():
    print('Hello')

schedule.every(5).seconds.do(job)

while 1:
    n = schedule.idle_seconds()
    if n is None:
        # no hi ha més tasques
        break
    elif n > 0:
        # dorm exactament el temps necessari
        time.sleep(n)
    schedule.run_pending()

Executar totes les tasques ara, independentment de la seva programació¶

Per executar totes les tasques, independentment de si estan programades per executar-se o no, fes servir schedule.run_all(). Les tasques es reprogramen després de finalitzar, tal com ho farien si s'executessin amb run_pending().

import schedule

def job_1():
    print('Foo')

def job_2():
    print('Bar')

schedule.every().monday.at("12:40").do(job_1)
schedule.every().tuesday.at("16:40").do(job_2)

schedule.run_all()

# Afegeix l'argument delay_seconds per executar les tasques amb un nombre
# de segons de retard entre elles.
schedule.run_all(delay_seconds=10)

Quan no usar Schedule¶

Siguem honestos, Schedule no és una biblioteca de planificació de "talla única". Aquesta biblioteca està dissenyada per ser una solució senzilla per a problemes de planificació senzills. Probablement hauries de buscar en un altre lloc si necessites:

Persistència de tasques (recordar el calendari entre reinicis)
Temporització exacta (execució de precisió inferior a un segon)
Execució simultània (diversos subprocessos)
Localització (dies laborables o festius)

Schedule no té en compte el temps que tarda la funció de tasca en executar-se. Per garantir una planificació d'execució estable, has de moure les tasques de llarga durada fora del subprocés principal (on s'executa el planificador). Consulta l'execució paral·lela per obtenir una implementació d'exemple.