leasedkeyq 0.0.4

Async keyed leased queue with FIFO semantics and O(1) operations

leasedkeyq

Async Keyed Leased Queue - An asyncio-friendly in-memory queue combining FIFO semantics, dictionary-style keyed access, and lease-based exclusive processing.

Features

• FIFO Queue Semantics: Process items in order with get()
• Keyed Access: Target specific items with take(key)
• Lease-Based Processing: Exclusive access with explicit ack/release
• O(1) Operations: Constant-time enqueue, dequeue, and key-based removal
• Automatic Timeouts: Optional lease expiration with auto-retry
• Blocking Behavior: Async waiting for items or specific keys
• Type Safe: Full type hints with strict mypy compliance
• Zero Dependencies: Pure Python stdlib implementation

Installation

pip install leasedkeyq

Quick Start

import asyncio
from leasedkeyq import LeasedKeyQueue

async def main():
    # Create queue with 30-second lease timeout
    async with LeasedKeyQueue[str, dict](default_lease_timeout=30.0) as queue:
        # Producer: add items
        await queue.put("task-1", {"action": "process", "data": 123})
        await queue.put("task-2", {"action": "send", "data": 456})

        # Consumer: FIFO consumption
        key, value, lease = await queue.get()
        print(f"Processing {key}: {value}")

        try:
            # Do work...
            await process(value)
            # Success: acknowledge
            await queue.ack(lease)
        except Exception:
            # Failure: release for retry
            await queue.release(lease, requeue_front=True)

        # Targeted consumption
        key, value, lease = await queue.take("task-2")
        print(f"Specifically got {key}: {value}")
        await queue.ack(lease)

asyncio.run(main())

Core Concepts

States

Each key is in exactly one state:

• ABSENT: Not in queue
• AVAILABLE: Ready for consumption
• IN_FLIGHT: Leased to a consumer

API Overview

Producer API

await queue.put(key, value, if_in_flight="update")  # update|reject|buffer

Consumer API

# FIFO consumption
key, value, lease = await queue.get(timeout=10.0, lease_timeout=30.0)

# Keyed consumption
key, value, lease = await queue.take("specific-key", timeout=10.0)

Lease Control

await queue.ack(lease)                              # Permanent removal
await queue.release(lease, requeue_front=True)      # Retry

Introspection

value = await queue.peek("key")
has_key = await queue.contains("key")
keys = await queue.available_keys()
inflight = await queue.inflight_keys()
size = await queue.qsize()

In-Flight Policies

Control behavior when putting a key that's currently leased:

• update (default): Update the in-flight value
• reject: Raise KeyAlreadyInFlightError
• buffer: Enqueue a second copy to available

await queue.put("key", new_value, if_in_flight="update")

Lease Timeouts

Prevent stuck items with automatic lease expiration:

# Queue-wide default timeout
queue = LeasedKeyQueue[str, int](default_lease_timeout=30.0)
await queue.start()

# Per-lease override
key, value, lease = await queue.get(lease_timeout=60.0)

Expired leases are automatically released to the front of the queue for retry.

Complexity Guarantees

Operation	Time Complexity
put()	O(1)
get()	O(1)
take(key)	O(1)
ack()	O(1)
release()	O(1)

Achieved through intrusive doubly-linked list with direct node references.

Use Cases

• Task Queues: FIFO processing with retry on failure
• Job Scheduling: Target specific jobs while maintaining order
• Rate Limiting: Lease-based exclusive access prevents double-processing
• Event Processing: Handle events by ID with guaranteed exclusivity
• Workflow Engines: Track in-flight work with timeout-based recovery

Documentation

• API Documentation
• Design Specification
• Examples

Development

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

# Run tests
pytest

# Type checking
mypy src/

# Linting
ruff check src/ tests/

License

MIT License - see LICENSE file for details.

Contributing

Contributions welcome! Please open an issue or PR on GitHub.