Skip to main content

⚡ NeoTask

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

Python License

PyPI Downloads GitHub Stars

中文 • 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.

Pain Point NeoTask Solution
Task blockage under heavy concurrency Priority queueing & task worker pool isolation
Unreliable long-running tasks Automatic state persistence, health check & retry mechanism
Complex SDK setup Lightweight Pythonic interface with zero friction

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

Scenario Description Recommended Configuration Entry Point
AI Text-to-Image/Video Generation Queue time-consuming tasks to avoid blocking main flow worker_concurrency=3 TaskPool
Batch File Processing Batch operations like transcoding, compression, uploading worker_concurrency=10 TaskPool
Web Scraping Scheduling Distributed scraping to prevent being blocked storage_type="redis" TaskPool
Scheduled Report Sending Send daily reports at 9 AM cron="0 9 * * *" TaskScheduler
Delayed Notifications Send reminders 5 minutes after user action delay_seconds=300 TaskScheduler
Heartbeat Detection Check service health status every 30 seconds interval_seconds=30 TaskScheduler
Background Data Analysis Execute data aggregation tasks at night cron="0 2 * * *" TaskScheduler

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

Method Description
pool.submit(data, priority=2, delay=0) Submit task
pool.wait_for_result(task_id, timeout=300) Wait for result
pool.get_status(task_id) Get status
pool.cancel(task_id) Cancel task
scheduler.submit_delayed(data, delay) Delayed task
scheduler.submit_interval(data, interval) Periodic task
scheduler.submit_cron(data, cron) Cron task

Detailed API documentation can be found here

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

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.

Metadata

Release files for neotask 1.0.2

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for neotask 1.0.2
File Size Uploaded
neotask-1.0.2.tar.gz 275.7 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for neotask 1.0.2
File Interpreter ABI Platform
neotask-1.0.2-py3-none-any.whl Python 3 none any Details

Total release size: 439.0 kB

Release files / neotask-1.0.2.tar.gz

Download URL neotask-1.0.2.tar.gz
Size 275.7 kB
Tags Source
SHA-256 checksum
How to use checksums
9fa49e4fe3be87bfd3f98223752b60b5f35817644bede9c40901775e831c2e62
BLAKE2b-256 checksum
How to use checksums
bd59deb1bf6d21073b4bcfaea806b4631fbd9b3ea6c1c3efe54056a6a4316ed9
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Sep 13, 2026.

Transparency log

Release files / neotask-1.0.2-py3-none-any.whl

Download URL neotask-1.0.2-py3-none-any.whl
Size 163.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums
f379f0865f4b3685ba75e1b1568d2805fd973655aa287aeaf401db825ccdad80
BLAKE2b-256 checksum
How to use checksums
2fa2ea535f4fe7a1bb041aa1488712935d31e1ecb6453c652c9b303577fed5ef
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
Yes
Uploaded via twine/7.0.0 CPython/3.13.14

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

PyPI Publish Attestation

PyPI verified that this artifact, at this checksum, originated from the publisher listed below.

Signed by GitHub Actions, verified by PyPI on Sep 13, 2026.

Transparency log

Release history Release notifications | RSS feed

1.0.4

2 release files

1.0.3

2 release files

This release

1.0.2 This release

2 release files

1.0.1

2 release files

1.0.0

2 release files

0.5.0

2 release files

0.4.1

2 release files

0.4.0

2 release files

0.3.0

2 release files

0.2.0

2 release files

0.1.1

2 release files

0.1.0

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page