Cloud Storage

from aion import AionVision

async with AionVision(api_key="aion_...") as client:
    # Start Google Drive OAuth
    auth = await client.cloud_storage.initiate_auth(
        provider="google_drive"
    )

    print(f"Authorization URL: {auth.authorization_url}")
    print(f"State token: {auth.state}")  # Save for CSRF verification
    # Redirect user to auth.authorization_url to grant permissions

# After user authorizes, complete the flow with the callback code
result = await client.cloud_storage.complete_auth(
    provider="google_drive",
    code="auth_code_from_callback",
    state="state_from_initiate"
)

print(f"Connection ID: {result.connection.id}")
print(f"Email: {result.connection.provider_email}")
print(f"New connection: {result.is_new}")

connections = await client.cloud_storage.list_connections(
    active_only=True
)

print(f"Total connections: {connections.total_count}")

for conn in connections.connections:
    print(f"  {conn.provider}: {conn.provider_email}")
    print(f"    Active: {conn.is_active}")
    print(f"    Connected: {conn.created_at}")

result = await client.cloud_storage.disconnect(
    connection_id="conn_abc123"
)

print(f"Success: {result.success}")
print(f"Message: {result.message}")

from aion import CloudFileInput

# Import specific files from Google Drive
result = await client.cloud_storage.start_import(
    connection_id="conn_abc123",
    files=[
        CloudFileInput(id="gdrive_file_id_1", name="photo1.jpg", mime_type="image/jpeg"),
        CloudFileInput(id="gdrive_file_id_2", name="photo2.png", mime_type="image/png"),
    ],
    auto_describe=True,  # Auto-generate AI descriptions
    tags=["imported", "google-drive"]
)

if result.is_async:
    # Large import runs as background job
    print(f"Job ID: {result.job_id}")
    print(f"Status: {result.status}")
else:
    # Small import completed synchronously
    print(f"Imported {result.total_files} files")
    print(f"Image IDs: {result.image_ids}")

from aion import CloudFileInput, CloudStorageJobProgressEvent

def on_progress(event: CloudStorageJobProgressEvent):
    print(f"Progress: {event.progress_percent:.0f}% "
          f"({event.job.completed_files}/{event.job.total_files})")

job = await client.cloud_storage.import_and_wait(
    connection_id="conn_abc123",
    files=[
        CloudFileInput(id="gdrive_file_1", name="photo1.jpg"),
        CloudFileInput(id="gdrive_file_2", name="photo2.png"),
    ],
    auto_describe=True,
    on_progress=on_progress,
    timeout=300,  # Max 5 minutes
)

print(f"Job status: {job.status}")  # "completed" or "partial"
print(f"Completed: {job.completed_files}/{job.total_files}")
if job.failed_files > 0:
    print(f"Failed: {job.failed_files}")

# Dicts work too - useful when forwarding from a file picker
result = await client.cloud_storage.start_import(
    connection_id="conn_abc123",
    files=[
        {"id": "gdrive_file_id_1", "name": "photo1.jpg", "mime_type": "image/jpeg"},
        {"id": "gdrive_file_id_2", "name": "photo2.png"},
    ],
)

# Export images to Google Drive
result = await client.cloud_storage.start_export(
    connection_id="conn_abc123",
    image_ids=["img_001", "img_002", "img_003"],
    folder_name="Aionvision Export"  # Create a new folder in Drive
)

if result.is_async:
    print(f"Export job: {result.job_id}")
    print(f"Status: {result.job_status}")
else:
    print(f"Exported: {result.cloud_file_ids}")
    print(f"Message: {result.message}")

job = await client.cloud_storage.export_and_wait(
    connection_id="conn_abc123",
    image_ids=["img_001", "img_002", "img_003"],
    folder_name="Aionvision Export",
    timeout=300,
)

print(f"Export {job.status}: {job.completed_files}/{job.total_files} files")

# Get current job status
job = await client.cloud_storage.get_job(job_id="job_abc123")

print(f"Type: {job.type}")        # "import" or "export"
print(f"Status: {job.status}")    # pending | in_progress | completed | partial | failed
print(f"Progress: {job.progress_percent:.0f}%")
print(f"Completed: {job.completed_files}/{job.total_files}")
print(f"Failed: {job.failed_files}")

if job.error:
    print(f"Error: {job.error}")

# If you started with start_import/start_export, wait for completion
result = await client.cloud_storage.start_import(
    connection_id="conn_abc123",
    files=files,
)

if result.is_async and result.job_id:
    job = await client.cloud_storage.wait_for_job(
        result.job_id,
        timeout=300,       # Max wait time in seconds
        poll_interval=2,   # Check every 2 seconds
    )
    print(f"Final status: {job.status}")

from aion import SyncAionVision, CloudFileInput

with SyncAionVision(api_key="aion_...") as client:
    # All methods work the same, without await
    connections = client.cloud_storage.list_connections()

    job = client.cloud_storage.import_and_wait(
        connection_id="conn_abc123",
        files=[CloudFileInput(id="gdrive_file_1", name="photo.jpg")],
    )
    print(f"Import {job.status}: {job.completed_files} files")