skiptask 0.1.0
A Django-based distributed task queue with PostgreSQL advisory locks
Requires Python: >=3.9
Keywords: django, task-queue, background-jobs

# SkipTask

A Django-based distributed task queue system that uses PostgreSQL advisory locks for reliable, distributed task processing.

## Features

- **Distributed Processing**: Uses PostgreSQL advisory locks for safe distributed task processing
- **Multi-Process Workers**: Built-in support for multi-process worker pools
- **Retry Logic**: Automatic retry of failed tasks with configurable attempts
- **Priority Queues**: Support for task prioritization and multiple queues
- **Thread Safety**: Proper database connection isolation and cleanup
- **Django Integration**: Native Django app with management commands and admin interface

## Installation

Install from PyPI:

```bash
pip install skiptask
```

Or install with development dependencies:

```bash
pip install skiptask[dev]
```

## Quick Start

### 1. Add to Django Settings

Add `skiptask` to your `INSTALLED_APPS`:

```python
INSTALLED_APPS = [
    # ... your other apps
    'skiptask',
]
```

### 2. Configure Database

SkipTask requires PostgreSQL for distributed locking. Update your database configuration:

```python
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'your_db_name',
        'HOST': 'localhost',
        'PORT': '5432',
        'USER': 'your_user',
        'PASSWORD': 'your_password',
    }
}
```

### 3. Run Migrations

```bash
python manage.py migrate
```

### 4. Create and Enqueue Tasks

```python
from skiptask.handlers import enqueue_callable

# Define a task function
def send_email(to_email, subject, message):
    # Your email sending logic here
    print(f"Sending email to {to_email}: {subject}")
    return f"Email sent to {to_email}"

# Enqueue the task
queueable, task = enqueue_callable(
    send_email,
    args=['user@example.com', 'Hello', 'Welcome to our service!'],
    queue='email',
    priority=100
)
```

### 5. Run Workers

Start a worker to process tasks:

```bash
python manage.py runworker
```

Or use the standalone worker command:

```bash
skiptask-worker
```

## Advanced Usage

### Custom Handlers

Create custom handlers for different types of tasks:

```python
from skiptask.handlers import Handler, registry

class CustomHandler(Handler):
    def handle(self, task):
        # Your custom task processing logic
        payload = task.payload
        # Process the task...
        return "Task completed"

# Register the handler
registry.register(CustomHandler)
```

### Multiple Queues

Organize tasks into different queues:

```python
# Enqueue to specific queues
enqueue_callable(urgent_task, queue='urgent', priority=1)
enqueue_callable(background_task, queue='background', priority=100)
```

### Retry Configuration

Configure retry behavior:

```python
queueable, task = enqueue_callable(
    potentially_failing_task,
    reattempts=3,  # Retry up to 3 times
    run_timeout=300,  # 5 minute timeout
)
```

### Worker Configuration

Run workers with custom settings:

```python
from skiptask.processing import Worker

# Multi-process worker
worker = Worker(n_worker_processes=4, stop_on_empty_queue=False)
worker.start()
```

## Management Commands

### runworker

Start a worker process:

```bash
python manage.py runworker
```

This command supports Django's auto-reloading for development.

## Models

### Task

Represents a task to be executed:

- `handler`: The handler class to process the task
- `payload`: JSON data containing task parameters
- `status`: Current task status (NEW, COMPLETED, ERRORED)
- `reattempts`: Number of retry attempts allowed
- `run_timeout`: Maximum execution time in seconds

### Queueable

Manages task queuing:

- `task`: Reference to the Task
- `queue`: Queue name (default: 'default')
- `priority`: Task priority (lower numbers = higher priority)
- `in_queue`: Whether the task is available for processing

### Attempt

Tracks task execution attempts:

- `task`: Reference to the Task
- `status`: Attempt status (COMPLETED, ERRORED)
- `value`: Return value from successful execution
- `message`: Error message for failed attempts

## Development

### Running Tests

```bash
pytest
```

### Code Formatting

```bash
black .
```

### Docker Development

The project includes Docker configuration for development:

```bash
# Build development container
docker build --target dev -t skiptask:dev .

# Run with PostgreSQL
docker run -d --name postgres -e POSTGRES_PASSWORD=password postgres
docker run -it --link postgres:db skiptask:dev
```

## Requirements

- Python 3.8+
- Django 3.2+
- PostgreSQL 10+
- psycopg2-binary

## License

MIT License - see LICENSE file for details.

## Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests for new functionality
5. Run the test suite
6. Submit a pull request

## Changelog

See [CHANGELOG.md](CHANGELOG.md) for version history.