Harmonogramy zadań w Pythonie pozwalają uruchamiać skrypty cyklicznie: co kilka minut, raz dziennie, w konkretny dzień tygodnia albo po określonym zdarzeniu. Dzięki temu jednorazowy skrypt może stać się automatyzacją, która działa w tle bez ręcznego uruchamiania.

To przydaje się w wielu sytuacjach: generowanie raportów, pobieranie danych z API, czyszczenie plików, wykonywanie backupów, wysyłka powiadomień, synchronizacja baz danych, monitoring usług albo cykliczne przetwarzanie kolejek zadań.

W Pythonie możesz podejść do harmonogramów na kilka sposobów. Do prostych zadań wystarczy biblioteka schedule. Do bardziej zaawansowanych przypadków sprawdzi się APScheduler. Jeśli skrypt ma działać na serwerze, warto znać też systemowy cron. A gdy zadania są asynchroniczne, np. pobierają dane z wielu źródeł, dobrym wyborem może być asyncio.

Kiedy używać harmonogramu zadań w Pythonie?

Harmonogram zadań warto zastosować wtedy, gdy czynność powtarza się regularnie i nie wymaga ręcznej decyzji za każdym razem.

Przykłady:

• codzienne generowanie raportu sprzedaży,
• pobieranie danych z API co godzinę,
• czyszczenie starych plików raz dziennie,
• wysyłanie przypomnień o określonej godzinie,
• sprawdzanie statusu strony co kilka minut,
• archiwizacja danych raz w tygodniu,
• synchronizacja danych między systemami,
• uruchamianie pipeline’u ETL w nocy.

Dobrze zaprojektowany harmonogram zadań powinien nie tylko uruchamiać funkcję o odpowiedniej porze, ale też obsługiwać błędy, logować wynik działania i zapobiegać nakładaniu się kilku instancji tego samego zadania.

Najprostszy harmonogram z biblioteką schedule

Biblioteka schedule jest dobrym wyborem do prostych, lokalnych automatyzacji. Ma czytelną składnię i pozwala szybko zdefiniować zadania uruchamiane co kilka sekund, minut, godzin albo o konkretnej porze dnia.

Instalacja:

Bash
Kopiuj
1
pip install schedule

Przykład prostego harmonogramu:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import schedule
import time
from datetime import datetime

def job():
    print(f"[{datetime.now()}] Wykonuję zadanie")

schedule.every(10).seconds.do(job)
schedule.every().minute.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("08:00").do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)

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

Ten kod działa w pętli i co sekundę sprawdza, czy jakieś zadanie powinno zostać uruchomione. To proste i wygodne, ale trzeba pamiętać, że skrypt musi cały czas działać. Jeśli go zamkniesz, harmonogram przestanie wykonywać zadania.

Przykład: codzienny raport z użyciem schedule

Załóżmy, że chcesz codziennie o 8:00 wygenerować raport. Funkcja może pobierać dane, przetwarzać je i zapisywać wynik do pliku.

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import schedule
import time
from datetime import datetime
from pathlib import Path

def generate_daily_report():
    reports_dir = Path("raporty")
    reports_dir.mkdir(exist_ok=True)

    today = datetime.now().strftime("%Y-%m-%d")
    report_file = reports_dir / f"raport_{today}.txt"

    with open(report_file, "w", encoding="utf-8") as file:
        file.write(f"Raport wygenerowany: {datetime.now()}\n")
        file.write("Tutaj mogą znaleźć się dane sprzedażowe, statusy lub wyniki ETL.\n")

    print(f"Zapisano raport: {report_file}")

schedule.every().day.at("08:00").do(generate_daily_report)

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

Taki harmonogram sprawdzi się w prostych automatyzacjach uruchamianych lokalnie lub na niewielkim serwerze. Jeśli jednak potrzebujesz większej kontroli, lepszym wyborem będzie APScheduler.

APScheduler - zaawansowane harmonogramy w Pythonie

APScheduler to bardziej rozbudowana biblioteka do planowania zadań. Obsługuje różne typy harmonogramów, w tym interwały, konkretne daty oraz wyrażenia podobne do crona.

Instalacja:

Bash
Kopiuj
1
pip install apscheduler

Przykład użycia APScheduler:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime

def scheduled_task():
    print(f"[{datetime.now()}] Wykonuję zadanie harmonogramowe")

scheduler = BlockingScheduler()

scheduler.add_job(
    scheduled_task,
    trigger=CronTrigger(hour=8, minute=0),
    id="daily_task",
    name="Zadanie codzienne"
)

scheduler.add_job(
    scheduled_task,
    trigger=CronTrigger(minute=0),
    id="hourly_task",
    name="Zadanie godzinowe"
)

scheduler.add_job(
    scheduled_task,
    trigger=CronTrigger(day_of_week="mon", hour=9, minute=0),
    id="weekly_task",
    name="Zadanie tygodniowe"
)

scheduler.start()

Ten przykład uruchamia zadanie codziennie o 8:00, co godzinę oraz w poniedziałki o 9:00. APScheduler daje większą kontrolę niż schedule, dlatego lepiej nadaje się do automatyzacji działających stale w tle.

APScheduler z obsługą błędów

W prawdziwych automatyzacjach zadania mogą się nie udać. API może nie odpowiedzieć, baza danych może być chwilowo niedostępna, a plik może nie istnieć. Dlatego warto dodać obsługę błędów i logowanie.

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import logging
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

logging.basicConfig(
    filename="scheduler.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def job_with_error_handling():
    try:
        logging.info("Start zadania")

        print(f"[{datetime.now()}] Wykonuję zadanie")

        result = 1 + 1
        logging.info(f"Zadanie zakończone poprawnie. Wynik: {result}")

    except Exception as error:
        logging.exception(f"Błąd w zadaniu: {error}")

scheduler = BlockingScheduler()

scheduler.add_job(
    job_with_error_handling,
    trigger=CronTrigger(minute="*/5"),
    id="periodic_job",
    name="Zadanie co 5 minut",
    max_instances=1
)

scheduler.start()

Parametr max_instances=1 zabezpiecza przed uruchomieniem kolejnej instancji zadania, jeśli poprzednia jeszcze się nie zakończyła. To ważne przy zadaniach, które mogą trwać dłużej niż planowany odstęp między uruchomieniami.

Cron i Python

cron to systemowy harmonogram zadań dostępny w systemach Linux i macOS. Jest bardzo często używany na serwerach do cyklicznego uruchamiania skryptów.

Najpierw przygotuj skrypt, np. report.py:

Python
Kopiuj
1
2
3
4
5
6
7
from datetime import datetime
from pathlib import Path

log_file = Path("cron_report.log")

with open(log_file, "a", encoding="utf-8") as file:
    file.write(f"Skrypt uruchomiony: {datetime.now()}\n")

Następnie otwórz edycję crona:

Bash
Kopiuj
1
crontab -e

Przykład uruchamiania skryptu codziennie o 8:00:

CRON
Kopiuj
1
0 8 * * * /usr/bin/python3 /home/user/project/report.py

Przykład uruchamiania co 15 minut:

CRON
Kopiuj
1
*/15 * * * * /usr/bin/python3 /home/user/project/check_status.py

Przykład uruchamiania w każdy poniedziałek o 9:30:

CRON
Kopiuj
1
30 9 * * 1 /usr/bin/python3 /home/user/project/weekly_report.py

Cron jest stabilny i prosty, ale nie jest częścią Pythona. Dobrze sprawdza się, gdy chcesz uruchamiać skrypt z zewnątrz, bez utrzymywania stale działającego procesu Pythona.

Cron a środowisko wirtualne Python

W projektach Pythonowych często używasz środowiska wirtualnego. Wtedy w cronie warto wskazać interpreter z katalogu venv.

Przykład:

CRON
Kopiuj
1
0 8 * * * /home/user/project/venv/bin/python /home/user/project/report.py

Możesz też przejść do katalogu projektu przed uruchomieniem skryptu:

CRON
Kopiuj
1
0 8 * * * cd /home/user/project && /home/user/project/venv/bin/python report.py

To ważne, bo cron uruchamia zadania w ograniczonym środowisku. Jeśli skrypt zakłada konkretny katalog roboczy albo zmienne środowiskowe, trzeba jawnie je ustawić.

Asynchroniczne harmonogramy z asyncio

asyncio przydaje się, gdy zadania dużo czekają: na odpowiedź API, odczyt pliku, bazę danych lub sieć. Dzięki asynchroniczności możesz wykonywać kilka operacji jednocześnie bez tworzenia wielu wątków.

Prosty przykład zadania cyklicznego:

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import asyncio
from datetime import datetime

async def async_task():
    print(f"[{datetime.now()}] Start zadania asynchronicznego")
    await asyncio.sleep(1)
    print(f"[{datetime.now()}] Koniec zadania asynchronicznego")

async def periodic_task(interval: int):
    while True:
        await async_task()
        await asyncio.sleep(interval)

asyncio.run(periodic_task(interval=60))

Ten kod uruchamia zadanie co 60 sekund. Różnica względem zwykłej pętli polega na tym, że asyncio może w tym samym czasie obsługiwać także inne zadania.

Uruchamianie wielu zadań równolegle z asyncio

Jeśli chcesz pobierać dane z kilku źródeł jednocześnie, możesz użyć asyncio.gather.

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import asyncio
from datetime import datetime

async def fetch_sales_data():
    print(f"[{datetime.now()}] Pobieram dane sprzedażowe")
    await asyncio.sleep(2)
    return "sales_data"

async def fetch_marketing_data():
    print(f"[{datetime.now()}] Pobieram dane marketingowe")
    await asyncio.sleep(3)
    return "marketing_data"

async def fetch_support_data():
    print(f"[{datetime.now()}] Pobieram dane supportu")
    await asyncio.sleep(1)
    return "support_data"

async def run_parallel_tasks():
    results = await asyncio.gather(
        fetch_sales_data(),
        fetch_marketing_data(),
        fetch_support_data()
    )

    print(f"[{datetime.now()}] Wyniki: {results}")

asyncio.run(run_parallel_tasks())

Zamiast czekać na każde zadanie osobno, program uruchamia je równolegle. To przyspiesza automatyzacje, które wykonują dużo operacji wejścia-wyjścia.

Cykliczne zadania równoległe z asyncio

Możesz też uruchomić kilka harmonogramów asynchronicznych jednocześnie. Każdy może mieć inny interwał.

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio
from datetime import datetime

async def task(name: str, interval: int):
    while True:
        print(f"[{datetime.now()}] Wykonuję: {name}")
        await asyncio.sleep(interval)

async def main():
    await asyncio.gather(
        task("sprawdzenie API", 30),
        task("synchronizacja danych", 60),
        task("czyszczenie cache", 120)
    )

asyncio.run(main())

To dobry wzorzec, gdy jeden proces ma wykonywać kilka niezależnych czynności w tle.

Harmonogram zadań z retry

W zadaniach cyklicznych warto dodać mechanizm ponawiania, bo błędy tymczasowe są normalne. Poniżej prosty przykład retry dla funkcji asynchronicznej.

Python
Kopiuj
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import asyncio
from datetime import datetime

async def unstable_task():
    print(f"[{datetime.now()}] Próba wykonania zadania")
    await asyncio.sleep(1)

    # tutaj normalnie byłoby np. zapytanie do API
    return "OK"

async def run_with_retry(retries: int = 3, delay: int = 5):
    for attempt in range(1, retries + 1):
        try:
            result = await unstable_task()
            print(f"Zadanie zakończone: {result}")
            return result

        except Exception as error:
            print(f"Błąd próby {attempt}: {error}")

            if attempt == retries:
                raise

            await asyncio.sleep(delay)

async def periodic_task():
    while True:
        await run_with_retry()
        await asyncio.sleep(60)

asyncio.run(periodic_task())

Retry chroni automatyzację przed zatrzymaniem przez pojedynczy problem z siecią lub chwilową niedostępność usługi.

Jak wybrać narzędzie do harmonogramu?

Wybór narzędzia zależy od projektu.

schedule sprawdzi się, gdy potrzebujesz prostego harmonogramu w małym skrypcie. Ma czytelną składnię i niski próg wejścia.

APScheduler będzie lepszy, gdy potrzebujesz bardziej zaawansowanych reguł, kontroli instancji, wielu zadań i stabilnego procesu działającego w tle.

cron jest dobrym wyborem na serwerze, gdy chcesz uruchamiać skrypt z zewnątrz i nie potrzebujesz stale działającej aplikacji Python.

asyncio przyda się wtedy, gdy zadania są asynchroniczne, równoległe i często czekają na operacje sieciowe.

W praktyce te podejścia można łączyć. Na przykład cron może raz dziennie uruchamiać skrypt Python, a ten skrypt może w środku używać asyncio do równoległego pobierania danych.

Dobre praktyki przy harmonogramach zadań

Przy automatyzacji zadań czasowych warto zadbać o niezawodność. Samo uruchomienie funkcji o danej godzinie to za mało.

Najważniejsze praktyki:

• loguj start, koniec i błędy zadania,
• ustawiaj limity czasu dla zapytań do API,
• unikaj nakładania się kilku instancji tego samego zadania,
• zapisuj wyniki działania do pliku, bazy lub systemu logów,
• używaj zmiennych środowiskowych zamiast wpisywać dane dostępowe w kodzie,
• testuj zadanie ręcznie przed dodaniem go do harmonogramu,
• przewiduj ponawianie po błędach tymczasowych,
• monitoruj, czy zadanie faktycznie się wykonuje,
• pilnuj stref czasowych,
• twórz osobne funkcje dla logiki zadania i samego harmonogramu.

Dzięki temu harmonogram będzie łatwiejszy w utrzymaniu i bezpieczniejszy w produkcji.

Podsumowanie

Harmonogramy zadań w Pythonie pozwalają zmienić zwykłe skrypty w automatyzacje działające cyklicznie. Możesz użyć prostego schedule, zaawansowanego APScheduler, systemowego cron albo asynchronicznego asyncio, zależnie od tego, jak złożony jest projekt.

Do prostych lokalnych zadań wystarczy schedule. Do większych procesów lepszy będzie APScheduler. Na serwerach bardzo dobrze sprawdza się cron, a przy wielu operacjach sieciowych warto rozważyć asyncio.

Jeśli chcesz zacząć praktycznie, wybierz jedno powtarzalne zadanie, np. codzienny raport albo sprawdzanie statusu API. Najpierw napisz funkcję, uruchom ją ręcznie, dodaj logowanie, a dopiero potem podepnij ją pod harmonogram.

➡️ Następny artykuł

Po opanowaniu harmonogramów naucz się zarządzać zadaniami i pipeline’ami:

Lista zadań i pipeline'y w Pythonie: makefile/invoke/fabric — standardyzacja komend, tworzenie powtarzalnych procesów i automatyzacja workflow’ów budowania i wdrażania.