aceiot-models 0.3.7

Pydantic models and API client for ACE IoT Aerodrome platform

aceiot-models

Pydantic models and API client for the ACE IoT Aerodrome Platform

Installation

pip install aceiot-models

For upload progress support:

pip install aceiot-models[upload-progress]

Features

• Pydantic Models: Type-safe data models for all ACE IoT entities
• Dual API Clients: Both synchronous and asynchronous clients for the ACE IoT API
• Async Support: Full async/await support with httpx for high-performance applications
• Pagination Support: Efficient handling of large datasets (both sync and async)
• Batch Processing: Process large operations in manageable chunks with concurrency control
• Type Hints: Full type annotations for better IDE support
• Error Handling: Robust error handling with detailed error information

Usage

Using the Synchronous API Client

from aceiot_models.api import APIClient, APIError

# Initialize the client
client = APIClient(
    base_url="https://flightdeck.aceiot.cloud/api",
    api_key="your-api-key"
)

# Or use environment variables:
# export ACEIOT_API_URL=https://flightdeck.aceiot.cloud/api
# export ACEIOT_API_KEY=your-api-key
client = APIClient()

# Get clients
clients = client.get_clients(page=1, per_page=20)

# Create a point
from aceiot_models import PointCreate

point = PointCreate(
    name="sensor/temperature/office",
    site_id=1,
    client_id=1,
    point_type="analog",
    collect_enabled=True
)
result = client.create_points([point.model_dump()])

Using the Asynchronous API Client

import asyncio
from aceiot_models.api import AsyncAPIClient, APIError

async def main():
    # Initialize the async client with context manager
    async with AsyncAPIClient(
        base_url="https://flightdeck.aceiot.cloud/api",
        api_key="your-api-key"
    ) as client:
        # Get clients asynchronously
        clients = await client.get_clients(page=1, per_page=20)

        # Make concurrent requests
        clients_task = client.get_clients()
        sites_task = client.get_sites()
        gateways_task = client.get_gateways()

        # Wait for all requests
        clients, sites, gateways = await asyncio.gather(
            clients_task, sites_task, gateways_task
        )

        print(f"Found {clients['total_items']} clients")
        print(f"Found {sites['total_items']} sites")
        print(f"Found {gateways['total_items']} gateways")

# Run the async function
asyncio.run(main())

Using Pagination

Synchronous Pagination:

from aceiot_models.api import PaginatedResults

# Paginate through all sites
paginator = PaginatedResults(client.get_sites, per_page=100)
for page_sites in paginator:
    for site in page_sites:
        print(f"Site: {site['name']}")

# Or get all items at once
all_sites = paginator.all_items()

Asynchronous Pagination:

from aceiot_models.api import AsyncPaginatedResults

# Async pagination through all sites
async def paginate_sites():
    paginator = AsyncPaginatedResults(client.get_sites, per_page=100)
    async for page_sites in paginator:
        for site in page_sites:
            print(f"Site: {site['name']}")

    # Or get all items at once
    all_sites = await paginator.all_items()

Batch Processing

Synchronous Batch Processing:

from aceiot_models.api import batch_process

# Process points in batches
points = [...]  # List of many points

def create_batch(batch):
    return client.create_points(batch)

results = batch_process(
    items=points,
    process_func=create_batch,
    batch_size=100,
    progress_callback=lambda current, total: print(f"{current}/{total}")
)

Asynchronous Batch Processing with Concurrency Control:

from aceiot_models.api import batch_process_async

# Process points in batches with concurrent requests
async def create_batch_async(batch):
    return await client.create_points(batch)

results = await batch_process_async(
    items=points,
    process_func=create_batch_async,
    batch_size=100,
    max_concurrent=5,  # Control concurrent requests
    progress_callback=lambda current, total: print(f"{current}/{total}")
)

Using the Models

from aceiot_models import Point, Sample, PointCreate

# Create a new point
point = PointCreate(
    name="sensor/temperature/office",
    site_id=1,
    client_id=1,
    point_type="analog",
    collect_enabled=True
)

Using the API Client

from aceiot_models.api import APIClient, APIError

# Initialize the client
client = APIClient(
    base_url="https://flightdeck.aceiot.cloud/api",
    api_key="your-api-key"
)

# Fetch sites
try:
    sites = client.get_sites(page=1, per_page=100)
    print(f"Found {sites['total_items']} sites")
except APIError as e:
    print(f"Error: {e}")

Pagination

from aceiot_models.api import PaginatedResults

# Iterate through all points page by page
paginator = PaginatedResults(client.get_points, per_page=500)
for page_points in paginator:
    print(f"Processing {len(page_points)} points...")
    # Process each page

# Or get all items at once
all_points = paginator.all_items()

Batch Processing

from aceiot_models.api import batch_process

# Process items in batches
def process_batch(items):
    return client.create_points(items)

results = batch_process(
    items=large_list_of_points,
    process_func=process_batch,
    batch_size=100,
    progress_callback=lambda current, total: print(f"{current}/{total}")
)

Configuration

The API client can be configured through environment variables:

• ACEIOT_API_URL: Base URL for the API
• ACEIOT_API_KEY: Your API key
• ACEIOT_API_TIMEOUT: Request timeout in seconds (default: 30)

Examples

See the examples directory for complete usage examples:

• Synchronous API Usage - Complete examples using the sync client
• Asynchronous API Usage - Complete examples using the async client with asyncio

Testing

Unit Tests

Run the unit tests:

pytest tests/ -v

Integration Tests

Integration tests verify the API client works with a real API endpoint. See tests/api/README_INTEGRATION.md for setup instructions.

# Set environment variables
export ACEIOT_API_URL="https://flightdeck.aceiot.cloud/api"
export ACEIOT_API_KEY="your-api-key"
export ACEIOT_INTEGRATION_TESTS="true"

# Run integration tests
pytest tests/api/test_integration.py -v

Development

This project uses uv for dependency management. To set up a development environment:

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Clone the repository
git clone https://github.com/yourusername/aceiot-models.git
cd aceiot-models

# Install dependencies
uv sync --all-extras

# Run tests
uv run pytest

# Run linting
uv run ruff check .
uv run ruff format .

# Run type checking
uv run pyrefly check aceiot_models/

License

MIT License