SDK API Reference¶

@classmethod
def from_env(cls) -> HCPClient:
    """Create a client configured from ``HCP_*`` environment variables."""
    settings = HCPSettings()
    return cls(
        endpoint=settings.endpoint,
        username=settings.username,
        password=settings.password,
        tenant=settings.tenant,
        timeout=settings.timeout,
        max_retries=settings.max_retries,
        retry_base_delay=settings.retry_base_delay,
        multipart_threshold=settings.multipart_threshold,
        multipart_chunk=settings.multipart_chunk,
        multipart_concurrency=settings.multipart_concurrency,
        verify_ssl=settings.verify_ssl,
    )

async def request(
    self,
    method: str,
    path: str,
    *,
    params: dict[str, Any] | None = None,
    json: Any = None,
    data: Any = None,
    content: bytes | None = None,
    headers: dict[str, str] | None = None,
) -> httpx.Response:
    """Send an HTTP request with retry, tracing, and automatic 401 refresh."""
    merged_headers = {**self._auth_headers(), **(headers or {})}
    refreshed = False

    with tracer.start_as_current_span(f"{method} {path}") as span:
        span.set_attribute("http.method", method)
        span.set_attribute("http.path", path)

        async for attempt in AsyncRetrying(
            stop=stop_after_attempt(self.max_retries + 1),
            wait=wait_exponential_jitter(
                initial=self.retry_base_delay,
                max=60,
                jitter=self.retry_base_delay,
            ),
            retry=retry_if_exception_type(RetryableError),
            reraise=True,
        ):
            with attempt:
                t0 = time.monotonic()
                try:
                    response = await self._http.request(
                        method,
                        path,
                        params=params,
                        json=json,
                        data=data,
                        content=content,
                        headers=merged_headers,
                    )
                except httpx.TransportError as exc:
                    duration = (time.monotonic() - t0) * 1000
                    log.warning(
                        "%s %s — transport error after %.0fms: %s",
                        method,
                        path,
                        duration,
                        exc,
                    )
                    raise RetryableError(str(exc)) from exc

                duration = (time.monotonic() - t0) * 1000
                span.set_attribute("http.status_code", response.status_code)

                log.debug(
                    "%s %s → %d (%.0fms)",
                    method,
                    path,
                    response.status_code,
                    duration,
                )

                # One-time token refresh on 401
                if response.status_code == 401 and self.username and not refreshed:
                    log.info("Token expired, refreshing...")
                    await self._login()
                    merged_headers = {**self._auth_headers(), **(headers or {})}
                    refreshed = True
                    raise RetryableError("token refresh")

                if response.status_code in _RETRYABLE_STATUSES:
                    raise error_for_status(
                        response.status_code, _redact(response.text)
                    )

                if response.status_code >= 400:
                    safe_body = _redact(response.text)
                    log_fn = log.debug if response.status_code == 404 else log.error
                    log_fn(
                        "%s %s → %d (%.0fms): %s",
                        method,
                        path,
                        response.status_code,
                        duration,
                        safe_body,
                    )
                    raise error_for_status(response.status_code, safe_body)

                return response

    raise RetryableError("All retries exhausted")  # pragma: no cover

async def presign_get(self, bucket: str, key: str, *, expires: int = 3600) -> str:
    """Get a presigned download URL."""
    resp = await self._client.request(
        "POST",
        "/presign",
        json={
            "bucket": bucket,
            "key": key,
            "method": "get_object",
            "expires_in": expires,
        },
    )
    return resp.json()["url"]

async def presign_put(self, bucket: str, key: str, *, expires: int = 3600) -> str:
    """Get a presigned upload URL."""
    resp = await self._client.request(
        "POST",
        "/presign",
        json={
            "bucket": bucket,
            "key": key,
            "method": "put_object",
            "expires_in": expires,
        },
    )
    return resp.json()["url"]

async def presign_bulk(
    self,
    bucket: str,
    keys: list[str],
    *,
    method: str = "get_object",
    expires: int = 3600,
) -> dict[str, str]:
    """Presign multiple keys in one API call. Returns {key: url} mapping."""
    resp = await self._client.request(
        "POST",
        f"/buckets/{bucket}/objects/presign",
        json={
            "keys": keys,
            "method": method,
            "expires_in": expires,
        },
    )
    return {item["key"]: item["url"] for item in resp.json()["urls"]}

async def upload(self, bucket: str, key: str, data: bytes | Path) -> str:
    """Upload data — auto-selects presigned PUT or multipart based on size.

    Returns the ETag of the uploaded object.
    """
    with tracer.start_as_current_span("s3.upload") as span:
        span.set_attribute("s3.bucket", bucket)
        span.set_attribute("s3.key", key)

        if isinstance(data, Path):
            size = await asyncio.to_thread(lambda: data.stat().st_size)
            span.set_attribute("s3.size", size)
            if size >= self._client.multipart_threshold:
                return await self.upload_multipart(bucket, key, data)
            content = await asyncio.to_thread(data.read_bytes)
        else:
            content = data
            span.set_attribute("s3.size", len(content))

        url = await self.presign_put(bucket, key)

        async def _put() -> str:
            async with self._make_http_client() as http:
                resp = await http.put(url, content=content)
            raise_if_transient(resp, bucket, key)
            _raise_for_presigned(resp, bucket, key)
            return resp.headers.get("etag", "")

        etag = await transfer_with_retry(
            _put,
            max_attempts=self._client.max_retries + 1,
            base_delay=self._client.retry_base_delay,
        )
        log.debug("Uploaded %s/%s (%s bytes)", bucket, key, len(content))
        return etag

async def download(self, bucket: str, key: str, dest: Path) -> int:
    """Download an object via presigned GET. Returns byte count."""
    with tracer.start_as_current_span("s3.download") as span:
        span.set_attribute("s3.bucket", bucket)
        span.set_attribute("s3.key", key)
        url = await self.presign_get(bucket, key)

        async def _get() -> int:
            # Each attempt re-opens dest in "wb", discarding any partial
            # bytes written before a mid-stream transport error.
            total = 0
            async with self._make_http_client() as http:
                async with http.stream("GET", url) as resp:
                    raise_if_transient(resp, bucket, key)
                    _raise_for_presigned(resp, bucket, key)
                    file_handle = await asyncio.to_thread(lambda: dest.open("wb"))
                    try:
                        async for chunk in resp.aiter_bytes(chunk_size=8192):
                            await asyncio.to_thread(file_handle.write, chunk)
                            total += len(chunk)
                    finally:
                        await asyncio.to_thread(file_handle.close)
            return total

        total = await transfer_with_retry(
            _get,
            max_attempts=self._client.max_retries + 1,
            base_delay=self._client.retry_base_delay,
        )
        span.set_attribute("s3.bytes", total)
        log.debug("Downloaded %s/%s (%d bytes)", bucket, key, total)
        return total

async def download_bytes(self, bucket: str, key: str) -> bytes:
    """Download an object as bytes via presigned GET."""
    url = await self.presign_get(bucket, key)

    async def _get() -> bytes:
        async with self._make_http_client() as http:
            resp = await http.get(url)
        raise_if_transient(resp, bucket, key)
        _raise_for_presigned(resp, bucket, key)
        return resp.content

    return await transfer_with_retry(
        _get,
        max_attempts=self._client.max_retries + 1,
        base_delay=self._client.retry_base_delay,
    )

async def upload_multipart(
    self,
    bucket: str,
    key: str,
    path: Path,
    *,
    concurrency: int | None = None,
) -> str:
    """Presigned multipart upload for large files.

    1. Initiate multipart → upload_id
    2. Presign each part
    3. Upload parts in parallel (bounded semaphore)
    4. Complete multipart (or abort on failure)
    """
    concurrency = concurrency or self._client.multipart_concurrency
    file_size = await asyncio.to_thread(lambda: path.stat().st_size)
    default_part_size = self._client.multipart_chunk

    resp = await self._client.request(
        "POST",
        f"/buckets/{bucket}/multipart/{key}",
    )
    upload_id = resp.json()["upload_id"]

    try:
        resp = await self._client.request(
            "POST",
            f"/buckets/{bucket}/multipart/{key}/presign",
            json={
                "upload_id": upload_id,
                "file_size": file_size,
                "part_size": default_part_size,
            },
        )
        data = resp.json()
        part_urls = [p["url"] for p in data["urls"]]
        part_size = data.get("part_size", default_part_size)

        sem = asyncio.Semaphore(concurrency)

        part_timeout = httpx.Timeout(
            max(300.0, self._client.timeout * 5), connect=30.0
        )

        async def _upload_part(part_num: int, url: str) -> dict[str, Any]:
            offset = part_num * part_size
            read_size = min(part_size, file_size - offset)

            def _read_chunk() -> bytes:
                with path.open("rb") as f:
                    f.seek(offset)
                    return f.read(read_size)

            part_data = await asyncio.to_thread(_read_chunk)
            label = f"{key} (part {part_num + 1})"

            async def _put_part() -> dict[str, Any]:
                async with httpx.AsyncClient(
                    verify=self._client.verify_ssl, timeout=part_timeout
                ) as http:
                    resp = await http.put(url, content=part_data)
                raise_if_transient(resp, bucket, label)
                _raise_for_presigned(resp, bucket, label)
                return {"part_number": part_num + 1, "etag": resp.headers["etag"]}

            async with sem:
                log.debug(
                    "Uploading part %d/%d (%d bytes) for %s/%s",
                    part_num + 1,
                    len(part_urls),
                    len(part_data),
                    bucket,
                    key,
                )
                return await transfer_with_retry(
                    _put_part,
                    max_attempts=self._client.max_retries + 1,
                    base_delay=self._client.retry_base_delay,
                )

        parts = await asyncio.gather(
            *[_upload_part(i, url) for i, url in enumerate(part_urls)]
        )

        parts_sorted = sorted(parts, key=lambda p: p["part_number"])
        resp = await self._client.request(
            "POST",
            f"/buckets/{bucket}/multipart/{key}/complete",
            json={
                "upload_id": upload_id,
                "parts": [
                    {"ETag": p["etag"], "PartNumber": p["part_number"]}
                    for p in parts_sorted
                ],
            },
        )
        log.info(
            "Multipart upload complete: %s/%s (%d parts)",
            bucket,
            key,
            len(parts_sorted),
        )
        return resp.json().get("etag", "")

    except Exception:
        log.warning("Multipart upload failed, aborting: %s/%s", bucket, key)
        try:
            await self._client.request(
                "POST",
                f"/buckets/{bucket}/multipart/{key}/abort",
                json={"upload_id": upload_id},
            )
        except Exception:
            log.warning("Failed to abort multipart upload: %s", upload_id)
        raise

async def list_buckets(self) -> dict[str, Any]:
    """List all S3 buckets."""
    resp = await self._client.request("GET", "/buckets")
    return resp.json()

async def list_objects(
    self,
    bucket: str,
    prefix: str = "",
    *,
    max_keys: int = 1000,
    continuation_token: str | None = None,
    delimiter: str | None = None,
) -> dict[str, Any]:
    """List objects in a bucket with optional prefix and pagination."""
    params: dict[str, Any] = {"prefix": prefix, "max_keys": max_keys}
    if continuation_token:
        params["continuation_token"] = continuation_token
    if delimiter:
        params["delimiter"] = delimiter
    resp = await self._client.request(
        "GET",
        f"/buckets/{bucket}/objects",
        params=params,
    )
    return resp.json()

async def delete(self, bucket: str, key: str) -> None:
    """Delete a single object."""
    await self._client.request("DELETE", f"/buckets/{bucket}/objects/{key}")

async def delete_bulk(self, bucket: str, keys: list[str]) -> dict[str, Any]:
    """Delete multiple objects."""
    resp = await self._client.request(
        "POST",
        f"/buckets/{bucket}/objects/delete",
        json={"keys": keys},
    )
    return resp.json()

async def copy(
    self,
    bucket: str,
    key: str,
    source_bucket: str,
    source_key: str,
) -> None:
    """Copy an object."""
    await self._client.request(
        "POST",
        f"/buckets/{bucket}/objects/{key}/copy",
        json={
            "source_bucket": source_bucket,
            "source_key": source_key,
        },
    )

async def head(self, bucket: str, key: str) -> dict[str, Any]:
    """Get object metadata (content-length, content-type, etag, last-modified)."""
    resp = await self._client.request("HEAD", f"/buckets/{bucket}/objects/{key}")
    return dict(resp.headers)

async def commit_staging(
    self, bucket: str, staging_prefix: str, dest_prefix: str
) -> int:
    """Move objects from staging prefix to destination. Returns count."""
    count = 0
    async for obj in self._paginate_objects(bucket, staging_prefix):
        src_key = obj["Key"]
        dest_key = src_key.replace(staging_prefix, dest_prefix, 1)
        await self.copy(bucket, dest_key, bucket, src_key)
        await self.delete(bucket, src_key)
        count += 1
    return count

async def cleanup_staging(self, bucket: str, staging_prefix: str) -> int:
    """Delete all objects under a staging prefix. Paginates. Returns count."""
    total = 0
    batch: list[str] = []
    async for obj in self._paginate_objects(bucket, staging_prefix):
        batch.append(obj["Key"])
        if len(batch) >= 1000:
            await self.delete_bulk(bucket, batch)
            total += len(batch)
            batch.clear()
    if batch:
        await self.delete_bulk(bucket, batch)
        total += len(batch)
    return total

async def list_namespaces(
    self, tenant: str, *, verbose: bool = False
) -> dict[str, Any]:
    """List all namespaces for a tenant.

    Returns the MAPI response shape ``{"name": ["ns1", "ns2", ...]}``.
    """
    params: dict[str, Any] = {}
    if verbose:
        params["verbose"] = "true"
    resp = await self._client.request(
        "GET", f"/mapi/tenants/{tenant}/namespaces", params=params
    )
    return resp.json()

async def get_namespace(
    self, tenant: str, ns: str, *, verbose: bool = False
) -> dict[str, Any]:
    """Get namespace details."""
    params: dict[str, Any] = {}
    if verbose:
        params["verbose"] = "true"
    resp = await self._client.request(
        "GET", f"/mapi/tenants/{tenant}/namespaces/{ns}", params=params
    )
    return resp.json()

async def create_namespace(
    self, tenant: str, ns_data: dict[str, Any]
) -> dict[str, Any]:
    """Create a namespace. Uses PUT (MAPI convention)."""
    resp = await self._client.request(
        "PUT", f"/mapi/tenants/{tenant}/namespaces", json=ns_data
    )
    return resp.json()

async def update_namespace(
    self, tenant: str, ns: str, data: dict[str, Any]
) -> None:
    """Update namespace settings. Uses POST (MAPI convention)."""
    await self._client.request(
        "POST", f"/mapi/tenants/{tenant}/namespaces/{ns}", json=data
    )

async def delete_namespace(self, tenant: str, ns: str) -> None:
    """Delete a namespace."""
    await self._client.request("DELETE", f"/mapi/tenants/{tenant}/namespaces/{ns}")

async def export_namespace(self, tenant: str, ns: str) -> dict[str, Any]:
    """Export a namespace as a reusable template."""
    resp = await self._client.request(
        "GET", f"/mapi/tenants/{tenant}/namespaces/{ns}/export"
    )
    return resp.json()

async def export_namespaces(self, tenant: str, names: list[str]) -> dict[str, Any]:
    """Export multiple namespaces as templates."""
    resp = await self._client.request(
        "GET",
        f"/mapi/tenants/{tenant}/namespaces/export",
        params={"names": ",".join(names)},
    )
    return resp.json()