Metadata-Version: 2.4
Name: postgresforge
Version: 1.0.2
Summary: A PostgreSQL-native distributed task queue for Python providing durable jobs, at-least-once execution, leases, and operational tooling without requiring Redis or Kafka.
Project-URL: Homepage, https://github.com/bhaskartank/postgresforge
Project-URL: Repository, https://github.com/bhaskartank/postgresforge
Author: Bhaskar Tank
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.9
Requires-Dist: asyncpg>=0.29.0
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: mypy>=1.10.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Description-Content-Type: text/markdown

# PostgresForge — Background Jobs for Python, Powered by PostgreSQL

> A durable, at-least-once task queue for Python applications that already use PostgreSQL — without requiring a separate message broker.

![Python Version](https://img.shields.io/badge/python-3.9%2B-blue)
![PostgreSQL Version](https://img.shields.io/badge/postgresql-12%2B-blue)
![License](https://img.shields.io/badge/license-MIT-green)
![PyPI Version](https://img.shields.io/pypi/v/postgresforge)

---

## The Problem

A typical modern Python application architecture often looks like this:

```text
FastAPI
   +
PostgreSQL (for user data)
   +
Redis (for queues)
   +
Celery (for workers)
```

If your application already relies heavily on PostgreSQL, adding a completely separate infrastructure component (like Redis or RabbitMQ) just for background jobs increases operational complexity, maintenance overhead, and security footprint.

More importantly, it introduces the **Transactional Dual-Write Problem**:

```text
Application DB update
        +
Background job enqueue
```

If your database transaction rolls back, but you already pushed the email job to Redis, your system is now inconsistent (an email is sent for an action that never occurred). If your application crashes after committing the database transaction but before pushing to Redis, the background job is lost forever.

## The PostgresForge Idea

PostgresForge eliminates this problem by using your existing PostgreSQL database as the durable message broker.

```text
                 Python Application
                        │
                    enqueue()
                        │
                        ▼
               ┌─────────────────┐
               │   PostgreSQL    │
               │                 │
               │ postforge_jobs  │
               │ idempotency     │
               │ workers         │
               └────────┬────────┘
                        │
                  SKIP LOCKED
                        │
             ┌──────────┼──────────┐
             ▼          ▼          ▼
          Worker 1   Worker 2   Worker 3
```

PostgreSQL natively stores the jobs, tracks idempotency, and coordinates distributed workers. No separate queue broker is required.

## Why PostgresForge?

* **PostgreSQL-native**: Jobs live in PostgreSQL alongside your application data.
* **Transactional enqueueing**: Application database changes and job creation happen in the exact same transaction.
* **No additional broker**: Zero need to deploy, monitor, or secure Redis, RabbitMQ, or Celery.
* **Concurrent workers**: Uses PostgreSQL row locking and `FOR UPDATE SKIP LOCKED` to allow multiple distributed workers to claim jobs safely without deadlocks.
* **Failure recovery**: Lease-based timeouts and background heartbeats allow jobs from abruptly failed or OOM-killed workers to be cleanly recovered.
* **Zombie-worker protection**: Cryptographic lease tokens prevent an old, stalled worker from completing a job after it has lost its lease, completely preventing split-brain executions.
* **At-least-once execution**: Guaranteed delivery semantics. Jobs may execute more than once in disaster scenarios, so task handlers should be idempotent.

## When Should I Use It?

PostgresForge is perfect for standard background tasks that need high durability:

* Email and SMS delivery
* Outbound webhooks
* Retryable third-party API calls
* PDF/CSV/Report generation
* Image or document processing
* Delayed notifications
* Data synchronization and cleanup workflows

**Example:**
```text
User creates order
       ↓
BEGIN Transaction
       ├── Insert into orders table
       └── Enqueue confirmation email job
       ↓
COMMIT
       ↓
Worker claims job and sends email
```
If the `COMMIT` fails, the email is never enqueued. Guaranteed consistency.

## When Should I NOT Use It?

PostgresForge is *not* intended to replace Kafka, Redpanda, or dedicated stream processors. Do not use PostgresForge for:

* Massive event streaming (millions of events per second)
* Real-time stream processing or sliding-window analytics
* Long-term event retention (jobs are designed to be completed and cleaned up)
* Sub-millisecond pub/sub notification fanout

*(Note: PostgresForge guarantees at-least-once execution. If your system strictly demands exactly-once execution, you must design your task handlers to be idempotent.)*

## Quick Start

### 1. Installation

```bash
pip install postgresforge
```

### 2. Database Initialization

`pip install postgresforge` installs the Python package and includes the required schema.

You can initialize the PostgresForge tables (`postforge_jobs`, `postforge_idempotency`, `postforge_workers`) using the built-in CLI command. This command is safe to run multiple times and will **not** modify your existing application tables.

```bash
export POSTFORGE_DSN="postgres://user:password@localhost:5432/mydatabase"
postforge init-db
```

### 3. Enqueueing Jobs

```python
import asyncio
import asyncpg
from postforge.queue import enqueue

async def main():
    conn = await asyncpg.connect("postgres://user:password@localhost:5432/db")
    try:
        await enqueue(
            conn,
            queue="emails",
            task_name="send_welcome_email",
            payload={"user_id": 123, "email": "user@example.com"}
        )
    finally:
        await conn.close()

asyncio.run(main())
```

### 4. Running a Worker

```python
import asyncio
from postforge.worker import Worker
from postforge.models import Job

async def send_welcome_email(job: Job):
    print(f"Sending email to {job.payload['email']}...")

async def main():
    worker = Worker(
        dsn="postgres://user:password@localhost:5432/db",
        queue="emails",
        concurrency=10
    )
    worker.register("send_welcome_email", send_welcome_email)
    
    print("Worker running...")
    await worker.start()

asyncio.run(main())
```

## Minimal Mental Model

When enqueueing a job:
* **Queue** (`queue="emails"`): WHERE the job lives (workers listen to specific queues).
* **Task Name** (`task_name="send_welcome_email"`): WHAT operation the worker should run.
* **Payload** (`payload={"user_id": 123}`): WITH WHAT DATA the task should operate on.

## Architecture

```text
Application
    ↓
enqueue()
    ↓
[ PostgreSQL ] → postforge_jobs, postforge_idempotency, postforge_workers
    ↓
Worker (claims via FOR UPDATE SKIP LOCKED)
    ↓
Task Handler
```

**Supporting Mechanisms:**
* `postforge_jobs`: The core queue table.
* `postforge_idempotency`: Prevents duplicate enqueueing via idempotency keys.
* `postforge_workers`: Registry of active workers for CLI observability.
* `leases`: Cryptographic UUIDs issued to workers when they claim jobs.
* `heartbeats`: Periodic background pulses from workers to extend their leases.
* `recovery`: Background sweeps to reclaim jobs whose leases have expired due to worker crashes.
* `cleanup`: Background sweeps to delete permanently failed or successfully completed terminal jobs to prevent database bloat.

## Job Lifecycle

**Standard Execution:**
```text
Enqueued → Available → Claimed → Executing → Completed
```

**Task Failure:**
```text
Executing → Exception Raised → Job Failed → Retry (Scheduled for future) → Available
```

**Worker Crash (OOM / Disconnect):**
```text
Executing → Worker abruptly dies → Heartbeat stops → Lease expires → Recovery sweeps job → Available → Another worker claims job
```

## Reliability Guarantees

* **At-least-once**: A job will successfully execute and `COMPLETE` at least once. (Under rare crash scenarios, a job might run twice).
* **Idempotency**: If you provide an `idempotency_key` during `enqueue()`, PostgresForge atomically ignores duplicate enqueue attempts.
* **Leases**: Workers take exclusive, temporary ownership of a job.
* **Heartbeats**: Active workers continuously extend their leases asynchronously.
* **Fencing**: If a worker loses connection, stalls, or hits an infinite loop, its lease expires. When it finally wakes up and tries to complete the job, its old lease token is rejected, preventing data corruption.

> **Note:** PostgresForge does *not* provide exactly-once execution. Design your handlers to be idempotent.

## Transactional Enqueueing

This is the most powerful feature of PostgresForge. 

```python
async with conn.transaction():
    # 1. Update the application database
    await conn.execute("INSERT INTO users (id, name) VALUES (123, 'Alice')")

    # 2. Enqueue the background job
    await enqueue(
        conn,
        queue="emails",
        task_name="send_welcome_email",
        payload={"user_id": 123}
    )
```

**If the transaction commits**, the user is saved and the job becomes available to workers instantly.
**If the transaction rolls back**, the user is *not* saved, and the job is never enqueued.
Consistency is guaranteed natively by PostgreSQL.

## Retries and Scheduling

### Retries
Jobs accept a `max_attempts` argument (default: 3). If your task handler raises an exception, the worker gracefully catches it, marks the job as failed, and schedules it for a retry. Once `attempts >= max_attempts`, the job enters a permanently failed terminal state.

### Scheduling
You can delay job execution by specifying `scheduled_at`.

```python
from datetime import datetime, timedelta, timezone

await enqueue(
    conn,
    queue="reminders",
    task_name="send_followup",
    payload={"user_id": 123},
    scheduled_at=datetime.now(timezone.utc) + timedelta(days=3)
)
```

## CLI

PostgresForge includes a lightweight CLI to inspect your queues.
*Requires the `POSTFORGE_DSN` environment variable to be set.*

```bash
export POSTFORGE_DSN="postgres://user:password@localhost:5432/db"

postforge queue stats
postforge job list
postforge job list --status failed
postforge job get <job-id>
postforge job retry <job-id>
postforge worker list
```

## PostgresForge vs Alternatives

| Capability                     | PostgresForge | Celery + Redis             | Kafka                              |
| ------------------------------ | ------------- | -------------------------- | ---------------------------------- |
| **PostgreSQL-native**          | Yes           | No                         | No                                 |
| **Separate broker required**   | No            | Yes                        | Yes                                |
| **Python-first**               | Yes           | Yes                        | No                                 |
| **Transactional job enqueue**  | Yes (Native)  | Requires complex outbox    | Different model                    |
| **Background tasks**           | Yes           | Yes                        | Possible, but not primary use case |
| **Event streaming**            | No            | No                         | Yes                                |
| **Long-term event retention**  | No            | No                         | Yes                                |

*The goal of PostgresForge is not to replace Kafka. It is to simplify architecture for teams that already use PostgreSQL and do not want to manage a separate message broker just for background tasks.*

## Benchmarks

PostgresForge is rigorously benchmarked locally to ensure native Postgres locking scales securely.

* **Enqueueing**: ~100k jobs/sec (via batch inserts)
* **Claim Contention**: ~1.7k claims/sec under extreme distributed lock contention.
* **End-to-End Throughput**: ~400 full job lifecycles (enqueue → claim → execute → complete) per second per Python worker process.

**Measured Results vs Expected Goals:**
These numbers represent local stress tests proving that PostgreSQL's `SKIP LOCKED` scales efficiently. Your actual production throughput will depend entirely on your PostgreSQL server capacity, connection pool sizing, network latency, and the duration/complexity of your task execution logic.

## Production Considerations

* **PostgreSQL Connection Capacity**: Workers require a dedicated `asyncpg` connection pool to manage concurrent claims, heartbeats, and recovery loops. Ensure your PostgreSQL `max_connections` (or PgBouncer limits) can handle your worker fleet.
* **Idempotent Handlers**: Because at-least-once delivery is guaranteed, your task handlers *must* be safe to execute multiple times in the event of a worker crash mid-execution.
* **Retention**: Terminal jobs are automatically purged by the cleanup loop after 24 hours to prevent table bloat.
* **Monitoring**: Use the CLI or query `postforge_jobs` directly to monitor queue depth.

## Limitations

* **PostgreSQL Overhead**: Background workloads compete with your normal database workloads for CPU, memory, and IOPS.
* **Not Kafka**: It does not support pub-sub fanout, stream processing, or indefinite event retention.
* **Python Asyncio**: The worker runtime relies heavily on Python's `asyncio` loop. Synchronous, blocking workloads must be delegated to thread pools manually within your handlers, otherwise they will block worker heartbeats and cause lease expiration.

## Roadmap

*While the core queue primitives are stable, the following features are actively being explored:*

* Native FastAPI dependency injection
* Rate limiting per queue or task
* Job dependencies (DAGs)
* Dead-letter queues for long-term failure inspection
* OpenTelemetry / Prometheus metrics integration

## Contributing

We welcome community contributions!

```bash
git clone https://github.com/bhaskartank/postgresforge.git
cd postgresforge
python -m venv venv
source venv/bin/activate
pip install -e ".[dev]"

# Run tests
pytest
```
Feel free to open issues or submit pull requests for bugs, feature requests, or documentation improvements.

## License

PostgresForge is released under the **MIT License**.
