Metadata-Version: 2.4
Name: cf_ai_platform_client
Version: 0.1.0
Summary: Official Python SDK for CloudFactory AI Platform - download and manage project assets and results
Project-URL: Homepage, https://github.com/cloudfactory/cf-ai-platform-client-python
Project-URL: Repository, https://github.com/cloudfactory/cf-ai-platform-client-python
Project-URL: Documentation, https://github.com/cloudfactory/cf-ai-platform-client-python#readme
Author-email: CloudFactory <support@cloudfactory.com>
License: MIT
License-File: LICENSE
Keywords: ai,api-client,cloudfactory,platform,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.6.0
Requires-Dist: tenacity>=8.2.0
Requires-Dist: tqdm>=4.66.0
Provides-Extra: dev
Requires-Dist: mypy>=1.8.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.1.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.2.0; extra == 'dev'
Requires-Dist: types-tqdm>=4.66.0; extra == 'dev'
Description-Content-Type: text/markdown


# CloudFactory AI Platform Python Client

[![CI](https://github.com/cloudfactory/cf-ai-platform-client-python/actions/workflows/ci.yml/badge.svg)](https://github.com/cloudfactory/cf-ai-platform-client-python/actions/workflows/ci.yml)
[![PyPI version](https://badge.fury.io/py/cf-ai-platform-client.svg)](https://badge.fury.io/py/cf-ai-platform-client)
[![Python Support](https://img.shields.io/pypi/pyversions/cf-ai-platform-client.svg)](https://pypi.org/project/cf-ai-platform-client/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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

```bash
uv add cf-ai-platform-client
```

Or with pip:

```bash
pip install cf-ai-platform-client
```

## Quick Start

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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):

```python
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:**
```python
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

```python
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](https://github.com/astral-sh/uv) for fast dependency management. The setup script installs uv with checksum verification for security:

```bash
# 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:

```bash
# 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

```bash
# 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

```bash
# 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

```bash
# 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](LICENSE) file for details.

## Support

- **Documentation**: [GitHub Repository](https://github.com/cloudfactory/cf-ai-platform-client-python)
- **Issues**: [GitHub Issues](https://github.com/cloudfactory/cf-ai-platform-client-python/issues)
- **Email**: support@cloudfactory.com
