Error handling guide

{
  "detail": "Invalid API key"
}

{
  "detail": "Invalid blob path"
}

{
  "detail": "Submission not found"
}

{
  "detail": "Asset with blobPath '<workspace-id>/campaign-video.mp4' already submitted"
}

{
  "detail": "File too large; use the presigned upload URL"
}

{
  "detail": "Unsupported content type"
}

import random
import time

import requests

API_BASE = "https://mm-midmarket-integrations-api-preview.azurewebsites.net"
RETRYABLE = {429, 500, 502}


def request_with_retries(
    method: str,
    url: str,
    max_retries: int = 4,
    base_delay: float = 1.0,
    **kwargs,
) -> requests.Response:
    """Make a request, retrying only 429/500/502 with backoff and jitter."""
    for attempt in range(max_retries):
        response = requests.request(method, url, timeout=10, **kwargs)

        if response.status_code not in RETRYABLE or attempt == max_retries - 1:
            return response

        # Honour Retry-After (seconds) on 429, otherwise back off exponentially.
        retry_after = response.headers.get("Retry-After")
        if response.status_code == 429 and retry_after:
            delay = float(retry_after)
        else:
            delay = base_delay * (2 ** attempt)
            delay += random.uniform(0, delay * 0.1)  # jitter

        print(f"HTTP {response.status_code}, retry {attempt + 1}/{max_retries} in {delay:.1f}s")
        time.sleep(delay)

    raise RuntimeError(f"Failed after {max_retries} attempts")


response = request_with_retries(
    "POST",
    f"{API_BASE}/api/integrations/submissions",
    headers={"X-API-Key": "your-api-key-here"},
    json={"assets": [{"blobPath": "<workspace-id>/campaign-video.mp4"}]},
)

import mimetypes
import os

ALLOWED_TYPES = {
    "video/mp4",
    "video/quicktime",
    "video/x-msvideo",
    "video/x-matroska",
    "audio/mpeg",
    "audio/wav",
    "audio/x-wav",
    "audio/aac",
    "audio/ogg",
    "image/jpeg",
    "image/png",
    "image/gif",
    "image/webp",
    "image/svg+xml",
    "application/pdf",
    "application/msword",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "text/plain",
}

# Maximum file size per category, in bytes.
MAX_SIZES = {
    "video": 5 * 1024**3,
    "audio": 500 * 1024**2,
    "image": 50 * 1024**2,
    "application": 100 * 1024**2,
    "text": 100 * 1024**2,
}


def validate_file(file_path: str) -> str:
    """Validate a file before upload. Returns the detected MIME type."""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File not found: {file_path}")
    if not os.access(file_path, os.R_OK):
        raise PermissionError(f"File not readable: {file_path}")

    mime_type, _ = mimetypes.guess_type(file_path)
    if mime_type not in ALLOWED_TYPES:
        raise ValueError(f"Unsupported content type: {mime_type}")

    category = mime_type.split("/", 1)[0]
    max_size = MAX_SIZES.get(category)
    size = os.path.getsize(file_path)
    if max_size and size > max_size:
        raise ValueError(f"File too large: {size} bytes (max {max_size} for {category})")

    return mime_type


try:
    validate_file("campaign-video.mp4")
except (FileNotFoundError, PermissionError, ValueError) as exc:
    print(f"Validation failed: {exc}")

import logging

import requests

logger = logging.getLogger(__name__)


def log_api_error(response: requests.Response) -> None:
    """Log an API error with the status code and message."""
    try:
        body = response.json()
    except ValueError:
        body = {}

    detail = body.get("detail") or body.get("message") or body.get("error")

    logger.error(
        "Integrations API error",
        extra={
            "status_code": response.status_code,
            "detail": detail,
            "fields": body.get("fields"),
            "url": response.url,
            "method": response.request.method,
        },
    )


response = requests.post("...")
if response.status_code >= 400:
    log_api_error(response)

import requests

API_BASE = "https://mm-midmarket-integrations-api-preview.azurewebsites.net"


def get_failed_assets(submission_id: str, api_key: str) -> list[dict]:
    """Return the assets that failed processing."""
    response = requests.get(
        f"{API_BASE}/api/integrations/submissions/{submission_id}",
        headers={"X-API-Key": api_key},
        timeout=10,
    )
    response.raise_for_status()
    submission = response.json()

    failed = [
        asset for asset in submission["assets"]
        if asset.get("status") == "failed"
    ]

    for asset in failed:
        message = asset.get("errorMessage")
        error_type = asset.get("errorType")
        retryable = asset.get("isRetryable", False)
        print(f"{asset['assetId']}: {error_type} - {message} (retryable={retryable})")

    return failed

import requests
from circuitbreaker import circuit

API_BASE = "https://mm-midmarket-integrations-api-preview.azurewebsites.net"


@circuit(failure_threshold=5, recovery_timeout=60)
def call_integrations_api(endpoint: str, **kwargs) -> dict:
    """Call the API with a circuit breaker."""
    response = requests.get(f"{API_BASE}{endpoint}", timeout=10, **kwargs)
    response.raise_for_status()
    return response.json()


submission_id = "550e8400-e29b-41d4-a716-446655440000"
try:
    submission = call_integrations_api(
        f"/api/integrations/submissions/{submission_id}",
        headers={"X-API-Key": "your-api-key-here"},
    )
    print(f"Status: {submission['status']} ({submission['submissionId']})")
except Exception as exc:
    print(f"Request failed or circuit open: {exc}")