⚡ NeoTask

Distributed Task Scheduling System / 轻量级 Python 异步分布式任务队列管理器

中文 • Website • Documentation • PyPI

🚀 Lightweight & Async: Pure Python task queue manager designed for time-consuming jobs (AI generation, video processing, data scraping, etc.)
📦 No External Services: No extra deployment required (no Redis/PostgreSQL needed). Ships with two base dependencies
⏰ Flexible Scheduling: Native support for scheduled, periodic (cron-like), and delayed tasks
⚡ Quick Integration: pip install neotask + few lines of configuration

💡 Why NeoTask?

NeoTask is designed to solve complex task orchestration, async queueing, and priority management for modern distributed systems and AI multi-agent pipelines.

Features

• No External Services - Pure Python implementation, no Redis/PostgreSQL required
• Immediate Tasks - Supports priority scheduling, high-priority tasks execute first
• Scheduled Tasks - Supports delayed execution, fixed intervals, and Cron expressions
• Asynchronous Concurrency - Based on asyncio, multi-worker concurrent processing
• Automatic Retry - Failed tasks automatically retry with configurable attempts
• Persistence - Multiple storage backends: Memory/SQLite/Redis
• Distributed Support - Redis shared queue, distributed lock, node heartbeat, and failed-task reclaim
• Event Callbacks - Supports task lifecycle event listeners

Use Cases

Architecture & Evolution

graph TB
    subgraph User["User Application Layer"]
        APP[User Code]
    end
    
    subgraph NeoTask["NeoTask Core"]
        TP[TaskPool<br/>Immediate Task Entry]
        TS[TaskScheduler<br/>Scheduled Task Entry]
        
        subgraph Core["Shared Core Components"]
            LM[LifecycleManager<br/>Task Lifecycle Management]
            QS[QueueScheduler<br/>Priority + Delayed Queue]
            WP[WorkerPool<br/>Worker Pool/Concurrency Control]
            FM[FutureManager<br/>Async Wait/Result Callback]
        end
        
        subgraph Internal["Internal Components"]
            EB[EventBus<br/>Event Bus]
            MC[MetricsCollector<br/>Metrics Collection]
            LF[LockFactory<br/>Distributed Lock]
        end
        
        EX[TaskExecutor<br/>User Business Logic]
    end
    
    subgraph Storage["Storage Layer"]
        MEM[MemoryStorage]
        SQLITE[(SQLiteStorage)]
        REDIS[(RedisStorage)]
    end
    
    APP -->|Immediate Task| TP
    APP -->|Scheduled Task| TS
    TS -->|Delegates| TP
    TP --> LM
    TP --> QS
    TP --> WP
    TP --> FM
    
    LM --> MEM
    LM --> SQLITE
    LM --> REDIS
    
    WP --> EX
    WP --> EB
    WP --> MC
    WP --> LF

Development Roadmap

timeline
    title NeoTask Architecture Evolution Roadmap
    
    section v0.1
        Basic Task Pool : Local Memory Queue
                        : Async Execution Engine
                        : Memory/SQLite Storage
                        
    section v0.2
        Observability : Event Bus
                      : Metrics Collection
                      : Health Check
                      
    section v0.3
        Scheduled Tasks : Delayed Queue/Time Wheel
                        : Periodic Tasks
                        : Cron Expression
                        
    section v0.4
        Distributed Base : Redis Shared Queue
                         : Distributed Lock
                         
    section v0.5
        Performance Optimization : Prefetch Mechanism
                                : Batch Operations
                                : Connection Pool

    section v1.0
        High Availability : Watchdog Renewal
                          : Timeout Detection
                          : Automatic Fault Recovery

    section v1.5
        Task Orchestration : DAG Workflow
                           : Conditional Branch
                           : Parallel Execution

    section v2.0
        Enterprise Features : Independent Web UI
                            : Multi-Tenancy Isolation
                            : Prometheus Integration

Quick Start

Installation

Requirements: Python ≥ 3.8. The base package depends on aiosqlite (SQLite storage) and croniter (Cron expressions); Redis, psutil, and the Web UI are optional extras.

# Basic installation
pip install neotask

# With Redis distributed support
pip install neotask[redis]

# Full installation
pip install neotask[full]

Immediate Tasks (TaskPool)

from neotask import TaskPool

async def process(data):
    return {"result": "done", "data": data}

# Create task pool
pool = TaskPool(executor=process)

# Submit task
task_id = pool.submit({"id": 123})

# Wait for result
result = pool.wait_for_result(task_id)

pool.shutdown()

Scheduled Tasks (TaskScheduler)

from neotask import TaskScheduler

scheduler = TaskScheduler(executor=process)

# Execute after 60 seconds delay
scheduler.submit_delayed({"id": 123}, delay_seconds=60)

# Execute every 5 minutes
scheduler.submit_interval({"id": 123}, interval_seconds=300)

# Execute daily at 9 AM
scheduler.submit_cron({"id": 123}, "0 9 * * *")

scheduler.shutdown()

Using Context Manager

with TaskPool(executor=process) as pool:
    task_id = pool.submit({"id": 123})
    result = pool.wait_for_result(task_id)

Using Event Callbacks

from neotask import TaskPool

async def on_task_created(event):
    print(f"Task created: {event.task_id}")

async def on_task_completed(event):
    print(f"Task completed: {event.task_id}, Result: {event.data}")

async def on_task_failed(event):
    print(f"Task failed: {event.task_id}, Error: {event.data}")

pool = TaskPool(executor=my_executor)
pool.start()

# Register event callbacks
pool.on_created(on_task_created)
pool.on_completed(on_task_completed)
pool.on_failed(on_task_failed)

task_id = pool.submit({"test": "event"})
result = pool.wait_for_result(task_id)

API Reference

Detailed API documentation can be found here

Usage Notes

A few behaviors are easy to trip over, so they are documented explicitly:

• get_result vs wait_for_result: wait_for_result(task_id) returns the executor's raw value, while get_result(task_id) returns a metadata wrapper dict {task_id, status, result, error, created_at, completed_at} — the real payload is under result.
• Event keys: lifecycle event types are namespaced as task.created / task.started / task.completed / task.failed / task.cancelled / task.retry (not bare created). Prefer the pool.on_completed(...) helpers.
• Graceful shutdown scope: shutdown(graceful=True) drains tasks that are already enqueued (pending + running) before returning. Unexpired delayed tasks and periodic/cron schedules are not fired during shutdown — cancel periodic schedules first (cancel_periodic) or use graceful=False for an immediate return.
• Crash / restart recovery: auto-recovery re-runs tasks that were already enqueued (their ids live in the persisted queue). A record created only via create_pending_task() and never pushed through enqueue_task() is not part of the queue, so it will not auto-run after a restart — enqueue it explicitly.
• Periodic task persistence: set SchedulerConfig(enable_persistence=True, storage_type="sqlite"|"redis") to persist interval / cron periodic tasks and auto-restore them after a restart; slots missed while down are handled by each task's missed_policy. With storage_type="memory" the store is in-process only and cannot survive a restart.
• Minimum interval granularity: periodic tasks are driven by a scan loop with SchedulerConfig.scan_interval (default 1.0s). submit_interval with an interval smaller than scan_interval may skip fires; lower scan_interval or enable the time wheel (enable_time_wheel=True) for finer granularity.

Usage Notes

A few behaviors are easy to trip over, so they are documented explicitly:

• get_result vs wait_for_result: wait_for_result(task_id) returns the executor's raw value, while get_result(task_id) returns a metadata wrapper dict {task_id, status, result, error, created_at, completed_at} — the real payload is under result.
• Event keys: lifecycle event types are namespaced as task.created / task.started / task.completed / task.failed / task.cancelled / task.retry (not bare created). Prefer the pool.on_completed(...) helpers.
• Graceful shutdown scope: shutdown(graceful=True) drains tasks that are already enqueued (pending + running) before returning. Unexpired delayed tasks and periodic/cron schedules are not fired during shutdown — cancel periodic schedules first (cancel_periodic) or use graceful=False for an immediate return.
• Crash / restart recovery: auto-recovery re-runs tasks that were already enqueued (their ids live in the persisted queue). A record created only via create_pending_task() and never pushed through enqueue_task() is not part of the queue, so it will not auto-run after a restart — enqueue it explicitly.
• Minimum interval granularity: periodic tasks are driven by a scan loop with SchedulerConfig.scan_interval (default 1.0s). submit_interval with an interval smaller than scan_interval may skip fires; lower scan_interval or enable the time wheel (enable_time_wheel=True) for finer granularity.

Configuration Example

from neotask import TaskPool, TaskPoolConfig

config = TaskPoolConfig(
    worker_concurrency=10,      # Number of concurrent workers
    max_retries=3,              # Number of retries
    storage_type="sqlite",      # Storage type
)

pool = TaskPool(executor=process, config=config)

Detailed usage examples can be found here

Contribution Guide

Setting Up Development Environment

# Clone repository
git clone https://github.com/neopen/neotask.git
cd neotask

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/

# View test coverage
pytest --cov=neotask tests/

# Run specific module tests
pytest tests/test_task_pool.py -v
pytest tests/test_task_scheduler.py -v

Contribution Workflow

Welcome to submit Issues and Pull Requests

1. Fork the project
2. Create a feature branch (git checkout -b feature/amazing)
3. Commit changes (git commit -m 'Add amazing feature')
4. Push branch (git push origin feature/amazing)
5. Submit Pull Request

Code Style

• Follow PEP 8 code style
• Add appropriate type annotations
• Write unit tests for new features (coverage ≥ 80%)
• Update relevant documentation and example code
• Commit messages follow Conventional Commits

Issue Reporting

• Submit Issue: https://github.com/neopen/neotask/issues
• Feature Suggestions: Use Enhancement label
• Bug Reports: Use Bug label and provide reproduction steps
• Security Vulnerabilities: Please send email directly to the author's email

License

MIT License © 2026 NeoPen

Acknowledgments

Thanks to all contributors and the open source community for their support.

Contact

• GitHub：https://github.com/neopen/neotask
• Email：helpenx@gmail.com
• Docs：https://task.helpenx.com/docs/
• Blog: https://pengline.cn/