# aiopythonik
Asynchronous wrapper for the
[pythonik](https://pypi.org/project/nsa-pythonik/) library, enabling its
use in async Python applications without blocking the event loop.
[](https://pypi.org/project/aiopythonik/)
[](https://pypi.org/project/aiopythonik/)
[](https://pypi.org/project/aiopythonik/)
## Overview
`aiopythonik` provides asynchronous versions of pythonik functionality
by wrapping the synchronous operations in a thread pool executor. This
approach is similar to how `aioboto3` wraps `boto3`, allowing you to use
asynchronous syntax while maintaining the original library's
capabilities.
### Features
- Complete async API for the pythonik library
- Automatic thread pool management for non-blocking operations
- Built-in rate limit handling with proactive throttling and
configurable retry strategies
- Extended functionality through patched pythonik methods
- Support for Python 3.11+
## Installation
### Requirements
- Python 3.11 or higher
```bash
# Install from PyPI (recommended for most users)
pip install aiopythonik
```
The required dependency `nsa-pythonik` will be automatically installed.
### Installing from Source
For development or to get the latest unreleased changes:
```bash
# Clone the repository
git clone https://bitbucket.org/chesa/aiopythonik.git
cd aiopythonik
# Install in development mode
pip install -e .
# Install with development dependencies
pip install -e ".[dev]"
```
## Quickstart
```python
import asyncio
from aiopythonik import AsyncPythonikClient
async def main():
# Initialize the client
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
timeout=60,
base_url="https://app.iconik.io",
)
try:
# Use async methods
asset = await client.assets().get("asset_id")
print(f"Asset title: {asset.data.title}")
# Get files for the asset
files = await client.files().get_asset_files("asset_id")
print(f"Number of files: {len(files.data.files)}")
# Search for assets
from pythonik.models.search.search_body import SearchBody
search_results = await client.search().search(
SearchBody(doc_types=["assets"], query="title:sample")
)
print(f"Found {len(search_results.data.objects)} assets")
finally:
# Always close the client when done
await client.close()
if __name__ == "__main__":
asyncio.run(main())
```
## Using the Context Manager
For convenience, you can use the async context manager to ensure proper
cleanup:
```python
import asyncio
from aiopythonik import AsyncPythonikClientContext
async def main():
async with AsyncPythonikClientContext(
app_id="your_app_id",
auth_token="your_auth_token",
timeout=60,
base_url="https://app.iconik.io",
) as client:
# Use async methods
asset = await client.assets().get("asset_id")
print(f"Asset title: {asset.data.title}")
if __name__ == "__main__":
asyncio.run(main())
```
## API Coverage
`aiopythonik` provides async wrappers for all pythonik APIs and extends
functionality with some additional methods. Each API from the original
library is accessible through the corresponding async wrapper:
```python
# Assets
asset = await client.assets().get("asset_id")
assets = await client.assets().fetch(params={"per_page": 50}) # Enhanced method
await client.assets().delete("asset_id")
# Collections
collection = await client.collections().get("collection_id")
info = await client.collections().get_info("collection_id")
contents = await client.collections().get_contents("collection_id")
# Files
files = await client.files().get_asset_files("asset_id")
# Enhanced method with automatic checksum calculation
files_by_checksum = await client.files().get_files_by_checksum("d41d8cd98f00b204e9800998ecf8427e")
# Or calculate checksum automatically from a file
files_by_file = await client.files().get_files_by_checksum("path/to/file.mp4")
# Metadata
views = await client.metadata().get_views()
view = await client.metadata().get_view("view_id")
metadata = await client.metadata().get_asset_metadata("asset_id", "view_id")
# Jobs
job = await client.jobs().get("job_id")
await client.jobs().cancel("job_id")
```
### Automatic Rate Limit Handling
The library includes built-in handling for API rate limits with both
**proactive throttling** and **reactive retry logic**:
```python
from aiopythonik import AsyncPythonikClient, RateLimitConfig
# Configure custom rate limiting behavior
rate_limit_config = RateLimitConfig(
max_retries=5, # Maximum number of retries for rate-limited requests
initial_backoff=1.0, # Initial backoff in seconds
max_backoff=30.0, # Maximum backoff in seconds
backoff_factor=2.0, # Exponential backoff factor
jitter=True, # Add randomness to backoff times
enable_proactive_throttling=True, # Enable proactive throttling (default: True)
proactive_throttling_threshold=0.8, # Start throttling at 80% quota usage
max_proactive_delay=5.0 # Maximum proactive delay in seconds
)
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
rate_limit_config=rate_limit_config
)
# Rate-limited requests will automatically be handled with:
# 1. Proactive throttling - slows down requests before hitting limits
# 2. Retry logic - handles 429 errors with exponential backoff
```
#### Proactive Throttling
The library automatically monitors your API quota usage through
`RateLimit-Remaining` headers and applies graduated delays **before**
hitting rate limits:
- **80%+ quota remaining**: No delay (full speed)
- **30-20% quota remaining**: Light throttling (0.1-0.5s delays)
- **20-10% quota remaining**: Moderate throttling (0.5-2.0s delays)
- **<10% quota remaining**: Aggressive throttling (2.0-5.0s delays)
This prevents 429 errors instead of just reacting to them, resulting in:
- Faster overall operations (no waiting for 5+ second retry delays)
- More predictable request timing
- Better resource efficiency
- Reduced server load
```python
# Proactive throttling can be disabled if needed
rate_limit_config = RateLimitConfig(
enable_proactive_throttling=False # Disable proactive throttling
)
```
## Advanced Usage
### Concurrent Operations
Running multiple operations concurrently:
```python
import asyncio
from aiopythonik import AsyncPythonikClientContext
async def main():
async with AsyncPythonikClientContext(
app_id="your_app_id",
auth_token="your_auth_token",
) as client:
# Run multiple operations concurrently
asset_ids = ["id1", "id2", "id3", "id4", "id5"]
tasks = [client.assets().get(asset_id) for asset_id in asset_ids]
results = await asyncio.gather(*tasks)
for i, result in enumerate(results):
print(f"Asset {i+1}: {result.data.title}")
if __name__ == "__main__":
asyncio.run(main())
```
### Custom Base URL
If you need to use a different API endpoint:
```python
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
base_url="https://custom.iconik.io"
)
```
### Customizing Thread Pool Size
Control the maximum number of worker threads:
```python
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
max_workers=10 # Set maximum number of worker threads
)
```
## Rate Limiting Details
The iconik APIs implement rate limiting to prevent individual users from
negatively impacting system performance. By default, the `aiopythonik`
library includes automatic handling of rate limits using a retry
strategy with exponential backoff.
Rate limits are enforced per authenticated user and application token:
- 50 requests per second sustained
- 1000 requests over any 20 second period
The library uses a **hybrid approach** to handle these limits:
1. **Proactive Throttling**: Monitors `RateLimit-Remaining` headers and
gradually slows down requests as you approach the limit, preventing
429 errors from occurring
2. **Reactive Retry Logic**: If 429 errors still occur, automatically
retries with exponential backoff
This combination provides optimal performance - prevention when
possible, recovery when necessary.
You can disable automatic rate limit handling if you prefer to manage it
yourself:
```python
# Disable all rate limit handling
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
disable_rate_limit_handling=True
)
# Or disable only proactive throttling while keeping retry logic
rate_limit_config = RateLimitConfig(
enable_proactive_throttling=False, # Disable proactive throttling
max_retries=3 # Keep retry logic
)
client = AsyncPythonikClient(
app_id="your_app_id",
auth_token="your_auth_token",
rate_limit_config=rate_limit_config
)
```
## License
MIT
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
Raw data
{
"_id": null,
"home_page": null,
"name": "aiopythonik",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": "Product Team <product@chesa.com>, \"Chesapeake Systems, LLC\" <info@chesa.com>",
"keywords": "iconik, api, pythonik, nsa-pythonik, asynchronous",
"author": null,
"author_email": "\"Brian F. Summa\" <brian.f.summa@chesa.com>",
"download_url": "https://files.pythonhosted.org/packages/1d/2f/a898f8514b3d6910d3dabbfc6f78aec9f2ec0df2312c7a6a1fbe0e3103cd/aiopythonik-2025.7b1.tar.gz",
"platform": null,
"description": "# aiopythonik\n\nAsynchronous wrapper for the\n[pythonik](https://pypi.org/project/nsa-pythonik/) library, enabling its\nuse in async Python applications without blocking the event loop.\n\n[](https://pypi.org/project/aiopythonik/)\n[](https://pypi.org/project/aiopythonik/)\n[](https://pypi.org/project/aiopythonik/)\n\n## Overview\n\n`aiopythonik` provides asynchronous versions of pythonik functionality\nby wrapping the synchronous operations in a thread pool executor. This\napproach is similar to how `aioboto3` wraps `boto3`, allowing you to use\nasynchronous syntax while maintaining the original library's\ncapabilities.\n\n### Features\n\n- Complete async API for the pythonik library\n- Automatic thread pool management for non-blocking operations\n- Built-in rate limit handling with proactive throttling and\n configurable retry strategies\n- Extended functionality through patched pythonik methods\n- Support for Python 3.11+\n\n## Installation\n\n### Requirements\n\n- Python 3.11 or higher\n\n```bash\n# Install from PyPI (recommended for most users)\npip install aiopythonik\n```\n\nThe required dependency `nsa-pythonik` will be automatically installed.\n\n### Installing from Source\n\nFor development or to get the latest unreleased changes:\n\n```bash\n# Clone the repository\ngit clone https://bitbucket.org/chesa/aiopythonik.git\ncd aiopythonik\n\n# Install in development mode\npip install -e .\n\n# Install with development dependencies\npip install -e \".[dev]\"\n```\n\n## Quickstart\n\n```python\nimport asyncio\nfrom aiopythonik import AsyncPythonikClient\n\nasync def main():\n # Initialize the client\n client = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n timeout=60,\n base_url=\"https://app.iconik.io\",\n )\n\n try:\n # Use async methods\n asset = await client.assets().get(\"asset_id\")\n print(f\"Asset title: {asset.data.title}\")\n\n # Get files for the asset\n files = await client.files().get_asset_files(\"asset_id\")\n print(f\"Number of files: {len(files.data.files)}\")\n\n # Search for assets\n from pythonik.models.search.search_body import SearchBody\n search_results = await client.search().search(\n SearchBody(doc_types=[\"assets\"], query=\"title:sample\")\n )\n print(f\"Found {len(search_results.data.objects)} assets\")\n\n finally:\n # Always close the client when done\n await client.close()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## Using the Context Manager\n\nFor convenience, you can use the async context manager to ensure proper\ncleanup:\n\n```python\nimport asyncio\nfrom aiopythonik import AsyncPythonikClientContext\n\nasync def main():\n async with AsyncPythonikClientContext(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n timeout=60,\n base_url=\"https://app.iconik.io\",\n ) as client:\n # Use async methods\n asset = await client.assets().get(\"asset_id\")\n print(f\"Asset title: {asset.data.title}\")\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## API Coverage\n\n`aiopythonik` provides async wrappers for all pythonik APIs and extends\nfunctionality with some additional methods. Each API from the original\nlibrary is accessible through the corresponding async wrapper:\n\n```python\n# Assets\nasset = await client.assets().get(\"asset_id\")\nassets = await client.assets().fetch(params={\"per_page\": 50}) # Enhanced method\nawait client.assets().delete(\"asset_id\")\n\n# Collections\ncollection = await client.collections().get(\"collection_id\")\ninfo = await client.collections().get_info(\"collection_id\")\ncontents = await client.collections().get_contents(\"collection_id\")\n\n# Files\nfiles = await client.files().get_asset_files(\"asset_id\")\n# Enhanced method with automatic checksum calculation\nfiles_by_checksum = await client.files().get_files_by_checksum(\"d41d8cd98f00b204e9800998ecf8427e\")\n# Or calculate checksum automatically from a file\nfiles_by_file = await client.files().get_files_by_checksum(\"path/to/file.mp4\")\n\n# Metadata\nviews = await client.metadata().get_views()\nview = await client.metadata().get_view(\"view_id\")\nmetadata = await client.metadata().get_asset_metadata(\"asset_id\", \"view_id\")\n\n# Jobs\njob = await client.jobs().get(\"job_id\")\nawait client.jobs().cancel(\"job_id\")\n```\n\n### Automatic Rate Limit Handling\n\nThe library includes built-in handling for API rate limits with both\n**proactive throttling** and **reactive retry logic**:\n\n```python\nfrom aiopythonik import AsyncPythonikClient, RateLimitConfig\n\n# Configure custom rate limiting behavior\nrate_limit_config = RateLimitConfig(\n max_retries=5, # Maximum number of retries for rate-limited requests\n initial_backoff=1.0, # Initial backoff in seconds\n max_backoff=30.0, # Maximum backoff in seconds\n backoff_factor=2.0, # Exponential backoff factor\n jitter=True, # Add randomness to backoff times\n enable_proactive_throttling=True, # Enable proactive throttling (default: True)\n proactive_throttling_threshold=0.8, # Start throttling at 80% quota usage\n max_proactive_delay=5.0 # Maximum proactive delay in seconds\n)\n\nclient = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n rate_limit_config=rate_limit_config\n)\n\n# Rate-limited requests will automatically be handled with:\n# 1. Proactive throttling - slows down requests before hitting limits\n# 2. Retry logic - handles 429 errors with exponential backoff\n```\n\n#### Proactive Throttling\n\nThe library automatically monitors your API quota usage through\n`RateLimit-Remaining` headers and applies graduated delays **before**\nhitting rate limits:\n\n- **80%+ quota remaining**: No delay (full speed)\n- **30-20% quota remaining**: Light throttling (0.1-0.5s delays)\n- **20-10% quota remaining**: Moderate throttling (0.5-2.0s delays)\n- **<10% quota remaining**: Aggressive throttling (2.0-5.0s delays)\n\nThis prevents 429 errors instead of just reacting to them, resulting in:\n\n- Faster overall operations (no waiting for 5+ second retry delays)\n- More predictable request timing\n- Better resource efficiency\n- Reduced server load\n\n```python\n# Proactive throttling can be disabled if needed\nrate_limit_config = RateLimitConfig(\n enable_proactive_throttling=False # Disable proactive throttling\n)\n```\n\n## Advanced Usage\n\n### Concurrent Operations\n\nRunning multiple operations concurrently:\n\n```python\nimport asyncio\nfrom aiopythonik import AsyncPythonikClientContext\n\nasync def main():\n async with AsyncPythonikClientContext(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n ) as client:\n # Run multiple operations concurrently\n asset_ids = [\"id1\", \"id2\", \"id3\", \"id4\", \"id5\"]\n\n tasks = [client.assets().get(asset_id) for asset_id in asset_ids]\n results = await asyncio.gather(*tasks)\n\n for i, result in enumerate(results):\n print(f\"Asset {i+1}: {result.data.title}\")\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### Custom Base URL\n\nIf you need to use a different API endpoint:\n\n```python\nclient = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n base_url=\"https://custom.iconik.io\"\n)\n```\n\n### Customizing Thread Pool Size\n\nControl the maximum number of worker threads:\n\n```python\nclient = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n max_workers=10 # Set maximum number of worker threads\n)\n```\n\n## Rate Limiting Details\n\nThe iconik APIs implement rate limiting to prevent individual users from\nnegatively impacting system performance. By default, the `aiopythonik`\nlibrary includes automatic handling of rate limits using a retry\nstrategy with exponential backoff.\n\nRate limits are enforced per authenticated user and application token:\n\n- 50 requests per second sustained\n- 1000 requests over any 20 second period\n\nThe library uses a **hybrid approach** to handle these limits:\n\n1. **Proactive Throttling**: Monitors `RateLimit-Remaining` headers and\n gradually slows down requests as you approach the limit, preventing\n 429 errors from occurring\n2. **Reactive Retry Logic**: If 429 errors still occur, automatically\n retries with exponential backoff\n\nThis combination provides optimal performance - prevention when\npossible, recovery when necessary.\n\nYou can disable automatic rate limit handling if you prefer to manage it\nyourself:\n\n```python\n# Disable all rate limit handling\nclient = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n disable_rate_limit_handling=True\n)\n\n# Or disable only proactive throttling while keeping retry logic\nrate_limit_config = RateLimitConfig(\n enable_proactive_throttling=False, # Disable proactive throttling\n max_retries=3 # Keep retry logic\n)\nclient = AsyncPythonikClient(\n app_id=\"your_app_id\",\n auth_token=\"your_auth_token\",\n rate_limit_config=rate_limit_config\n)\n```\n\n## License\n\nMIT\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Asynchronous wrapper for NSA's pythonik client library",
"version": "2025.7b1",
"project_urls": {
"Homepage": "https://chesa.com/",
"Repository": "https://bitbucket.org/chesa/aiopythonik/"
},
"split_keywords": [
"iconik",
" api",
" pythonik",
" nsa-pythonik",
" asynchronous"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "002cd3bc298d7afaee990909b4807a0f0eaad08680d1cb71a376127e2baf2e49",
"md5": "4096fa38081a29c192c65366117aede5",
"sha256": "42952caa3b124a950694161d4fece81391be44362bd7e7792634aafa54790572"
},
"downloads": -1,
"filename": "aiopythonik-2025.7b1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4096fa38081a29c192c65366117aede5",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 39874,
"upload_time": "2025-07-30T00:49:05",
"upload_time_iso_8601": "2025-07-30T00:49:05.966380Z",
"url": "https://files.pythonhosted.org/packages/00/2c/d3bc298d7afaee990909b4807a0f0eaad08680d1cb71a376127e2baf2e49/aiopythonik-2025.7b1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "1d2fa898f8514b3d6910d3dabbfc6f78aec9f2ec0df2312c7a6a1fbe0e3103cd",
"md5": "3c43163b834f06bb6022b8e79fcca68e",
"sha256": "15798dee48a88cea46e6fecaef523ae6cb54a892ee250ec01a201a62014328c1"
},
"downloads": -1,
"filename": "aiopythonik-2025.7b1.tar.gz",
"has_sig": false,
"md5_digest": "3c43163b834f06bb6022b8e79fcca68e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 35378,
"upload_time": "2025-07-30T00:49:07",
"upload_time_iso_8601": "2025-07-30T00:49:07.284029Z",
"url": "https://files.pythonhosted.org/packages/1d/2f/a898f8514b3d6910d3dabbfc6f78aec9f2ec0df2312c7a6a1fbe0e3103cd/aiopythonik-2025.7b1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-30 00:49:07",
"github": false,
"gitlab": false,
"bitbucket": true,
"codeberg": false,
"bitbucket_user": "chesa",
"bitbucket_project": "aiopythonik",
"lcname": "aiopythonik"
}
JSON