ice9 0.0.4

Python SDK for the ice9 image analysis API

ice9 SDK

Python SDK for the ice9 image analysis API.

Installation

pip install ice9

Quickstart

Synchronous (scripts, notebooks):

from ice9 import Ice9

client = Ice9(api_key="ice9_...")
result = client.analyze("photo.jpg")

print(result.nudenet)   # content moderation
print(result.colors)    # dominant colors
print(result.metadata)  # EXIF and file info

Asynchronous (FastAPI, Discord bots, async web apps):

from ice9 import AsyncIce9

async with AsyncIce9(api_key="ice9_...") as client:
    result = await client.analyze("photo.jpg")
    print(result.nudenet)

Both clients have identical APIs - just add async/await for the async version.

The SDK accepts images from multiple sources:

# Local file path
result = client.analyze("photo.jpg")

# URL (SDK downloads it for you)
result = client.analyze("https://example.com/photo.jpg")

# File object
with open("photo.jpg", "rb") as f:
    result = client.analyze(f)

Note: URLs are downloaded by the SDK (up to 10MB) and then submitted to the API. The API itself does not store images.

Authentication

Pass your API key directly or set the ICE9_API_KEY environment variable:

export ICE9_API_KEY=ice9_...

client = Ice9()  # picks up ICE9_API_KEY automatically

Async/Await

AsyncIce9 is a fully async client perfect for web servers and bots. It doesn't block your event loop while waiting for analysis results, so one process can handle many concurrent requests efficiently.

When to use async:

• ✅ Web servers (FastAPI, Flask with async routes, aiohttp)
• ✅ Discord/Telegram bots
• ✅ Any async application processing multiple images concurrently
• ❌ Simple scripts or Jupyter notebooks (use sync Ice9 instead)

All methods have async equivalents: await client.analyze(), await client.get_result(), await client.tiers(), await client.services().

Async streaming for real-time progress updates:

async for result in await client.analyze("photo.jpg", stream=True):
    if result.is_complete:
        print("Done!")
    else:
        print(f"Service {result.services_submitted[-1]} ready")

Both Ice9 (sync) and AsyncIce9 use the same underlying HTTP library (httpx), so there's no extra installation needed for async support.

Tiers and Services

The API processes images at different tiers. Each tier runs a different set of services. To see what's available:

# See which services are in each tier
tiers = client.tiers()
# {
#   "free":    ["colors", "metadata", "nudenet"],
#   "basic":   ["blip", "colors", "florence2", ...],
#   "premium": [...],
# }

# Or get a list of all available services
services = client.services()
# ["blip2", "colors", "florence2", "metadata", "nudenet", "yolo_v8", ...]

Pass a tier to analyze():

result = client.analyze("photo.jpg", tier="basic")

If you omit tier, the server uses the default for your key.

Results

Services that ran are accessible as attributes on the result:

result.nudenet.detections
result.colors.dominant
result.yolo_v8.boxes

Accessing a service that didn't run returns None.

Serializing results

For passing results to other systems (web frontends, databases, etc.), use to_dict():

result.to_dict()
# {
#   "image_id": 123,
#   "services_submitted": ["colors", "nudenet"],
#   "services_failed": {},
#   "services": {
#     "colors": {
#       "dominant": ["#FF0000", "#00FF00"],
#       "palette": [...]
#     },
#     "nudenet": {
#       "detections": [...]
#     }
#   }
# }

The SDK cleans up the structure:

• Services are nested under "services" (prevents field name collisions with top-level metadata)
• Redundant "service" and "status" fields are stripped
• Service data is unwrapped (no extra "data" nesting)

Note: The raw API response is available at result._raw, but you shouldn't need it. If you find yourself reaching for ._raw, that's a sign the SDK isn't surfacing something it should—please open an issue.

Retrieving past results

Results are stored and can be retrieved later using the image ID:

# Initial analysis
result = client.analyze("photo.jpg")
image_id = result.image_id  # Save this

# Later, retrieve the same results
result = client.get_result(image_id)

This is useful for:

• Retrieving results across different sessions
• Avoiding re-analysis of the same image
• Building dashboards or reports from historical data

Real-time Progress Updates

For real-time UIs, dashboards, or monitoring tools that need to show analysis progress as it happens, use streaming (recommended) or manual polling (fallback).

Streaming (Recommended)

The SDK handles SSE (Server-Sent Events) connections internally and yields partial results as services complete:

# Synchronous
for result in client.analyze("photo.jpg", stream=True):
    # result.is_complete is False until final result
    completed = len([s for s in result.services_submitted if getattr(result, s)])
    print(f"Progress: {completed}/{len(result.services_submitted)} services")

    if result.is_complete:
        print("Analysis complete!")
        break

# Asynchronous
async for result in await client.analyze("photo.jpg", stream=True):
    update_dashboard(result)  # Update UI with partial results
    if result.is_complete:
        break

Why streaming? It's efficient, real-time, and doesn't require you to write polling loops. The SDK manages the connection and yields results as they arrive.

Deployment note: For web servers proxying SSE to browsers, use async workers (e.g., gunicorn --worker-class gevent) to handle many concurrent connections without blocking.

Manual Polling (Fallback)

If streaming isn't available in your environment, use get_status() for manual polling:

import time

# Submit analysis
result = client.analyze("photo.jpg")
image_id = result.image_id

# Or if you already have an image_id:
while True:
    status = client.get_status(image_id)
    print(f"Progress: {status.get('services_completed', {})}")

    if status['is_complete']:
        # Status dict has all fields from /status endpoint
        final_result = status
        break

    time.sleep(0.5)  # Poll every 500ms

When to use polling: Only when SSE isn't available (e.g., restrictive firewalls, legacy infrastructure). Streaming is preferred for most use cases.

Error handling

from ice9 import (
    Ice9Error,           # base — catch this for any SDK error
    AuthError,           # invalid or deactivated key
    ImageRejectedError,  # bad format, too large, empty
    RateLimitError,      # check .retry_after for backoff hint
    AnalysisTimeoutError,# didn't complete within timeout
    PartialResultError,  # completed, but some services failed
)

try:
    result = client.analyze("photo.jpg")
except PartialResultError as e:
    # Some services failed — the partial result is still accessible
    result = e.result
    print("Failed services:", result.services_failed)
except AnalysisTimeoutError:
    print("Timed out waiting for results")
except Ice9Error as e:
    print("API error:", e)

PartialResultError carries the partial result on .result so you can decide whether what succeeded is enough for your use case.

Handling partial results without exceptions

If partial results are acceptable for your use case (e.g., you only need nudenet and other services are optional), use raise_on_partial=False:

# Returns result with services_failed populated, logs a warning instead of raising
result = client.analyze("photo.jpg", raise_on_partial=False)

if result.services_failed:
    print(f"Warning: some services failed: {result.services_failed}")

# Use the partial result
if result.nudenet:
    print("Nudenet succeeded:", result.nudenet.detections)

This is cleaner than catching PartialResultError when you know partial results are acceptable.

Timeout and retries

The default timeout is 30 seconds, which is sufficient for most tiers (free: <1s, basic: ~8s, premium: ~13s). The SDK automatically retries transient errors (rate limits, 5xx, connection errors) up to 3 times with exponential backoff.

# Configure timeout and retries
client = Ice9(
    api_key="ice9_...",
    timeout=45,         # seconds to wait for analysis (increase for batch tier)
    max_retries=3,      # number of retries (default: 3, set to 0 to disable)
)

# Per-call timeout override
result = client.analyze("photo.jpg", timeout=60)

What gets retried:

• ✅ Rate limits (429) - respects Retry-After header
• ✅ Server errors (5xx) - exponential backoff with jitter
• ✅ Connection errors - transient network issues
• ❌ Auth errors (401/403) - won't fix themselves
• ❌ Client errors (400/404) - won't fix themselves
• ❌ Initial /analyze submission - never retried (costs money + bandwidth)

The SDK retries reads (tiers(), get_result(), status polling) but never retries the initial image submission to protect against accidental charges.

Batch processing

For high-volume batch workloads, use the batch tier:

from concurrent.futures import ThreadPoolExecutor

def analyze_image(path):
    return client.analyze(path, tier="batch")

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(analyze_image, image_paths))

The batch tier is designed for parallel processing and uses LLM consensus (GPT, Gemini, Claude) for maximum accuracy. See examples/batch_tier.py for a complete example.

Key differences:

• Batch tier: Parallelization supported, LLM consensus, no NSFW support
• Basic/Premium tiers: Real-time optimized, single-image analysis, NSFW support

For pricing details, see ice9.ai/pricing.

Logging

The SDK uses Python's standard logging module to log HTTP requests and responses. This can be helpful for debugging or monitoring API usage.

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("ice9").setLevel(logging.DEBUG)

# Now all SDK operations will log
client = Ice9()
result = client.analyze("photo.jpg")

Log levels:

• DEBUG: Every API call (GET /tiers, POST /analyze, polling GET /status)
• INFO: Successful submissions and completions
• WARNING: Retry attempts with backoff delays

Example output:

DEBUG:ice9:POST /analyze (tier=default, file=photo.jpg)
INFO:ice9:POST /analyze -> 202 Accepted (image_id=12345)
DEBUG:ice9:GET /status/12345 -> in progress (2/5 services)
DEBUG:ice9:GET /status/12345 -> in progress (4/5 services)
INFO:ice9:GET /status/12345 -> complete

When to use:

• Debugging failed requests or unexpected behavior
• Monitoring rate limit retries
• Tracking which services are taking longest
• Understanding retry/backoff patterns

By default, logging is off (only WARNING and above). Enable it explicitly when needed.

Running the tests

Unit tests (no credentials required):

pip install ice9[dev]
pytest

Integration tests (requires ICE9_API_KEY):

export ICE9_API_KEY=ice9_...
pytest tests/integration/ -v

Support

Issues and feature requests: GitHub Issues

Documentation: ice9.ai/docs

Questions or feedback: Visit ice9.ai for contact information.

We're actively maintaining this SDK and respond to issues promptly. If something isn't working as expected or you'd like to see a feature added, please let us know.