Async Support¶
Overview¶
Two SDK modules have async variants: FileAPI and HarvesterAPI. All other modules (BrowsingAPI, UserManagementAPI, OrgManagementAPI, LogExtractionAPI) are synchronous.
Async support uses aiohttp — a required dependency, so no extra install is needed.
When async isn't available
For modules that don't have async variants (like BrowsingAPI), you can wrap sync calls with asyncio.to_thread() to run them concurrently in thread pools. See Mixing Sync and Async below.
When to Use Async¶
| Scenario | Recommended |
|---|---|
| Single file operation, simple script | Sync (upload_file, download_file) |
| Multiple files — batch processing | Async (bulk_upload_async, bulk_download_async) |
| Large files (> 50 MB) | Async chunked (upload_file_chunked_async) |
| Unreliable network | Async with retry (download_file_with_retry_async) |
| Long-running harvest tasks — polling | Async (wait_for_completion_async) |
FileAPI Async Methods¶
import asyncio
from silo_sdk import FileAPI
files = FileAPI(config)
# Upload multiple files concurrently
async def upload_batch():
uploads = [
{"bucket_id": "my-bucket", "file_path": "report1.pdf", "name": "report1.pdf"},
{"bucket_id": "my-bucket", "file_path": "report2.pdf", "name": "report2.pdf"},
{"bucket_id": "my-bucket", "file_path": "report3.pdf", "name": "report3.pdf"},
]
results = await files.bulk_upload_async(uploads, max_concurrent=3)
for r in results:
print(f"Uploaded: {r.get('file_id')}")
asyncio.run(upload_batch())
# Download with automatic retry on failure
async def safe_download():
result = await files.download_file_with_retry_async(
file_id="abc123",
output_path="./downloads/important.pdf",
max_retries=5,
retry_delay=2.0,
)
return result
asyncio.run(safe_download())
HarvesterAPI Async Methods¶
import asyncio
from silo_sdk import HarvesterAPI
harvester = HarvesterAPI(config)
# Create task and wait for completion (non-blocking poll)
async def screenshot_and_wait():
task_id = harvester.create_harvest_task(
task_type="visual", # valid types: "visual", "asset", "video", "single"
urls=["https://example.com"],
egress_info={"name": "New York, NY"},
dest_path="/screenshots/",
dest_name="example",
dest_bucket_id="my-bucket",
)
result = await harvester.wait_for_completion_async(
task_id=task_id,
check_interval=10, # check every 10 seconds
max_wait_time=120, # give up after 2 minutes
)
print(f"Task {task_id}: {result.get('status')}")
asyncio.run(screenshot_and_wait())
Mixing Sync and Async¶
You can call sync SDK methods from an async function — they run in the calling thread and don't block the event loop for short operations. For I/O-bound sync methods in a high-throughput async app, use asyncio.to_thread():
import asyncio
from silo_sdk import BrowsingAPI, UserManagementAPI
async def main():
browsing = BrowsingAPI(config)
users = UserManagementAPI(config)
# Run two sync API calls concurrently using threads
context_id, user_list = await asyncio.gather(
asyncio.to_thread(
browsing.create_context,
url="https://example.com",
username="user@org.com",
),
asyncio.to_thread(users.list_users, "my_org"),
)
asyncio.run(main())
Testing Async Code¶
The pytest configuration uses asyncio_mode = "auto" — no @pytest.mark.asyncio needed: