castor-io 0.2.1

Minimalistic, embedded task manager.

Castor

Castor is a minimalistic, embedded task manager built on BeaverDB. It's designed to run background tasks (both process and thread-based) in applications where a complicated, distributed broker like Redis is overkill.

It embraces the minimalistic philosophy of BeaverDB while still being feature-full for the use cases where it makes sense.

Core Philosophy

• Explicit Over Implicit: The library avoids global state. Configuration is handled through explicit Manager objects, making applications more robust and testable.
• Single Responsibility: castor is a task queueing and execution library. Nothing else, nothing more.
• Developer Experience: The API is designed to be intuitive, requiring minimal boilerplate to turn a function into a background task.
• Decoupled Architecture: The application that enqueues a task is fully separate from the worker process that executes it. They communicate only through the shared database file.
• Targeted Concurrency: Provides clear, mandatory choices for both I/O-bound (thread) and CPU-bound (process) concurrency models on a per-task basis.

Installation

pip install castor-io

Quickstart

1. Create your application file

# main.py
import time
from beaver import BeaverDB
from castor import Manager

# 1. Setup the manager
db = BeaverDB("tasks.db")
manager = Manager(db)

# 2. Define a background task
@manager.task(mode='thread')
def send_email(recipient: str):
    """Simulates a background I/O task."""
    print(f"-> Sending email to {recipient}...")
    time.sleep(2)
    print(f"<- Email sent to {recipient}.")
    return {"recipient": recipient, "status": "sent"}

# 3. Dispatch the task (if running this file directly)
if __name__ == "__main__":
    print("--- Dispatching background task ---")
    email_task = send_email.delay("[e-mail]")
    print(f"Dispatched email task with ID: {email_task.id}")

    print("\n--- Waiting for result ---")
    result = email_task.join(timeout=5)
    print(f"Result from email task: {result}")
    print("\n--- Example finished ---")

2. Run the worker from your terminal

The worker needs to know where your Manager instance is. You provide this as an import path.

castor main:manager

You will see the worker start and process the task.

Starting worker server...

3. Run your application

python main.py

You will see tasks being processed in the worker log.

Features

• Task Decorator: A simple @manager.task decorator to turn any function into a background task.
• Execution Modes: Explicitly define tasks as thread (for I/O-bound work) or process (for CPU-bound work).
• Task Handle: Calling .delay() on a task returns a TaskHandle object, allowing you to check the status (.status()) or wait for the result.
• Synchronous and Asynchronous Results: Block for a result with .join() or wait for it asynchronously with .resolve().
• Reliable Backend: Uses beaver-db for a simple and reliable file-based persistence layer.
• CLI Worker: A built-in command-line interface to run the worker server.

Roadmap

This is a work in progress. The immediate roadmap includes:

• Process-based tasks: While the mode='process' is available in the API, the underlying process pool execution is not yet fully implemented. This is the highest priority.
• Monitoring UI: A more advanced terminal-based monitoring dashboard for the worker.
• Retries and error handling: More robust mechanisms for automatic retries and dead-letter queues.