File upload guide

import asyncio
import aiohttp

API_KEY = "your-api-key-here"
API_BASE = "https://mm-midmarket-integrations-api-preview.azurewebsites.net"

async def request_presigned_url(session, file_name, content_type):
    async with session.post(
        f"{API_BASE}/api/integrations/uploads/url",
        headers={"X-API-Key": API_KEY},
        json={"filename": file_name, "contentType": content_type},
    ) as resp:
        resp.raise_for_status()
        return await resp.json()

async def upload_to_azure(session, upload_url, file_path):
    with open(file_path, "rb") as f:
        async with session.put(
            upload_url,
            headers={"x-ms-blob-type": "BlockBlob"},
            data=f,
        ) as resp:
            resp.raise_for_status()

async def upload_files_parallel(files: list[tuple[str, str]]) -> list[str]:
    """Upload multiple files in parallel. files = [(path, contentType), ...]."""
    blob_paths = []
    async with aiohttp.ClientSession() as session:
        # Request all presigned URLs first
        url_data = await asyncio.gather(
            *[
                request_presigned_url(session, path.split("/")[-1], ctype)
                for path, ctype in files
            ]
        )
        # Upload all files concurrently
        await asyncio.gather(
            *[
                upload_to_azure(session, data["uploadUrl"], path)
                for data, (path, _) in zip(url_data, files)
            ]
        )
        blob_paths = [data["blobPath"] for data in url_data]
    return blob_paths

from azure.storage.blob import BlobClient

def upload_large_file(upload_url: str, file_path: str):
    """Upload a large file to the presigned SAS URL using the Azure SDK."""

    blob_client = BlobClient.from_blob_url(upload_url)

    with open(file_path, "rb") as f:
        blob_client.upload_blob(f, overwrite=True)

    print("Upload complete")

import time
from requests.exceptions import RequestException

def upload_with_retries(file_path: str, max_retries: int = 3) -> str:
    """Upload with automatic retry on transient failure."""

    for attempt in range(max_retries):
        try:
            return upload_file_direct(file_path)
        except RequestException:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Upload failed, retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                raise

import os

def validate_before_upload(file_path: str, max_mb: int) -> bool:
    """Check file exists, is readable, and within the category size limit."""

    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File not found: {file_path}")

    size_mb = os.path.getsize(file_path) / (1024 * 1024)
    if size_mb > max_mb:
        raise ValueError(f"File too large: {size_mb:.1f} MB (max {max_mb} MB)")

    if not os.access(file_path, os.R_OK):
        raise PermissionError(f"File not readable: {file_path}")

    return True