Skip to content

README.md

Latest commit

 

History

History
112 lines (84 loc) · 3.95 KB

README.md

File metadata and controls

112 lines (84 loc) · 3.95 KB

🐍 python-utils-toolkit

Python License: MIT Pure Python CI Changelog

A collection of practical, production-ready Python utilities built while working on AI agent systems and automation pipelines. Pure Python — zero external dependencies.

Modules

Module Description
utils/retry.py @retry with exponential backoff + jitter
utils/rate_limiter.py Token-bucket rate limiter (sync + async)
utils/timer.py @timer decorator + Timer context manager
utils/file_helpers.py Safe read/write, atomic saves, JSON helpers
utils/logger.py Structured logging with colour + file rotation
utils/env.py Typed .env / environment variable loader
utils/string_utils.py slugify, truncate, camel_to_snake
utils/datetime_utils.py UTC helpers, humanize_delta, market hours
utils/number_utils.py Currency format, tick rounding, pct_change
utils/list_utils.py chunk, flatten, deduplicate, batch_by
utils/dict_utils.py deep_merge, safe_get, flatten_dict
utils/cache_utils.py TTLCache + @memoize for API caching
utils/async_utils.py async retry, timeout, gather_safe
utils/crypto_utils.py HMAC-SHA256 signing, nonce generation
utils/validation_utils.py Guard clauses, symbol/email validation

Install

pip install "git+https://github.com/GhravenLabs/python-utils-toolkit"

Or for local development:

git clone https://github.com/GhravenLabs/python-utils-toolkit
cd python-utils-toolkit
pip install -e ".[dev]"
python examples/retry_demo.py

Examples

Retry with backoff

from utils.retry import retry

@retry(max_attempts=3, base_delay=1.0, exceptions=(ConnectionError,))
def fetch_resource(resource_id: str) -> dict:
    return api.get(resource_id)

Rate limiter (API calls)

from utils.rate_limiter import RateLimiter

limiter = RateLimiter(calls_per_second=5)
for item in work_queue:
    limiter.acquire()
    result = api.process(item)

Sign an API request (HMAC)

from utils.crypto_utils import hmac_sha256, timestamp_nonce

payload = f"user=alice&action=export&timestamp={timestamp_nonce()}"
signature = hmac_sha256(api_secret, payload)

Cache API responses (avoid rate limits)

from utils.cache_utils import memoize

@memoize(ttl=30)
def get_status(service: str) -> dict:
    return api.fetch_status(service)

Async retry for streaming connections

from utils.async_utils import async_retry

@async_retry(max_attempts=3, base_delay=0.5)
async def fetch_events(channel: str) -> dict:
    return await ws.get_events(channel)

Why

Built for multi-agent AI systems and automation pipelines. When you're calling external APIs, running LLM chains, and processing live streams, you need reliable retry logic, rate limiting, and clean logging — without heavy frameworks.

Contributing

Found a bug or want to add a utility? See CONTRIBUTING.md. PRs welcome — please include a test for any new module.

Changelog

See CHANGELOG.md for release history.

AI Assistance Transparency

I use AI-assisted development tools, including Codex and Claude, while building and maintaining this project. All code, design decisions, testing, commits, and releases are reviewed and shipped by me as the repository owner; AI tools are not listed as project contributors.

License

MIT — see LICENSE