cf-ai-platform-client 0.1.0

Official Python SDK for CloudFactory AI Platform - download and manage project assets and results

CloudFactory AI Platform Python Client

Official Python SDK for the CloudFactory AI Platform. Download COCO format annotations from your projects with flexible filtering (batches, work items, timestamps), concurrent workers, automatic retries, and progress tracking.

Features

✨ Simple API - Clean, intuitive async interface with comprehensive type hints
🚀 Concurrent Downloads - Configurable async tasks for fast parallel downloads
🔄 Auto-Retry - Exponential backoff retry logic for failed downloads (3 attempts)
📊 Progress Tracking - Real-time progress bars with tqdm.asyncio
🔒 Type Safe - Built with Pydantic for robust data validation
🎯 Production Ready - Comprehensive error handling and logging
🔍 Flexible Filtering - Filter by batch IDs, work item IDs, or timestamp ranges
📁 Smart Organization - Downloads organized by batch names automatically

Installation

uv add cf-ai-platform-client

Or with pip:

pip install cf-ai-platform-client

Quick Start

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    # Initialize client with your API key
    async with CFUserClient.create_with_api_key("your-api-key-here") as client:
        # Download all COCO annotations from a project
        summary = await client.download_project(
            project_id="proj_123",
            space_id="space_456",  # Required for generating signed URLs
            destination="./downloads"
        )

        print(f"✅ Downloaded {summary.successful}/{summary.total_items} items")
        print(f"❌ Failed: {summary.failed}")

if __name__ == "__main__":
    asyncio.run(main())

Usage Examples

Basic Download

Download all COCO annotations from a project:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download entire project
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads"
        )

        # Check results
        if summary.failed > 0:
            print("Some downloads failed:")
            for result in summary.results:
                if not result.success:
                    print(f"  - Item {result.item_id}: {result.error}")

if __name__ == "__main__":
    asyncio.run(main())

Filter by Batch IDs

Download specific batches only:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download only specific batches
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            batch_ids=[1, 2, 3]  # List of batch IDs (max 100)
        )

if __name__ == "__main__":
    asyncio.run(main())

Filter by Work Item IDs

Download specific work items:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download specific work items by their IDs
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            work_item_ids=["550e8400-e29b-41d4-a716-446655440000", "6ba7b810-9dad-11d1-80b4-00c04fd430c8"]
        )

if __name__ == "__main__":
    asyncio.run(main())

Filter by Timestamp Ranges

Download items created or updated within specific time ranges:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download items created in January 2026
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            created_at_from="2026-01-01T00:00:00Z",
            created_at_to="2026-01-31T23:59:59Z"
        )

        # Download items updated in the last week
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            updated_at_from="2026-02-11T00:00:00Z",
            updated_at_to="2026-02-18T23:59:59Z"
        )

if __name__ == "__main__":
    asyncio.run(main())

Combine Multiple Filters

Use multiple filters together:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download specific batches created in a date range
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            batch_ids=[1, 2],
            created_at_from="2026-01-01T00:00:00Z",
            created_at_to="2026-01-31T23:59:59Z",
            max_workers=15
        )

if __name__ == "__main__":
    asyncio.run(main())

Configure Concurrency

Control the number of concurrent download workers:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Use more workers for faster downloads (default: 10)
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads",
            max_workers=20  # Use 20 concurrent tasks
        )

if __name__ == "__main__":
    asyncio.run(main())

Context Manager Usage

Automatically cleanup resources:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads"
        )
        print(f"Downloaded {summary.successful} items")

    # Client is automatically closed

if __name__ == "__main__":
    asyncio.run(main())

Inspect Download Results

Access detailed information about each downloaded item:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads"
        )

        # Iterate through results
        for result in summary.results:
            if result.success:
                print(f"✅ {result.item_id} -> {result.file_path}")
            else:
                print(f"❌ {result.item_id}: {result.error}")

if __name__ == "__main__":
    asyncio.run(main())

Custom Base URL

Use a custom API endpoint (e.g., for testing or staging):

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key(
        api_key="MY_API_KEY",
        base_url="https://api.staging.ai.cloudfactory.app/api/v1"
    ) as client:
        summary = await client.download_project(
            project_id="proj_abc123",
            space_id="space_456",
            destination="./my_downloads"
        )

if __name__ == "__main__":
    asyncio.run(main())

File Organization

Downloaded COCO annotation files are organized by batch name:

destination/
├── Training_Batch_001/
│   ├── item_abc123.json
│   ├── item_def456.json
│   └── item_ghi789.json
├── Validation_Batch/
│   ├── item_jkl012.json
│   └── item_mno345.json
└── unknown_batch/
    └── item_orphan.json

• Files are grouped by batch name (fetched from batch metadata)
• File naming: {work_item_id}.{extension}
• Extension is auto-detected from URL (defaults to .json)
• Items without a batch go to unknown_batch/
• Only COMPLETED work items are downloaded by default

API Reference

CFUserClient

Main client class for CloudFactory AI Platform.

Architecture

The download process follows these steps:

1. Fetch Batches: Retrieve batch metadata (ID → name mapping) from the Platform API
2. Fetch Manifest: Query work items with optional filters (status defaults to "COMPLETED")
3. Parallel Downloads: Use asyncio with semaphore to download annotations concurrently
4. For Each Item:
   • Fetch annotation metadata from Platform API
   • Extract COCO results URL from annotation metadata
   • Generate signed download URL from Space API
   • Download file content with retry logic
   • Save to disk organized by batch name
5. Return Summary: Aggregate statistics and per-item results

Methods

create_with_api_key(api_key: str, base_url: Optional[str] = None) -> CFUserClient

Create a client instance with API key authentication.

Parameters:

• api_key (str): Your CloudFactory API key
• base_url (Optional[str]): Custom API base URL (default: production)

Returns: Configured CFUserClient instance

---

download_project(project_id: str, space_id: str, destination: str, batch_ids: Optional[list[int]] = None, work_item_ids: Optional[list[str]] = None, created_at_from: Optional[str] = None, created_at_to: Optional[str] = None, updated_at_from: Optional[str] = None, updated_at_to: Optional[str] = None, max_workers: int = 10) -> DownloadSummary

Download all COCO annotations from a project with flexible filtering options. This is an async method that must be awaited.

Parameters:

• project_id (str): The project ID to download
• space_id (str): The space ID (required for generating signed URLs)
• destination (str): Local directory path to save files
• batch_ids (Optional[list[int]]): Filter by specific batch IDs (max 100). Downloads all if None
• work_item_ids (Optional[list[str]]): Filter by specific work item IDs (UUID format). Downloads all if None
• created_at_from (Optional[str]): Filter items created on or after this timestamp (ISO 8601 format, e.g., "2026-01-15T10:30:00Z")
• created_at_to (Optional[str]): Filter items created on or before this timestamp (ISO 8601 format)
• updated_at_from (Optional[str]): Filter items updated on or after this timestamp (ISO 8601 format)
• updated_at_to (Optional[str]): Filter items updated on or before this timestamp (ISO 8601 format)
• max_workers (int): Maximum concurrent async tasks (default: 10)

Returns: DownloadSummary with statistics and detailed results

Raises:

• httpx.HTTPStatusError: If API requests fail
• OSError: If file system operations fail

Example:

import asyncio
from cf_ai_platform_client import CFUserClient

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        # Download with multiple filters
        summary = await client.download_project(
            project_id="proj_123",
            space_id="space_123",
            destination="./downloads",
            batch_ids=[1, 2],
            created_at_from="2026-01-01T00:00:00Z",
            created_at_to="2026-01-31T23:59:59Z",
            max_workers=5
        )
        print(f"Downloaded {summary.successful}/{summary.total_items} items")

if __name__ == "__main__":
    asyncio.run(main())

---

close() -> None

Close HTTP clients and cleanup resources. Called automatically when using async context manager. This is an async method that must be awaited.

---

Models

DownloadSummary

Summary statistics for a download operation.

Attributes:

• total_items (int): Total number of items processed
• successful (int): Number of successful downloads
• failed (int): Number of failed downloads
• results (list[DownloadResult]): Detailed results for each item

DownloadResult

Result of a single work item download.

Attributes:

• item_id (str): ID of the work item
• batch_id (Optional[int]): Batch ID the item belongs to
• success (bool): Whether the download succeeded
• file_path (Optional[str]): Path where file was saved (if successful)
• error (Optional[str]): Error message (if failed)

WorkItem

Represents a work item from the Platform API.

Attributes:

• id (str): Unique identifier
• batch_id (Optional[int]): Batch identifier
• result_url (Optional[HttpUrl]): Result URL
• asset_url (Optional[str]): Asset URL

SearchResponse

Response from the work items search endpoint.

Attributes:

• items (list[WorkItem]): List of work items matching the search criteria
• total (int): Total count of items
• meta (Optional[dict]): Pagination metadata

Error Handling

The SDK includes comprehensive error handling:

Automatic Retries

Downloads automatically retry up to 3 times with exponential backoff (2-10 seconds) for:

• Network errors and timeouts (httpx.HTTPError, httpx.TimeoutException)
• All HTTP errors (including 4xx and 5xx status codes)

Retries apply only to the download phase (fetching files from signed URLs). API calls to fetch manifests, batches, and annotations do not have automatic retries.

Graceful Failures

If a download fails after retries:

• Error is logged
• Item is marked as failed in results
• Download continues for remaining items
• Summary includes failure details

Example Error Handling

import asyncio
import logging
from cf_ai_platform_client import CFUserClient

# Enable debug logging
logging.basicConfig(level=logging.INFO)

async def main():
    async with CFUserClient.create_with_api_key("MY_API_KEY") as client:
        try:
            summary = await client.download_project(
                project_id="proj_abc123",
                space_id="space_456",
                destination="./downloads"
            )
            
            if summary.failed > 0:
                print(f"\n⚠️  {summary.failed} items failed to download:")
                for result in summary.results:
                    if not result.success:
                        print(f"  - {result.item_id}: {result.error}")
                        
        except Exception as e:
            print(f"❌ Download failed: {e}")
            raise

if __name__ == "__main__":
    asyncio.run(main())

Requirements

• Python 3.10 to 3.14 (actively supported versions)
• Dependencies:
  • httpx >= 0.27.0 - HTTP client
  • pydantic >= 2.6.0 - Data validation
  • tqdm >= 4.66.0 - Progress bars
  • tenacity >= 8.2.0 - Retry logic

Development

Setup

This project uses uv for fast dependency management. The setup script installs uv with checksum verification for security:

# Clone the repository
git clone https://github.com/cloudfactory/cf-ai-platform-client-python.git
cd cf-ai-platform-client-python

# Run setup (installs uv with checksum verification)
./setup.sh --dev

Alternative uv installation methods:

If you prefer to install uv yourself:

# macOS (recommended)
brew install uv

# Using pipx
pipx install uv

# Using pip
pip install uv

# Then install dependencies
uv sync --all-extras

Security note: The setup script downloads a pinned uv release (v0.10.4) with SHA256 checksum verification to prevent supply-chain attacks. CI/CD workflows use the official astral-sh/setup-uv@v4 GitHub Action.

Running Tests

# Run all tests with coverage
uv run pytest

# Run with verbose output
uv run pytest -v

# Run specific test file
uv run pytest tests/test_client.py

# Generate HTML coverage report
uv run pytest --cov=src --cov-report=html

Code Quality

# Format code
uv run ruff format src tests

# Lint code
uv run ruff check src tests

# Type checking
uv run mypy src

# Run all checks
uv run ruff format src tests && uv run ruff check --fix src tests && uv run mypy src

Building and Publishing

# Build distribution packages
uv build

# Publish to TestPyPI (for testing)
uv publish --publish-url https://test.pypi.org/legacy/

# Publish to PyPI (production)
uv publish

Note: Publishing is automated via GitHub Actions. Push to main triggers the production release workflow.

License

MIT License - see LICENSE file for details.

Support

• Documentation: GitHub Repository
• Issues: GitHub Issues
• Email: [e-mail]