Skip to main content

Cloud Module API Specification

Version: v0.2.0 | Status: Stable | Last Updated: February 2026

1. Overview​

The cloud module provides unified APIs for interacting with cloud storage, document, and infrastructure services. This specification covers all public interfaces across AWS, GCP, Azure, Coda.io, and Infomaniak providers.


2. Storage APIs​

2.1 AWS S3 (S3Client)​

from codomyrmex.cloud import S3Client

Constructor​

S3Client(region_name: Optional[str] = None)
  • Uses boto3 default credential chain

Methods​

MethodSignatureReturnsDescription
upload_file(file_path: str, bucket: str, object_name: Optional[str] = None) -> boolSuccess flagUpload local file to S3
download_file(bucket: str, object_name: str, file_path: str) -> boolSuccess flagDownload S3 object to local file
list_objects(bucket: str) -> list[str]Object keysList all objects in bucket
get_object_metadata(bucket: str, key: str) -> dict[str, Any]Metadata dictGet object metadata
create_bucket(name: str, region: Optional[str] = None) -> boolSuccess flagCreate a bucket

2.2 GCP Cloud Storage (GCSClient)​

from codomyrmex.cloud import GCSClient

Constructor​

GCSClient(project: Optional[str] = None)
  • Uses Application Default Credentials

Methods​

MethodSignatureReturnsDescription
upload_file(bucket: str, key: str, file_path: str) -> boolSuccess flagUpload file to GCS
download_file(bucket: str, key: str, file_path: str) -> boolSuccess flagDownload object to local file
list_objects`(bucket: str, prefix: strNone = None) -> list[str]`Object keys
get_object_metadata(bucket: str, key: str) -> dict[str, Any]Metadata dictGet object metadata
create_bucket(name: str, region: str = "US") -> boolSuccess flagCreate a bucket

2.3 Azure Blob Storage (AzureBlobClient)​

from codomyrmex.cloud import AzureBlobClient

Constructor​

AzureBlobClient(account_url: Optional[str] = None)
  • Falls back to AZURE_STORAGE_ACCOUNT_URL environment variable
  • Uses DefaultAzureCredential for authentication

Methods​

MethodSignatureReturnsDescription
upload_file(bucket: str, key: str, file_path: str) -> boolSuccess flagUpload file to a container
download_file(bucket: str, key: str, file_path: str) -> boolSuccess flagDownload object to local file
list_objects`(bucket: str, prefix: strNone = None) -> list[str]`Object keys
get_object_metadata(bucket: str, key: str) -> dict[str, Any]Metadata dictGet object metadata
create_bucket(name: str) -> boolSuccess flagCreate a container

3. Coda.io API (CodaClient)​

from codomyrmex.cloud import CodaClient

3.1 Constructor​

CodaClient(
api_token: str,
base_url: str = "https://coda.io/apis/v1",
timeout: int = 30
)

3.2 Documents API​

MethodSignatureDescription
list_docs(**filters) -> DocListList accessible documents
create_doc(title, **kwargs) -> DocCreate new document
get_doc(doc_id: str) -> DocGet document metadata
update_doc(doc_id: str, **kwargs) -> dictUpdate document
delete_doc(doc_id: str) -> dictDelete document

3.3 Pages API​

MethodSignatureDescription
list_pages(doc_id: str, **params) -> PageListList pages in doc
create_page(doc_id: str, **kwargs) -> dictCreate new page
get_page(doc_id: str, page_id: str) -> PageGet page details
update_page(doc_id: str, page_id: str, **kwargs) -> dictUpdate page
delete_page(doc_id: str, page_id: str) -> dictDelete page

3.4 Tables & Rows API​

MethodSignatureDescription
list_tables(doc_id: str, **params) -> TableListList tables in doc
get_table(doc_id: str, table_id: str) -> TableGet table details
list_columns(doc_id: str, table_id: str) -> ColumnListList table columns
list_rows(doc_id: str, table_id: str, **params) -> RowListList table rows
insert_rows(doc_id: str, table_id: str, rows: List) -> InsertRowsResultInsert/upsert rows
get_row(doc_id: str, table_id: str, row_id: str) -> RowGet single row
update_row(doc_id: str, table_id: str, row_id: str, **kwargs) -> dictUpdate row
delete_row(doc_id: str, table_id: str, row_id: str) -> dictDelete row

3.5 Exceptions​

ExceptionHTTP StatusDescription
CodaAPIErrorVariousBase exception for API errors
CodaAuthenticationError401Invalid or missing API token
CodaForbiddenError403Permission denied
CodaNotFoundError404Resource not found
CodaRateLimitError429Rate limit exceeded
CodaValidationError400Invalid request parameters
CodaGoneError410Resource deleted

4. Infomaniak Cloud APIs​

4.1 Overview​

from codomyrmex.cloud.infomaniak import (
InfomaniakComputeClient,
InfomaniakVolumeClient,
InfomaniakNetworkClient,
InfomaniakObjectStorageClient,
InfomaniakS3Client,
InfomaniakIdentityClient,
InfomaniakDNSClient,
InfomaniakHeatClient,
InfomaniakMeteringClient,
)
from codomyrmex.cloud.infomaniak.newsletter import InfomaniakNewsletterClient

All Infomaniak OpenStack clients support from_env() and from_credentials() factory methods.

4.2 Compute (InfomaniakComputeClient)​

MethodReturnsDescription
list_instances()List[Dict]List all compute instances
get_instance(server_id)Optional[Dict]Get instance details
create_instance(name, flavor, image, network, ...)Optional[Dict]Create a new instance
start_instance(server_id)boolStart a stopped instance
stop_instance(server_id)boolStop a running instance
reboot_instance(server_id, reboot_type)boolReboot an instance
delete_instance(server_id)boolDelete an instance
list_flavors()List[Dict]List available flavors
list_images()List[Dict]List available images
list_keypairs()List[Dict]List SSH key pairs

4.3 Volume (InfomaniakVolumeClient)​

MethodReturnsDescription
list_volumes()List[Dict]List block storage volumes
create_volume(size, name, ...)Optional[Dict]Create a volume
delete_volume(volume_id)boolDelete a volume
extend_volume(volume_id, new_size)boolExtend a volume
attach_volume(volume_id, server_id)boolAttach volume to instance
detach_volume(volume_id, server_id)boolDetach volume from instance
list_snapshots()List[Dict]List volume snapshots
list_backups()List[Dict]List volume backups

4.4 Network (InfomaniakNetworkClient)​

MethodReturnsDescription
list_networks()List[Dict]List networks
create_network(name, ...)Optional[Dict]Create a network
delete_network(network_id)boolDelete a network
list_routers()List[Dict]List routers
list_security_groups()List[Dict]List security groups
add_security_group_rule(...)Optional[Dict]Add a security group rule
list_floating_ips()List[Dict]List floating IPs
list_loadbalancers()List[Dict]List load balancers

4.5 Object Storage (InfomaniakObjectStorageClient / InfomaniakS3Client)​

Swift API:

MethodReturnsDescription
list_containers()List[str]List Swift containers
create_container(name)boolCreate a container
upload_object(container, name, data)boolUpload an object
download_object(container, name)bytesDownload an object
delete_object(container, name)boolDelete an object

S3-compatible API:

MethodReturnsDescription
list_buckets()List[str]List S3 buckets
list_objects(bucket)List[str]List objects in bucket
upload_data(bucket, key, data)boolUpload data
download_data(bucket, key)bytesDownload data
delete_object(bucket, key)boolDelete an object
generate_presigned_url(bucket, key, ...)Optional[str]Generate presigned URL
bucket_exists(bucket)boolCheck if bucket exists

4.6 Identity (InfomaniakIdentityClient)​

MethodReturnsDescription
get_current_user()Optional[Dict]Get authenticated user info
list_projects()List[Dict]List accessible projects
list_application_credentials()List[Dict]List app credentials
create_ec2_credentials()Optional[Dict]Create EC2-style credentials
delete_ec2_credentials(cred_id)boolDelete EC2 credentials

4.7 DNS (InfomaniakDNSClient)​

MethodReturnsDescription
list_zones()List[Dict]List DNS zones
create_zone(name, email, ...)Optional[Dict]Create a zone
delete_zone(zone_id)boolDelete a zone
list_records(zone_id)List[Dict]List records in zone
create_record(zone_id, name, type, records, ...)Optional[Dict]Create a record
delete_record(zone_id, record_id)boolDelete a record

4.8 Orchestration (InfomaniakHeatClient)​

MethodReturnsDescription
list_stacks()List[Dict]List Heat stacks
get_stack(stack_id)Optional[Dict]Get stack details
create_stack(name, template, ...)Optional[Dict]Create a stack
delete_stack(stack_id)boolDelete a stack
list_stack_resources(stack_id)List[Dict]List stack resources
validate_template(template)DictValidate a Heat template

4.9 Metering (InfomaniakMeteringClient)​

MethodReturnsDescription
get_all_usage()DictComprehensive usage summary
get_compute_usage()DictCompute usage (vCPUs, RAM, disk)
get_storage_usage()DictBlock storage usage
get_network_usage()DictNetwork resource usage
get_compute_quotas()DictCompute quota limits

4.10 Newsletter (InfomaniakNewsletterClient)​

from codomyrmex.cloud.infomaniak.newsletter import InfomaniakNewsletterClient
client = InfomaniakNewsletterClient.from_env() # uses INFOMANIAK_NEWSLETTER_TOKEN, INFOMANIAK_NEWSLETTER_ID

Campaigns:

MethodReturnsDescription
list_campaigns()List[Dict]List all campaigns
get_campaign(campaign_id)Optional[Dict]Get campaign details
create_campaign(subject, sender_email, sender_name, content_html, mailing_list_id)Optional[Dict]Create a campaign
update_campaign(campaign_id, **kwargs)Optional[Dict]Update a campaign
delete_campaign(campaign_id)boolDelete a campaign
send_test(campaign_id, email)boolSend test email
schedule_campaign(campaign_id, send_at)boolSchedule campaign
unschedule_campaign(campaign_id)boolUnschedule campaign
send_campaign(campaign_id)boolSend campaign immediately
get_campaign_statistics(campaign_id)Optional[Dict]Get campaign stats

Mailing Lists:

MethodReturnsDescription
list_mailing_lists()List[Dict]List mailing lists
get_mailing_list(list_id)Optional[Dict]Get mailing list details
create_mailing_list(name)Optional[Dict]Create a mailing list
update_mailing_list(list_id, **kwargs)Optional[Dict]Update a mailing list
delete_mailing_list(list_id)boolDelete a mailing list
get_list_contacts(list_id)List[Dict]Get contacts in list
import_contacts(list_id, contacts)Optional[Dict]Import contacts to list
manage_contact(list_id, contact_id, action)boolSubscribe/unsubscribe contact

Contacts:

MethodReturnsDescription
get_contact(contact_id)Optional[Dict]Get contact details
update_contact(contact_id, **kwargs)Optional[Dict]Update contact
delete_contact(contact_id)boolDelete contact

Utility:

MethodReturnsDescription
get_task_status(task_id)Optional[Dict]Check async task status
get_credits()Optional[Dict]Get newsletter credits

6. Common Abstractions​

from codomyrmex.cloud.common import (
CloudProvider,
ResourceType,
CloudCredentials,
CloudResource,
StorageClient,
ComputeClient,
ServerlessClient,
)

6.1 Enums​

class CloudProvider(Enum):
AWS = "aws"
GCP = "gcp"
AZURE = "azure"
INFOMANIAK = "infomaniak"
LOCAL = "local"

class ResourceType(Enum):
COMPUTE = "compute"
STORAGE = "storage"
DATABASE = "database"
NETWORK = "network"
SERVERLESS = "serverless"
CONTAINER = "container"
QUEUE = "queue"

6.2 StorageClient ABC​

class StorageClient(ABC):
@abstractmethod
def list_buckets(self) -> List[str]: ...

@abstractmethod
def create_bucket(self, name: str) -> bool: ...

@abstractmethod
def upload_file(self, bucket: str, key: str, data: bytes,
content_type: Optional[str] = None) -> str: ...

@abstractmethod
def download_file(self, bucket: str, key: str) -> bytes: ...

@abstractmethod
def delete_object(self, bucket: str, key: str) -> bool: ...

@abstractmethod
def generate_presigned_url(self, bucket: str, key: str,
expires_in: int = 3600) -> str: ...

7. Usage Examples​

7.1 Cross-Provider Storage​

from codomyrmex.cloud import S3Client, GCSClient, AzureBlobClient

# Choose provider based on environment
import os
provider = os.environ.get("CLOUD_PROVIDER", "aws")

if provider == "aws":
client = S3Client()
client.upload_file("data.csv", "my-bucket", "uploads/data.csv")
elif provider == "gcp":
client = GCSClient()
client.upload_file("my-bucket", "uploads/data.csv", "data.csv")
elif provider == "azure":
client = AzureBlobClient()
client.upload_file("my-container", "uploads/data.csv", "data.csv")

7.2 Coda.io Integration​

from codomyrmex.cloud import CodaClient, CodaRateLimitError
import time

client = CodaClient(api_token="your-token")

# Handle rate limiting
def safe_list_rows(doc_id, table_id, max_retries=3):
for attempt in range(max_retries):
try:
return client.list_rows(doc_id, table_id)
except CodaRateLimitError:
if attempt < max_retries - 1:
time.sleep(6) # Wait for rate limit window
else:
raise

8. Error Handling​

All storage clients log errors internally and return False or empty results on failure. For detailed error handling:

import logging
logging.getLogger("codomyrmex.cloud.aws").setLevel(logging.DEBUG)

from codomyrmex.cloud import S3Client
client = S3Client()

# Check return values
if not client.upload_file("local.txt", "bucket", "remote.txt"):
print("Upload failed - check logs for details")