schedcore 0.3.3

A lightweight, thread-safe, and zero-dependency concurrent task scheduling engine for Python.

schedcore

A lightweight, thread-safe, and zero-dependency concurrent task scheduling engine for Python.

schedcore allows you to schedule tasks to execute in the future, handle recurring tasks, and process them concurrently using a custom-built, lightweight thread pool—all without the overhead of heavy third-party libraries.

🚀 Features

• Zero Dependencies: Built entirely using Python's standard library (threading, queue, time).
• Custom Execution Engine: Uses a custom ThreadPool designed for absolute bare-metal performance.
• Thread-Safe Architecture: Protected by condition variables and thread-safe queues to prevent race conditions and spurious wakeups.
• Configurable error handling: uses callback function for error handling which can be supplied via set_error_handler()
• Graceful Shutdown: Implements the "Poison Pill" pattern to ensure all background OS threads terminate cleanly without leaking memory.
• Modern Python: Fully type-hinted and compatible with Python 3.10+.

🛠️ Installation

This project uses uv, you can easily set it up for local development.

1. Clone the repository:

   git clone [https://github.com/ckalandk/scheduler.git](https://github.com/ckalandk/schedcore.git)
   cd schedcore
   

2. Install the package in editable mode using uv:

    uv sync

📖 Quick Start

The following example, demonstrate how to start a scheduler, submit tasks and gracefully shut it down

import time
from schedcore import Scheduler, Task

# 1. Define your tasks
def greet(name: str):
    print(f"[{time.strftime('%X')}] Hello, {name}!")

def recurring_ping():
    print(f"[{time.strftime('%X')}] Ping...")

def raise_error():
    raise ValueError("Task raised and exception")

def error_handler(exception, task):
    print(f"Task: {task} didn't execute du to an exception: {exception}")

if __name__ == '__main__':
    scheduler = Scheduler(workers=4)
    scheduler.set_error_handler(error_hander)

    scheduler.start()

    print(f"[{time.strftime('%X')}] Scheduler started.")

    # Run this task once
    task_one = Task(interval=2, repeat=False, func=greet, name="Alice")
    scheduler.schedule(task_one)

    # Runs repeatedly, every 1.5 seconds
    task_two = Task(interval=1.5, repeat=True, func=recurring_ping)
    scheduler.schedule(task_two)

    # Use error handler to manage failed to execute task
    task_three = Task(interval=3, repeat=False, func=raise_error)
    scheduler.schedule(task_three)
    try:
        time.sleep(6)
    except KeyboardInterrupt:
        pass
    finally:
        # 5. Cleanly shut down all background threads
        print("Shutting down gracefully...")
        scheduler.request_stop()

🧠 Architecture Overview

This library is split into three core components:

1. Task: A lightweight container that holds the target function, (stored internally in a functools.partial), and its timing requirements (timeout, repeat).

2. ThreadPool: A custom-built worker pool utilizing queue.Queue. It bypasses the heavy machinery of concurrent.futures.ThreadPoolExecutor for fire-and-forget tasks, ensuring maximum throughput.

3. Scheduler: The orchestrator. It uses a queue.PriorityQueue and a threading.Condition to precisely calculate time deltas and hand off tasks to the ThreadPool at the exact right microsecond.

🧪 Running Tests

To run the test suite:

uv run pytest -v

📄 License

MIT License. See LICENSE for more information.