Cloud Storage Sources¶

craft-easy-file-import supports three managed cloud storage backends as file sources: Amazon S3 (and S3-compatible stores), Azure Blob Storage, and Google Cloud Storage. All three implement the full FileSource protocol: polling, event-driven ingress, file lifecycle (move/delete), and health checks.

Install the relevant extra before use:

pip install 'craft-easy-file-import[s3]'      # AWS, MinIO, R2, Wasabi
pip install 'craft-easy-file-import[azure]'   # Azure Blob Storage
pip install 'craft-easy-file-import[gcs]'     # Google Cloud Storage

S3 and S3-compatible stores¶

S3Source connects to AWS S3, MinIO, Cloudflare R2, or Wasabi. Set endpoint_url to override the AWS endpoint for compatible stores.

Polling¶

from craft_easy_file_import.sources.s3 import S3Source

source = S3Source(
    name="erp-drops",
    bucket="my-import-bucket",
    prefix="incoming/",          # only files under this prefix
    credential_name="erp_s3",    # key in the credential store
    region="eu-north-1",
)

poll() returns a list[FileInfo] — metadata only, no content is downloaded yet. The runner calls source.open(path) per file to stream content.

MinIO / Cloudflare R2¶

source = S3Source(
    name="minio-local",
    bucket="imports",
    endpoint_url="http://minio:9000",  # overrides the AWS endpoint
    credential_name="minio_creds",
    region="us-east-1",  # MinIO ignores this but aioboto3 requires it
)

Event-driven ingress via SQS¶

Configure S3 event notifications (s3:ObjectCreated:*) to publish to an SQS queue, then set event_queue_url. The runner calls subscribe_events() instead of poll() when this is set.

source = S3Source(
    name="erp-drops",
    bucket="my-import-bucket",
    prefix="incoming/",
    credential_name="erp_s3",
    region="eu-north-1",
    event_queue_url="https://sqs.eu-north-1.amazonaws.com/123456789/import-events",
)

subscribe_events() is an async generator that long-polls SQS (20 s wait, 10 messages per call), yields FileEvent(event_type="created", path=..., size=..., etag=...) for each new object, and deletes each SQS message after processing.

Both plain S3 event notifications and SNS-to-SQS fan-out (where the SQS message body is a nested JSON with a Message field) are handled automatically.

Credentials¶

Credentials are resolved at runtime from the credential store. The value stored under credential_name must be a JSON object:

{"access_key_id": "AKIAIOSFODNN7EXAMPLE", "secret_access_key": "wJalrXUtnFEM..."}

When no credential_name is set, aioboto3 falls back to the standard AWS credential chain (env vars, ~/.aws/credentials, instance profile).

Azure Blob Storage¶

AzureBlobSource connects to an Azure Blob Storage container. It supports two authentication modes: connection string and DefaultAzureCredential (Managed Identity / service principal).

from craft_easy_file_import.sources.azure_blob import AzureBlobSource

# Connection string auth (from credential store)
source = AzureBlobSource(
    name="azure-imports",
    container="file-imports",
    prefix="incoming/",
    connection_string_credential="azure_storage_conn_str",
)

# Managed Identity / DefaultAzureCredential
source = AzureBlobSource(
    name="azure-imports",
    container="file-imports",
    account_url="https://mystorageaccount.blob.core.windows.net",
)

One of connection_string_credential or account_url must be set. If both are set, connection_string_credential wins.

The connection string stored under connection_string_credential must be the raw Azure Storage connection string:

DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;EndpointSuffix=core.windows.net

Event-driven ingress via Event Grid + Service Bus¶

Set up an Azure Event Grid system topic on the storage account that routes Microsoft.Storage.BlobCreated events to a Service Bus queue. Then configure:

source = AzureBlobSource(
    name="azure-imports",
    container="file-imports",
    connection_string_credential="azure_storage_conn_str",
    service_bus_connection_credential="azure_sb_conn_str",
    service_bus_queue="import-events",
)

subscribe_events() connects to the Service Bus queue (30 s wait), parses Event Grid event payloads, yields FileEvent objects for BlobCreated events on blobs under the configured prefix, and completes each message after processing.

The azure-servicebus package is required for event-driven mode:

pip install azure-servicebus

Google Cloud Storage¶

GCSSource uses the google-cloud-storage SDK (sync API wrapped in asyncio.to_thread).

from craft_easy_file_import.sources.gcs import GCSSource

source = GCSSource(
    name="gcs-imports",
    bucket="my-import-bucket",
    prefix="incoming/",
    credential_name="gcs_service_account",  # JSON service account key
    project_id="my-gcp-project",
)

The credential stored under credential_name must be the full service account JSON key as a string. When credential_name is not set, Application Default Credentials are used.

Event-driven ingress via Cloud Pub/Sub¶

Enable GCS change notifications to push to a Pub/Sub topic, then create a subscription:

gcloud storage buckets notifications create gs://my-import-bucket \
  --topic=import-events --event-types=OBJECT_FINALIZE

source = GCSSource(
    name="gcs-imports",
    bucket="my-import-bucket",
    prefix="incoming/",
    project_id="my-gcp-project",
    pubsub_subscription="projects/my-gcp-project/subscriptions/import-sub",
)

subscribe_events() pulls up to 10 messages per call (30 s timeout), yields FileEvent for OBJECT_FINALIZE events on objects under prefix, and acknowledges messages in batch after processing.

The google-cloud-pubsub package is required:

pip install google-cloud-pubsub

Using cloud sources with the runner¶

Cloud sources drop in directly as the source in an ImportBinding:

from craft_easy_file_import.runner import ImportRunner, ImportBinding
from craft_easy_file_import.sources.s3 import S3Source

source = S3Source(
    name="erp-drops",
    bucket="my-import-bucket",
    prefix="incoming/",
    credential_name="erp_s3",
)

runner = ImportRunner(api_client=api)
bindings = [
    ImportBinding(
        name="erp-products",
        source=source,
        pipeline=pipeline,
        template=template,
    )
]

await runner.run_loop(lambda: bindings, poll_interval=300)

Processed files are moved to template.completed_folder (default: completed) or template.error_folder (default: error) within the same bucket/container using the source's move_file() method.

Health checks¶

All three sources implement health_check(), which verifies connectivity and returns a HealthStatus with latency:

status = await source.health_check()
# HealthStatus(healthy=True, latency_ms=42.3, last_successful_check=datetime(...))

S3 uses HeadBucket, Azure uses GetContainerProperties, GCS uses Bucket.reload().