Metadata-Version: 2.4
Name: interlocking
Version: 0.2.1
Summary: A lightweight task dispatcher with explicit control.
Keywords: task-runner,batch,exclusive-lock,cli,mutex,distributed-lock
Author: alpine
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: Utilities
Requires-Dist: boto3>1.40.0,<2 ; extra == 'aws'
Requires-Python: >=3.11
Provides-Extra: aws
Description-Content-Type: text/markdown

# Interlocking

> One track, one train.  
> Easy to run. Explicit to control.

Interlocking is a framework for managing background task execution in distributed environments.
The name comes from the railway interlocking system, which prevents two trains from entering the same track at the same time. Exclusive control is at the core of the library.
Using a railway metaphor, it manages the lifecycle of batch tasks (exclusive control, run records, and CLI integration) in a single, consistent way.

## Features

- **Exclusive control**: The same task never runs concurrently.
- **Run records**: Stores the result of the most recent run and hands it over to the next one (`Coupler`).
- **CLI integration**: Builds a command line from type hints and reports the result through the exit code.
- **Pluggable**: Extend behavior with a `Station` (storage backend) or `TrainMiddleware`.

## Installation

```bash
pip install interlocking
```

To use S3 as the storage backend:

```bash
pip install "interlocking[aws]"
```

Requires Python 3.11 or later.

## Quick start

```python
# tasks.py
from typing import Annotated
from interlocking import Line, Coupler

line = Line()

@line.track(name="daily", help="Daily job")
def daily_task(c: Coupler, target_date: Annotated[str, "Target date (YYYY-MM-DD)"]) -> dict:
    last_id = c.last_data.get("last_id", 0)
    # ... do the work ...
    return {"last_id": 9999}
```

```python
# main.py
import sys

from interlocking import TrainDispatcher, LocalStation, InterlockingCLI

dispatcher = TrainDispatcher(
    station=LocalStation(base_path=".interlocking"),
    lines={"batch": "tasks:line"},
)

cli = InterlockingCLI(dispatcher=dispatcher, prog="my-task", description="Batch tasks")
sys.exit(cli.execute(cli.parse_args()))  # report the result via exit code
```

```bash
# `start` makes the intent to run explicit: without it, nothing runs
python main.py start batch daily --target-date 2026-04-01

# Release a stuck lock (a separate management operation)
python main.py abort batch daily
```

## License

MIT