Error Handling, Retries, and Idempotency¶

import httpx

async def ensure_namespace(
    client: httpx.AsyncClient,
    tenant: str,
    name: str,
    hard_quota: str = "100 GB",
) -> dict:
    """Create a namespace, or return the existing one."""
    # Try to create
    resp = await client.put(
        f"/mapi/tenants/{tenant}/namespaces",
        json={"name": name, "hardQuota": hard_quota},
    )
    match resp.status_code:
        case 201 | 200:
            print(f"Created namespace '{name}'")
            return resp.json()
        case 409:
            # Already exists -- fetch and return it
            resp = await client.get(
                f"/mapi/tenants/{tenant}/namespaces/{name}",
                params={"verbose": True},
            )
            resp.raise_for_status()
            print(f"Namespace '{name}' already exists")
            return resp.json()
        case _:
            resp.raise_for_status()  # raises for unexpected errors
            return {}  # unreachable, satisfies type checker

# Bash equivalent -- create-if-not-exists
CODE=$(curl -s -o /dev/null -w "%{http_code}" -X PUT \
  "$BASE/mapi/tenants/$TENANT/namespaces" \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d "{\"name\": \"$NS\", \"hardQuota\": \"100 GB\"}")

if [ "$CODE" = "201" ] || [ "$CODE" = "200" ]; then
  echo "Created namespace $NS"
elif [ "$CODE" = "409" ]; then
  echo "Namespace $NS already exists (OK)"
else
  echo "Failed to create namespace: HTTP $CODE" >&2
  exit 1
fi

templates:
  - name: process-data
    retryStrategy:
      limit: 3                    # max 3 retries (4 total attempts)
      retryPolicy: Always         # retry on both errors and node failures
      backoff:
        duration: "10s"           # initial delay
        factor: 2                 # exponential: 10s, 20s, 40s
        maxDuration: "2m"         # cap at 2 minutes
    activeDeadlineSeconds: 600    # hard timeout: kill pod after 10 min
    container:
      image: python:3.13-slim
      command: [python, -c]
      args:
        - |
          # your processing code here
          print("processing...")

from hera.workflows import models as m, script

@script(
    image="python:3.13-slim",
    retry_strategy=m.RetryStrategy(
        limit="3",
        retry_policy="Always",
        backoff=m.Backoff(duration="10s", factor=2, max_duration="2m"),
    ),
    active_deadline_seconds=600,
)
def process_data():
    """Processing with Argo-managed retries and timeout."""
    print("processing...")

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: hcp-safe-upload-
spec:
  entrypoint: upload-pipeline
  onExit: cleanup                    # always runs, even on failure
  arguments:
    parameters:
      - name: hcp-api-base
        value: "http://hcp-api.default.svc:8000/api/v1"
      - name: hcp-token
        value: "<your-token>"
      - name: bucket
        value: "datasets"
  templates:
    - name: upload-pipeline
      steps:
        - - name: upload
            template: do-upload

    - name: do-upload
      retryStrategy:
        limit: 2
      script:
        image: python:3.13-slim
        command: [python]
        source: |
          # ... upload logic ...
          print("uploading data")

    # ── Guaranteed cleanup ──────────────────────────────────────
    - name: cleanup
      script:
        image: curlimages/curl:latest
        command: [sh]
        source: |
          BASE="{{workflow.parameters.hcp-api-base}}"
          TOKEN="{{workflow.parameters.hcp-token}}"
          BUCKET="{{workflow.parameters.bucket}}"
          STATUS="{{workflow.status}}"    # Succeeded / Failed / Error

          if [ "$STATUS" != "Succeeded" ]; then
            echo "Workflow $STATUS -- cleaning up partial uploads..."

            # Abort any in-progress multipart uploads
            # Delete partial results to avoid polluting the bucket
            curl -s -X POST "$BASE/buckets/$BUCKET/objects/delete" \
              -H "Authorization: Bearer $TOKEN" \
              -H "Content-Type: application/json" \
              -d "{\"keys\": [\"partial/{{workflow.name}}/\"]}" || true

            echo "Cleanup complete"
          else
            echo "Workflow succeeded -- no cleanup needed"
          fi

from hera.workflows import DAG, Parameter, Steps, Workflow, models as m, script

HCP_BASE = "http://hcp-api.default.svc:8000/api/v1"


@script(image="python:3.13-slim", retry_strategy=m.RetryStrategy(limit="2"))
def do_upload(hcp_api_base: str, hcp_token: str, bucket: str):
    """Upload data to HCP."""
    print("uploading data")


@script(image="curlimages/curl:latest")
def cleanup(hcp_api_base: str, hcp_token: str, bucket: str):
    """Clean up partial uploads on failure. Runs as onExit handler."""
    import subprocess, os

    status = os.environ.get("ARGO_WORKFLOW_STATUS", "Unknown")
    if status != "Succeeded":
        print(f"Workflow {status} -- cleaning up...")
        subprocess.run(
            [
                "curl", "-s", "-X", "POST",
                f"{hcp_api_base}/buckets/{bucket}/objects/delete",
                "-H", f"Authorization: Bearer {hcp_token}",
                "-H", "Content-Type: application/json",
                "-d", '{"keys": ["partial/"]}',
            ],
            check=False,  # don't fail cleanup on errors
        )
        print("Cleanup complete")
    else:
        print("Workflow succeeded -- no cleanup needed")


with Workflow(
    generate_name="hcp-safe-upload-",
    entrypoint="upload-pipeline",
    on_exit="cleanup",
    arguments=[
        Parameter(name="hcp-api-base", value=HCP_BASE),
        Parameter(name="hcp-token", value="<your-token>"),
        Parameter(name="bucket", value="datasets"),
    ],
) as w:
    with Steps(name="upload-pipeline"):
        do_upload(
            name="upload",
            arguments={
                "hcp_api_base": "{{workflow.parameters.hcp-api-base}}",
                "hcp_token": "{{workflow.parameters.hcp-token}}",
                "bucket": "{{workflow.parameters.bucket}}",
            },
        )

    # Exit handler template (registered by name via on_exit="cleanup")
    cleanup(
        name="cleanup",
        arguments={
            "hcp_api_base": "{{workflow.parameters.hcp-api-base}}",
            "hcp_token": "{{workflow.parameters.hcp-token}}",
            "bucket": "{{workflow.parameters.bucket}}",
        },
    )

w.create()

import httpx

BASE = "http://localhost:8000/api/v1"

async def staged_upload(
    token: str,
    bucket: str,
    final_key: str,
    data: bytes,
    workflow_id: str,
):
    """Upload to a staging key, verify, then copy to the final location."""
    headers = {"Authorization": f"Bearer {token}"}
    staging_key = f"staging/{workflow_id}/{final_key}"

    async with httpx.AsyncClient(base_url=BASE, headers=headers, timeout=60.0) as c:
        # 1. Write to staging
        resp = await c.put(
            f"/buckets/{bucket}/objects/{staging_key}",
            content=data,
        )
        resp.raise_for_status()

        # 2. Verify — HEAD to confirm size
        resp = await c.head(f"/buckets/{bucket}/objects/{staging_key}")
        resp.raise_for_status()
        actual_size = int(resp.headers.get("content-length", 0))
        if actual_size != len(data):
            # Mismatch — delete staging and raise
            await c.delete(f"/buckets/{bucket}/objects/{staging_key}")
            raise RuntimeError(
                f"Size mismatch: expected {len(data)}, got {actual_size}"
            )

        # 3. Commit — copy from staging to final
        resp = await c.post(
            f"/buckets/{bucket}/objects/{final_key}/copy",
            json={"source_bucket": bucket, "source_key": staging_key},
        )
        resp.raise_for_status()

        # 4. Clean up staging
        await c.delete(f"/buckets/{bucket}/objects/{staging_key}")

        print(f"Committed {final_key} ({actual_size} bytes)")

# Bash equivalent — staged commit
STAGING_KEY="staging/$(uuidgen)/$FINAL_KEY"

# 1. Write to staging
curl -s -X PUT "$BASE/buckets/$BUCKET/objects/$STAGING_KEY" \
  -H "$AUTH" --data-binary @"$LOCAL_FILE"

# 2. Verify
SIZE=$(curl -s -I "$BASE/buckets/$BUCKET/objects/$STAGING_KEY" \
  -H "$AUTH" | grep -i content-length | awk '{print $2}' | tr -d '\r')
EXPECTED=$(stat -f%z "$LOCAL_FILE")
if [ "$SIZE" != "$EXPECTED" ]; then
  echo "ERROR: Size mismatch ($SIZE != $EXPECTED)" >&2
  curl -s -X DELETE "$BASE/buckets/$BUCKET/objects/$STAGING_KEY" -H "$AUTH"
  exit 1
fi

# 3. Commit — copy to final location
curl -s -X POST "$BASE/buckets/$BUCKET/objects/$FINAL_KEY/copy" \
  -H "$AUTH" -H "Content-Type: application/json" \
  -d "{\"source_bucket\": \"$BUCKET\", \"source_key\": \"$STAGING_KEY\"}"

# 4. Clean up staging
curl -s -X DELETE "$BASE/buckets/$BUCKET/objects/$STAGING_KEY" -H "$AUTH"

echo "Committed $FINAL_KEY"

from rahcp_client import HCPClient
from pathlib import Path

async def staged_upload_sdk(
    bucket: str,
    local_dir: Path,
    staging_prefix: str,
    final_prefix: str,
):
    """Upload files to staging, then commit to final — or clean up on failure."""
    async with HCPClient.from_env() as client:
        try:
            # 1. Write all files to staging
            for f in local_dir.rglob("*"):
                if f.is_file():
                    key = f"{staging_prefix}{f.relative_to(local_dir)}"
                    await client.s3.upload(bucket, key, f)

            # 2. Commit: copy staging → final, delete staging
            count = await client.s3.commit_staging(bucket, staging_prefix, final_prefix)
            print(f"Committed {count} files to {final_prefix}")

        except Exception:
            # 3. Rollback: delete staging prefix
            deleted = await client.s3.cleanup_staging(bucket, staging_prefix)
            print(f"Rolled back {deleted} staged files")
            raise