bizone-cloud-helpers 0.1.3

Shared helpers library for Bizone projects.

bizone-cloud-helpers

Reusable helpers and decorators shared across Bizone Python projects. Designed for FastAPI applications running in cloud environments (AWS, Azure, GCP).

Features

• Asynchronous Task Processing: @large_task decorator to turn heavy endpoints into background tasks with callbacks and cross-process idempotency.
• Concurrency Control: @limit_concurrency decorator for sync functions to prevent resource exhaustion and manage execution timeouts.
• Cloud File Management: Unified interface for S3, Azure Blob Storage, Google Cloud Storage, and local files with built-in caching (FileCacheManager) and uploading (RemoteUploader).
• Idempotency: Cross-process file-based locking to ensure tasks with the same X-call-id aren't executed multiple times concurrently.

Installation

From PyPI

pip install bizone-cloud-helpers==0.1.3

The source repository can remain private. Client projects should consume published release artifacts from PyPI instead of installing from Git.

Usage

1. Asynchronous Tasks with Callbacks

Use @large_task to handle long-running requests. It automatically responds with 202 Accepted and runs the function in the background, POSTing the result back to a callback URL.

from bizone_cloud_helpers.async_task import large_task

@app.post("/process")
@large_task(callback_field="callback_url")
async def heavy_computation(data: dict):
    # This runs in background if callback_url is provided in request
    result = perform_work(data)
    return {"status": "done", "result": result}

Support for:

• Idempotency via X-call-id header.
• Automatic result/error reporting via HTTP POST.
• Context propagation for FastAPI Request.
• Heartbeat timer: Sends PUT pings to the caller during long tasks. Can be overridden via X-heartbeat-interval header (minimum 1s).
• Concurrency control: Limit how many background instances of a task run at once.

2. Concurrency Limiting

Limit how many instances of a sync or async function can run at once.

from bizone_cloud_helpers.concurrency import limit_concurrency

@limit_concurrency(group="cpu_intensive", max_concurrency=2, max_runtime=60)
def compute_heavy_stuff():
    # Only 2 of these will run at a time across the process
    ...

@limit_concurrency(group="io_intensive", max_concurrency=5)
async def async_io_task():
    # Also works for async functions
    await do_something()

3. Remote Files and Caching

Unified access to cloud storage with local caching.

from bizone_cloud_helpers.remote_files import RemoteFileInfo

# Download/Cache a file from S3
info = RemoteFileInfo(
    provider="s3",
    bucket="my-bucket",
    key="path/to/file.txt",
    access_key="...",
    secret_key="..."
)

local_path = info.local_file # Downloads if not cached or expired

Configuration

Environment variables:

• IDEMPOTENCY_LOCK_DIR: Directory for idempotency locks (default: /tmp/myapp_idemp_locks).
• BIZONE_CALLBACK: Default base URL for flow callbacks.
• MAX_CONCURRENCY: Global default concurrency limit.
• MAX_CONCURRENCY_<GROUP>: Group-specific concurrency limit.
• FILE_CACHE_DIR: Directory for local file cache.

Development

Running Tests

The project uses pytest for testing.

1. Install development dependencies:

   pip install -e ".[dev]"
   

2. Run all tests:

   python -m pytest
   

3. Run specific test file:

   python -m pytest tests/test_async_task_new.py
   

Building a Release Locally

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

The build creates a wheel and source distribution under dist/.

Publishing Releases

Releases are published to public PyPI from GitHub Actions using PyPI Trusted Publishing. No PyPI API token is required in GitHub secrets.

One-time PyPI setup:

1. Create or claim the bizone-cloud-helpers project on PyPI.
2. In PyPI project settings, add a Trusted Publisher for GitHub Actions:
   • owner: Bizone-ai
   • repository: bizone-cloud-helpers
   • workflow: release.yml
   • environment: pypi

Release steps:

git checkout main
git pull
python -m pytest
git tag v0.1.3
git push origin v0.1.3

After PyPI publishes the release, client projects can depend on:

bizone-cloud-helpers==0.1.3

License

MIT