Design a Cloud-Provider Upload Adapter

Scenario

Your application allows users to upload large assets directly to cloud storage providers (such as AWS S3, Google Cloud Storage, and Azure Blob) to avoid saturating backend service bandwidth. However, each cloud provider has distinct conventions for generating presigned/signed URLs, maximum chunk sizes, multipart upload life-cycles, and error codes.

You are tasked with designing a unified StorageAdapter interface and an upload orchestration flow that isolates core business logic from cloud provider specifics.

Requirements

1. Unified Storage Adapter Interface:

   • Define an abstract base adapter class that concrete providers implement.
   • Methods must support generating direct-upload instructions (single-part presigned URL or multi-part sessions) and verifying completed uploads from the provider.
   • Expose explicit provider capabilities (e.g., min_part_size, max_part_size, supports_resumable_uploads).

2. Error Normalization:

   • Map vendor-specific network and protocol errors (e.g., AWS EntityTooLarge, Azure InvalidBlobOrBlock, GCS 403 Forbidden) into standardized application-level exceptions: UploadExpiredException, SizeLimitExceededException, ObjectNotFoundException, and ProviderTransientException.

3. Upload State & Backend-Verified Acknowledgement:

   • Track upload session states: PENDING, READY_FOR_UPLOAD, CONFIRMED, ABORTED.
   • Do not trust client claims alone. When the client reports completion, the backend must query the provider directly to verify object metadata (size, hash/ETag, existence) before transitioning to CONFIRMED.
   • Ensure acknowledgement endpoints are idempotent.

4. Security & Observability:

   • Ensure sensitive credentials, signature query parameters, and vendor authorization headers are sanitized before any log message is emitted.

Interface Sketch

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, Optional, List

@dataclass
class ProviderCapabilities:
    min_part_size_bytes: int
    max_part_size_bytes: int
    supports_multipart: bool

@dataclass
class DirectUploadRequest:
    file_id: str
    byte_size: int
    content_type: str
    checksum_sha256: Optional[str] = None

@dataclass
class UploadInstruction:
    upload_url: str
    method: str
    headers: Dict[str, str]
    expires_in_seconds: int
    parts: Optional[List[Dict[str, str]]] = None

@dataclass
class ObjectMetadata:
    file_id: str
    byte_size: int
    etag: str
    content_type: str

class CloudStorageAdapter(ABC):
    @abstractmethod
    def capabilities(self) -> ProviderCapabilities:
        pass

    @abstractmethod
    def create_upload_session(self, req: DirectUploadRequest) -> UploadInstruction:
        pass

    @abstractmethod
    def verify_upload(self, file_id: str, expected_size: int) -> ObjectMetadata:
        """Checks provider directly. Raises normalized exceptions if invalid or missing."""
        pass

    @abstractmethod
    def abort_upload(self, file_id: str) -> None:
        pass

Deliverables

• Provide the complete interface and class structure in Python (or your preferred language).
• Implement an example S3StorageAdapter or generic mock implementation demonstrating error normalization and URL sanitization.
• Describe the client-to-backend communication protocol for chunked/multipart uploads and how concurrent verification requests are handled idempotently.