pyasynckit 0.1.1

Zero-dependency async and concurrency utilities for Python.

pyasynckit

A zero-dependency toolkit for reliable asynchronous Python programs.

pyasynckit provides small, composable building blocks for common asyncio problems: retrying transient failures, applying backoff, limiting concurrency, and coordinating groups of tasks. It is designed for crawlers, API clients, bots, data pipelines, and services that need predictable async behaviour.

Project status: early development. Retry is the first public utility; queue, rate-limit, timeout, task, and parallel helpers are planned. The project does not claim that unimplemented APIs are available yet.

Why pyasynckit?

• Zero runtime dependencies. The installed library uses only Python's standard library.
• Python 3.8+. The public package supports CPython 3.8 and newer.
• Async-first. APIs use async/await naturally and preserve task cancellation instead of accidentally swallowing it.
• Predictable failure handling. Retry policies validate configuration early, preserve the final original exception, and support bounded delays.
• Readable defaults. A policy should be obvious in a code review without requiring a large configuration framework.

Compared to existing libraries

The Python ecosystem already has mature alternatives for most of these problems. Reach for the existing library when you want its full feature surface; reach for pyasynckit when you want a small, zero-dep, single import.

Need	Established library	pyasynckit
Retry with backoff	tenacity, backoff	Retry — small, async-only, deterministic via injectable sleep / random_fn
Async stream / queue / parallel	aiostream	Planned Queue + parallel (M2)
Rate limiting	asyncio-throttle, aiolimiter	Planned Throttler (M1)
Async memoize / cache	async-lru, cachetools	Planned memoize (M3)
Timeout / deadline	async-timeout	Planned timeout / deadline (M1)

The differentiator is composability (every primitive shares the same async-context-manager shape and injectable clock) and deterministic testability (no time, randomness, or scheduling is hidden behind asyncio internals).

Installation

Install the published package once it is available on PyPI:

python -m pip install pyasynckit

For development from a clone:

git clone https://github.com/Goldhobbang/pyasynckit.git
cd pyasynckit
python -m pip install -e ".[dev]"

Quick start

Retry executes an asynchronous callable again after retryable failures. The times option means the total number of attempts, including the first one.

import asyncio

from pyasynckit import Retry


async def fetch_profile(user_id: str) -> dict:
    # Replace this with an async HTTP client call.
    return {"id": user_id}


async def main() -> None:
    async with Retry(times=3, delay=0.5, backoff="exponential") as retry:
        profile = await retry(fetch_profile, "42")
    print(profile)


asyncio.run(main())

The example attempts the operation at most three times. After failures, it waits 0.5 seconds and then 1.0 second before the next attempt.

Retry guide

Basic policy

retry = Retry(times=4, delay=0.25, backoff="linear")
result = await retry(operation, arg1, keyword="value")

The callable receives all positional and keyword arguments supplied to retry(...). It must return an awaitable, normally by being declared with async def.

Backoff strategies

backoff	Delays for delay=1 after failed attempts
"constant"	1, 1, 1, ...
"linear"	1, 2, 3, ...
"exponential" (default)	1, 2, 4, ...

Use max_delay to prevent an exponential policy from growing without bound:

retry = Retry(
    times=5,
    delay=1.0,
    backoff="exponential",
    max_delay=10.0,
)

Jitter

When many workers retry the same dependency, identical delays can cause a retry spike. Enable jitter to distribute retries over time:

# Full jitter: a random value from 0 to the calculated delay.
retry = Retry(times=3, delay=1.0, jitter=True)

# Proportional jitter: vary the calculated delay by at most +/-20%.
retry = Retry(times=3, delay=1.0, jitter=0.2)

Retry only expected failures

The default retries subclasses of Exception. Narrow this for operations where only specific transient errors are safe to retry:

retry = Retry(
    times=3,
    delay=0.2,
    exceptions=(ConnectionError, TimeoutError),
)
result = await retry(call_remote_service)

The final retryable exception is raised unchanged. Exceptions outside the configured tuple are raised immediately.

Cancellation is never retried

asyncio.CancelledError always propagates immediately, even if a broad exception tuple is supplied. This keeps task cancellation, graceful shutdown, and timeout ownership under the caller's control.

API reference

Retry

Retry(
    *,
    times: int = 3,
    delay: float = 0.0,
    backoff: str = "exponential",
    max_delay: Optional[float] = None,
    jitter: Union[bool, float] = False,
    exceptions: Tuple[Type[BaseException], ...] = (Exception,),
)

Parameter	Meaning
times	Maximum total attempts; must be at least 1.
delay	Initial delay in seconds; must be non-negative.
backoff	"constant", "linear", or "exponential".
max_delay	Optional upper bound for a calculated delay.
jitter	True for full jitter, or a float from 0.0 to 1.0 for proportional jitter.
exceptions	Non-empty tuple of exception classes eligible for retries.

For deterministic tests or custom event-loop integration, advanced callers may also inject sleep and random_fn. See the full docstring in pyasynckit.retry.

Development

Run the test suite

python -m pip install -e ".[dev]"
python -m pytest --cov

The project requires 100% line and branch coverage. New public behaviour must include normal, failure, cancellation, and boundary-case tests where relevant.

Project layout

src/pyasynckit/   # Library source code
tests/            # pytest and pytest-asyncio test suite
docs/             # Sphinx documentation source
.github/workflows/# CI and release automation

Contributing

1. Create a focused branch from main.
2. Add type hints and user-facing docstrings for public APIs.
3. Add tests that keep the coverage gate at 100%.
4. Run the test command above before opening a pull request.

Please keep runtime code limited to the standard library. Development-only tools belong in the dev optional dependency group in pyproject.toml.

Releases

Releases are created by pushing a version tag such as v0.1.0. The GitHub workflow verifies that the tag matches [project].version, builds both an sdist and wheel, and publishes them to PyPI using the repository's PYPI_API_TOKEN. See the release workflow for the exact process.