blog.furas.pl
# prywatne notatki - Python, Linux, Machine Learning, etc.

Shedulers - wykonywanie zadań w odstępie czasu

#:status: draft

Argumenty dla wywoływanego zadania podaje się jako listę lub krotkę - nawet jak to jest tylko jeden argument - stąd nawiasy i przecinek w ('3s',)

time.sleep()

Kilka pojedynczych zadań:

 1 #!/usr/bin/env python3
 2 
 3 #from __future__ import print_function # Python 2
 4 import time
 5 
 6 # --- task with argument
 7 
 8 def task(x):
 9     print('task:', x)
10 
11 # --- run
12 
13 time.sleep(3)
14 task('3s')
15 
16 time.sleep(3)
17 task('6s')
18 
19 time.sleep(3)
20 task('9s')

Wielokrotnie to samo zadanie:

 1 #!/usr/bin/env python3
 2 
 3 #from __future__ import print_function # Python 2
 4 import time
 5 
 6 # --- task with argument
 7 
 8 def task(x):
 9     print('task:', x)
10 
11 # --- run
12 
13 while True:
14     time.sleep(3)
15     task('+3s')

sched

Standardowy moduł.

Kilka pojedynczych zadań:

#!/usr/bin/env python3

#from __future__ import print_function # Python 2
#import time # Python 2 - required parameters
import sched

# --- task with argument

def task(x):
    print('task:', x)

# --- create scheduler

#s = sched.scheduler(time.time, time.sleep) # Python 2 - required parameters
s = sched.scheduler()

# --- add few tasks to scheduler

s.enter(3, 0, task, ('3s',))
s.enter(6, 0, task, ('6s',))
s.enter(9, 0, task, ('9s',))

# --- start sheduler

s.run()

Wielokrotnie to samo zadanie:

#!/usr/bin/env python3

#from __future__ import print_function # Python 2
#import time # Python 2 - required parameters
import sched

DELAY = 3

# --- task with argument

def task(x):
    # --- add task again
    s.enter(DELAY, 0, task, (str(DELAY),))

    print('task:', x)

# --- create scheduler

#s = sched.scheduler(time.time, time.sleep) # Python 2 - required parameters
s = sched.scheduler()

# --- add first task to scheduler

s.enter(DELAY, 0, task, (str(DELAY),))

# --- start sheduler

s.run()

Wielokrotnie to samo zadanie z natychmiastowym pierwszym wywołaniem (które dopiero doda kolejne wywołanie do schedulera)

#!/usr/bin/env python3

#from __future__ import print_function # Python 2
#import time # Python 2 - required parameters
import sched

DELAY = 3

# --- task with argument

def task(x):
    # --- add task again
    s.enter(DELAY, 0, task, (str(DELAY),))

    print('task:', x)

# --- create scheduler

#s = sched.scheduler(time.time, time.sleep) # Python 2 - required parameters
s = sched.scheduler()

# --- run first task

task('0')

# --- start sheduler

s.run()

Inne: PyMOTW 2: sched - Generic event scheduler.

schedule

#!/usr/bin/env python

#from __future__ import print_function # Python 2
import schedule
import time

DELAY = 3

def task():
    print('task:')

schedule.every(DELAY).seconds.do(task)

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

Advanced Python Scheduler

Wymaga zainstalowania APScheduler

$ pip install apscheduler

Może wykorzystywać: Pamięć, SQLAlchemy, MongoDB, Redis.

W tle:

#!/usr/bin/env python3

#from __future__ import print_function # Python 2
import time # for test only

from apscheduler.schedulers.background import BackgroundScheduler

# --- task with argument

def task(x):
    print('task:', x)

# --- create scheduler

scheduler = BackgroundScheduler()

# --- add tasks

scheduler.add_job(task, trigger='interval', seconds=3, args=('3s',))
scheduler.add_job(task, trigger='date', date='2016.07.01', args=('today',))

# --- run scheduler

scheduler.start()

time.sleep(10) # for test only

# --- stop scheduler

scheduler.shutdown()

Blokujące:

#!/usr/bin/env python3

#from __future__ import print_function # Python 2
import time # for test only

from apscheduler.schedulers.blocking import BlockingScheduler

# --- task with argument

def task(x):
    print('task:', x)

# --- create scheduler

scheduler = BlockingScheduler()

# --- add tasks

scheduler.add_job(task, trigger='interval', seconds=3, args=('3s',))

# --- run scheduler

scheduler.start()

Inne: repozytorium oraz przykłady

threading Timer

  • Standardowy moduł.
  • Wykorzystuje wątki.
  • Pozwala wykonywać inne rzeczy w tym samym czasie.
#!/usr/bin/env python3

#from __future__ import print_function # Python 2
from threading import Timer
import time

# --- task with argument

def task(x):
    print('task:', x)

# --- add few tasks

Timer(3, task, ('3s',)).start()
Timer(6, task, ('6s',)).start()
Timer(9, task, ('9s',)).start()

# --- start

time.sleep(10)

Celery

Wymaga zainstalowania Celery

$ pip install celery

Wymaga zainstalowania RabbitMQ

$ sudo apt-get install rabbitmq-server

lub Redis

$ sudo apt-get install redis

lub SQLAlchemy


CronTab

Wymaga zainstalowania

$ pip install python-crontab

from crontab import CronTab

Dokumentacja: python-crontab

Inne: - https://github.com/josiahcarlson/parse-crontab - Managing Cron Jobs with Python-Crontab


Książki: python-dla-kazdego-podstawy-programowania python-wprowadzenie python-leksykon-kieszonkowy python-receptury python-programuj-szybko-i-wydajnie python-projekty-do-wykorzystania black-hat-python-jezyk-python-dla-hackerow-i-pentesterow efektywny-python-59-sposobow-na-lepszy-kod tdd-w-praktyce-niezawodny-kod-w-jezyku-python aplikacje-internetowe-z-django-najlepsze-receptury