How to Run a Mock Cron Pipeline Scheduler in Python

This code schedules a mock pipeline job to run every 2 seconds and hourly at :30 using the schedule library, then runs pending tasks for 10 seconds.

Easy Python 3.9+ Aug 9, 2026 Data pipelines & processing 13 views 0 copies

schedule cron pipeline automation
Requires third-party packages — install first
pip install schedule

Python code

20 lines
Python 3.9+ 
import time
import schedule
from datetime import datetime


def run_pipeline():
    print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Pipeline executed")


schedule.every(2).seconds.do(run_pipeline)
schedule.every().hour.at(":30").do(run_pipeline)

print("Scheduler started. Press Ctrl+C to stop.")
end_time = time.time() + 10

while time.time() < end_time:
    schedule.run_pending()
    time.sleep(1)

print("Scheduler stopped after 10 seconds.")

Output

stdout 
Scheduler started. Press Ctrl+C to stop.
2025-04-23 10:15:00 - Pipeline executed
2025-04-23 10:15:02 - Pipeline executed
2025-04-23 10:15:04 - Pipeline executed
2025-04-23 10:15:06 - Pipeline executed
2025-04-23 10:15:08 - Pipeline executed
Scheduler stopped after 10 seconds.

How it works

The schedule library lets you define jobs with a human-readable syntax, like every(2).seconds or every().hour.at(":30"). In the loop, schedule.run_pending() checks which jobs are due and runs them. The time.sleep(1) keeps the loop from spinning too fast and wasting CPU. This pattern simulates a lightweight cron-like scheduler inside your Python process, useful for mock pipeline runs or simple automation before moving to production orchestration.

Common mistakes

  • Forgetting to call `schedule.run_pending()` inside the loop, so jobs never run.
  • Using `time.sleep(0)` which causes CPU spin and high usage.
  • Not defining a stop condition, making the script run forever—add an end time or Ctrl+C handler.

Variations

  1. Use a while True loop with keyboard interrupt to run indefinitely until stopped.
  2. Replace the print with a real function that calls your pipeline or sends a message to a queue.

Real-world use cases

  • Simulating a cron job on a local machine without setting up a real scheduler, for testing pipeline logic.
  • Running a periodic health check or data refresh inside a long-lived Python service.
  • Creating a lightweight mock job for integration tests that need time-based triggers.

Sponsored

Run locally

This sample needs third-party packages, so it cannot run in the browser IDE. Copy the code above, install the packages shown at the top, then run it in your own Python environment.

More from Data pipelines & processing

Keep learning

Related tutorials and quizzes for this topic.

Tutorials

Quizzes