shutilspy 0.1.1

Common Python utilities including DAG task framework, caching, rate limiting, RW locks, and parameter helpers

shutilspy

Python utilities including DAG task framework, caching, rate limiting, RW locks, and parameter helpers.

Features

• DAG Framework - Build and execute directed acyclic graph workflows with sync/async tasks, routing, streaming, and service tasks
• Caching - TTL and LRU caches with persistent storage (lzma+pickle), auto-save by step/interval, and decorator support
• Rate Limiting - Token-bucket, QPS, and concurrency limiters with sync/async support and decorator
• RW Locks - Read-write locks for both sync (RWLock) and async (AsyncRWLock) code
• Smart Lock - Adaptive lock (SmartLock, SmartRWLock) with automatic strategy selection based on operation duration
• Parameter Helpers - Dataclass utilities for serialization (asdict, asjson), deserialization (dict_to_dataclass), and hidden fields (Hide/HIDE)
• Image Size - Get image dimensions from binary headers without full decoding (GIF, PNG, JPEG, TIFF, SVG, WebP)

Installation

pip

# without DAG visualizer
pip install shutilspy

# with DAG visualizer
pip install shutilspy[visualizer]

conda

# without DAG visualizer
conda install shlllshlll::shutilspy
# with DAG visualizer
conda install shlllshlll::shutilspy-visualizer

Optional dependencies

pip install shutilspy[visualizer]  # Flask-based DAG visualization

Quick Examples

Singleton

from shutils import singleton

@singleton
class MyClass:
    pass

assert MyClass() is MyClass()

Caching

from shutils import cached

@cached(backend="ttl", ttl=300)
def expensive(x):
    return x * 2

DAG Pipeline

from shutils.dag import DAG, AsyncFunctionTask, Executor

dag = DAG()

async def step_a(ctx):
    ctx.context.sync_white_board["a"] = "done"
    return ctx

async def step_b(ctx):
    val = ctx.context.sync_white_board.get("a")
    ctx.context.sync_white_board["b"] = f"{val}_processed"
    return ctx

dag.add_task(AsyncFunctionTask(step_a, name="A"))
dag.add_task(AsyncFunctionTask(step_b, name="B"), [task_a])
dag.build()

results = await Executor(dag).run()

Rate Limiting

from shutils import limiter

@limiter(calls=10, period=1)
def api_call():
    return "response"

Documentation

Full documentation is available at the docs.

• Getting Started
• DAG Framework Overview
• DAG Guide

Development

pixi install
pixi run test      # Run tests with coverage
pixi run lint      # Run ruff linter
pixi run format    # Run ruff formatter
pixi run docs-serve  # Start docs server

License

MIT