statuspro_client.py

"""
StatusProClient - The pythonic StatusPro API client with automatic resilience.

This client uses httpx's native transport layer to provide automatic retries,
rate limiting, error handling, and pagination for all API calls without any
decorators or wrapper methods needed.
"""

import contextlib
import json
import logging
import netrc
import os
from collections.abc import Awaitable, Callable
from http import HTTPStatus
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import parse_qs, quote, urlencode, urlparse, urlunparse

if TYPE_CHECKING:
    from .helpers.orders import Orders
    from .helpers.statuses import Statuses

import httpx
from dotenv import load_dotenv
from httpx import AsyncHTTPTransport
from httpx_retries import Retry, RetryTransport

from ._logging import Logger
from .api_wrapper import ApiNamespace
from .client import AuthenticatedClient

# Patterns used to identify sensitive query parameters and body fields in logs.
# Values matching these patterns are redacted to prevent information disclosure.
# See also: statuspro_mcp_server/src/statuspro_mcp/logging.py filter_sensitive_data()
# for the MCP equivalent.
_SENSITIVE_PARAMS: frozenset[str] = frozenset(
    {
        "api_key",
        "auth",
        "authorization",
        "credential",
        "email",
        "key",
        "password",
        "secret",
        "token",
    }
)

_REDACTED = "***"


def _is_sensitive(name: str) -> bool:
    """Check if a parameter/field name matches any sensitive pattern."""
    lower = name.lower()
    return any(pattern in lower for pattern in _SENSITIVE_PARAMS)


def _sanitize_url(url: str) -> str:
    """Redact sensitive query parameter values from a URL for safe logging."""
    try:
        parsed = urlparse(url)
        if not parsed.query:
            return url
        params = parse_qs(parsed.query, keep_blank_values=True)
        sanitized = {}
        for k, values in params.items():
            if _is_sensitive(k):
                sanitized[k] = [_REDACTED]
            else:
                sanitized[k] = values
        # Use urlencode with custom quote function that preserves * characters
        clean_query = urlencode(
            sanitized,
            doseq=True,
            quote_via=lambda s, safe="", encoding=None, errors=None: quote(
                s, safe=safe + "*", encoding=encoding, errors=errors
            ),
        )
        return urlunparse(parsed._replace(query=clean_query))
    except Exception:
        # If URL parsing fails, strip the query string entirely
        base, _, _ = url.partition("?")
        return f"{base}?{_REDACTED}"


def _sanitize_body(body: Any) -> Any:
    """Redact sensitive field values from nested dict/list bodies for safe logging."""

    def _sanitize_value(value: Any) -> Any:
        """Recursively sanitize nested structures."""
        if isinstance(value, dict):
            return {
                k: _REDACTED if _is_sensitive(k) else _sanitize_value(v)
                for k, v in value.items()
            }
        if isinstance(value, list):
            return [_sanitize_value(item) for item in value]
        return value

    if not isinstance(body, dict):
        return "[non-dict body]"
    return _sanitize_value(body)


class RateLimitAwareRetry(Retry):
    """
    Custom Retry class that allows non-idempotent methods (POST, PATCH) to be
    retried ONLY when receiving a 429 (Too Many Requests) status code.

    For all other retryable status codes (502, 503, 504), only idempotent methods
    (HEAD, GET, PUT, DELETE, OPTIONS, TRACE) will be retried.

    This ensures we don't accidentally retry non-idempotent operations after
    server errors, but we DO retry them when we're being rate-limited.
    """

    # Idempotent methods that are always safe to retry
    IDEMPOTENT_METHODS = frozenset(["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"])

    def __init__(self, *args: Any, **kwargs: Any):
        """Initialize and track the current request method."""
        super().__init__(*args, **kwargs)
        self._current_method: str | None = None

    def is_retryable_method(self, method: str) -> bool:
        """
        Allow all methods to pass through the initial check.

        Store the method for later use in is_retryable_status_code.
        """
        self._current_method = method.upper()
        # Accept all methods - we'll filter in is_retryable_status_code
        return self._current_method in self.allowed_methods

    def is_retryable_status_code(self, status_code: int) -> bool:
        """
        Check if a status code is retryable for the current method.

        For 429 (rate limiting), allow all methods.
        For other errors (502, 503, 504), only allow idempotent methods.
        """
        # First check if the status code is in the allowed list at all
        if status_code not in self.status_forcelist:
            return False

        # If we don't know the method, fall back to default behavior
        if self._current_method is None:
            return True

        # Rate limiting (429) - retry all methods
        if status_code == HTTPStatus.TOO_MANY_REQUESTS:
            return True

        # Other retryable errors - only retry idempotent methods
        return self._current_method in self.IDEMPOTENT_METHODS

    def increment(self) -> "RateLimitAwareRetry":
        """Return a new retry instance with the attempt count incremented."""
        # Call parent's increment which creates a new instance of our class
        new_retry = cast(RateLimitAwareRetry, super().increment())
        # Preserve the current method across retry attempts
        new_retry._current_method = self._current_method
        return new_retry


class ErrorLoggingTransport(AsyncHTTPTransport):
    """
    Transport layer that adds detailed error logging for 4xx client errors.

    This transport wraps another AsyncHTTPTransport and intercepts responses
    to log detailed error information using the generated error models.
    """

    def __init__(
        self,
        wrapped_transport: AsyncHTTPTransport | None = None,
        logger: Logger | None = None,
        **kwargs: Any,
    ):
        """
        Initialize the error logging transport.

        Args:
            wrapped_transport: The transport to wrap. If None, creates a new AsyncHTTPTransport.
            logger: Logger instance for capturing error details. If None, creates a default logger.
            **kwargs: Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.
        """
        super().__init__()
        if wrapped_transport is None:
            wrapped_transport = AsyncHTTPTransport(**kwargs)
        self._wrapped_transport = wrapped_transport
        self.logger: Logger = logger or logging.getLogger(__name__)

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        """Handle request and log detailed error information for 4xx responses."""
        response = await self._wrapped_transport.handle_async_request(request)

        # Log detailed information for 400-level client errors
        if 400 <= response.status_code < 500:
            await self._log_client_error(response, request)

        return response

    async def _log_client_error(
        self, response: httpx.Response, request: httpx.Request
    ) -> None:
        """Log 4xx client errors.

        StatusPro returns either ``ErrorResponse`` ({"message": str}) or
        ``ValidationErrorResponse`` ({"message": str, "errors": {field: [str]}}).
        We log the untyped JSON since both shapes are simple enough to render
        directly without a dedicated typed model.
        """
        method = request.method
        url = _sanitize_url(str(request.url))
        status_code = response.status_code

        request_body: Any = None
        if request.content:
            with contextlib.suppress(
                json.JSONDecodeError, UnicodeDecodeError, AttributeError, TypeError
            ):
                request_body = json.loads(request.content.decode("utf-8"))

        if hasattr(response, "aread"):
            with contextlib.suppress(TypeError, AttributeError):
                await response.aread()

        try:
            error_data = response.json()
        except (json.JSONDecodeError, TypeError, ValueError):
            self.logger.error(
                f"Client error {status_code} for {method} {url} - "
                f"Response: {getattr(response, 'text', '')[:500]}..."
            )
            return

        prefix = (
            f"Validation error 422 for {method} {url}"
            if status_code == 422
            else f"Client error {status_code} for {method} {url}"
        )

        if isinstance(error_data, dict):
            message = error_data.get("message") or "(not provided)"
            log_message = f"{prefix}\n  Error: {message}"

            errors = error_data.get("errors")
            if isinstance(errors, dict) and errors:
                log_message += f"\n  Validation errors ({len(errors)} fields):"
                for field, field_errors in errors.items():
                    sent_value = (
                        request_body.get(field)
                        if isinstance(request_body, dict)
                        else None
                    )
                    if sent_value is not None and _is_sensitive(str(field)):
                        sent_value = _REDACTED
                    log_message += f"\n    - {field}: {field_errors}"
                    if sent_value is not None:
                        log_message += f"\n      Sent: {sent_value!r}"
            self.logger.error(log_message)
        else:
            self.logger.error(f"{prefix}\n  Raw error: {_sanitize_body(error_data)}")


class PaginationTransport(AsyncHTTPTransport):
    """
    Transport layer that adds automatic pagination for GET requests.

    Auto-pagination behavior (for StatusPro's ``page``/``per_page`` scheme):
    - ON by default for GET requests with NO page parameter in URL
    - Uses 100 items per page (StatusPro's max) when no ``per_page`` is specified
    - If caller specifies ``per_page``, that value is respected
    - ANY explicit ``page`` parameter disables auto-pagination
    - Disabled when request has ``extensions={"auto_pagination": False}``
    - Only applies to GET requests
    - Only applies to wrapped list responses (``{"data": [...], "meta": {...}}``);
      raw-array responses like ``/statuses`` are passed through unchanged.

    Controlling pagination limits:
    - ``max_pages`` (constructor): Maximum number of pages to fetch
    - ``max_items`` (extension): Maximum total items to collect, e.g.,
      ``extensions={"max_items": 200}``
    """

    def __init__(
        self,
        wrapped_transport: AsyncHTTPTransport | None = None,
        max_pages: int = 100,
        logger: Logger | None = None,
        **kwargs: Any,
    ):
        """
        Initialize the pagination transport.

        Args:
            wrapped_transport: The transport to wrap. If None, creates a new AsyncHTTPTransport.
            max_pages: Maximum number of pages to collect during auto-pagination. Defaults to 100.
            logger: Logger instance for capturing pagination operations. If None, creates a default logger.
            **kwargs: Additional arguments passed to AsyncHTTPTransport if wrapped_transport is None.
        """
        # If no wrapped transport provided, create a base one
        if wrapped_transport is None:
            wrapped_transport = AsyncHTTPTransport(**kwargs)
            super().__init__()
        else:
            super().__init__()

        self._wrapped_transport = wrapped_transport
        self.max_pages = max_pages
        self.logger: Logger = logger or logging.getLogger(__name__)

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        """Handle request with automatic pagination for GET requests.

        Auto-pagination is ON by default for GET requests. It is disabled when:
        - `extensions={"auto_pagination": False}` is set, OR
        - ANY explicit `page` parameter is in the URL (e.g., `?page=1` or `?page=2`)

        To get auto-pagination, simply don't pass a page parameter. The transport
        will automatically use 100 items per page (StatusPro's max) unless you specify
        a limit, in which case your limit will be respected.
        """
        # Check if auto-pagination is explicitly disabled via request extensions
        auto_pagination = request.extensions.get("auto_pagination", True)

        # ANY page param in URL disables auto-pagination - caller wants specific page
        has_explicit_page = "page" in request.url.params

        # Only paginate GET requests when auto_pagination is enabled and no explicit page
        should_paginate = (
            request.method == "GET" and auto_pagination and not has_explicit_page
        )

        if should_paginate:
            return await self._handle_paginated_request(request)
        else:
            # For non-paginated requests, just pass through to wrapped transport
            return await self._wrapped_transport.handle_async_request(request)

    async def _handle_paginated_request(self, request: httpx.Request) -> httpx.Response:
        """
        Handle paginated requests by automatically collecting all pages.

        This method detects paginated responses and automatically collects all available
        pages up to the configured maximum. It preserves the original request structure
        while combining data from multiple pages.

        Args:
            request: The HTTP request to handle (must be a GET request).

        Returns:
            A combined HTTP response containing data from all collected pages with
            pagination metadata in the response body.

        Note:
            - Auto-pagination is ON by default for all GET requests
            - If response has no pagination info, returns the single response as-is
            - The response contains an 'auto_paginated' flag in the pagination metadata
            - Data from all pages is combined into a single 'data' array
            - Use `extensions={"max_items": N}` to limit total items collected
        """
        all_data: list[Any] = []
        current_page = 1
        total_pages: int | None = None
        page_num = 1
        response: httpx.Response | None = None
        original_is_raw_list = False

        # Get max_items limit from extensions (None = unlimited)
        max_items: int | None = request.extensions.get("max_items")

        # Parse initial parameters, preserving multi-value query params
        # (e.g., tags[]=a&tags[]=b). Using multi_items() instead of dict()
        # to avoid losing duplicate keys.
        base_params = [
            (k, v)
            for k, v in request.url.params.multi_items()
            if k not in ("page", "per_page")
        ]

        # Get caller's per_page or default to 100 (StatusPro's max)
        original_per_page = request.url.params.get("per_page")
        try:
            page_size = int(original_per_page) if original_per_page else 100
            if page_size <= 0:
                self.logger.warning(
                    "Invalid per_page parameter %r (must be positive), using 100",
                    original_per_page,
                )
                page_size = 100
            elif page_size > 100:
                self.logger.warning(
                    "per_page %d exceeds StatusPro's max of 100, clamping",
                    page_size,
                )
                page_size = 100
        except (ValueError, TypeError):
            self.logger.warning(
                "Invalid per_page parameter %r, using 100", original_per_page
            )
            page_size = 100

        self.logger.info("Auto-paginating request: %s", _sanitize_url(str(request.url)))

        for page_num in range(1, self.max_pages + 1):
            # Determine per_page for this request
            if max_items is not None:
                remaining = max_items - len(all_data)
                if remaining <= 0:
                    break
                current_per_page = str(min(page_size, remaining))
            else:
                current_per_page = str(page_size)

            # Build params with updated page/per_page, preserving multi-value params
            url_params = [
                *base_params,
                ("page", str(page_num)),
                ("per_page", current_per_page),
            ]

            # Create a new request with updated parameters
            paginated_request = httpx.Request(
                method=request.method,
                url=request.url.copy_with(params=url_params),
                headers=request.headers,
                content=request.content,
                extensions=request.extensions,
            )

            # Make the request using the wrapped transport
            response = await self._wrapped_transport.handle_async_request(
                paginated_request
            )

            if response.status_code != 200:
                # If we get an error, return the original response
                return response

            # Parse the response
            try:
                # Read the response content if it's streaming
                if hasattr(response, "aread"):
                    with contextlib.suppress(TypeError, AttributeError):
                        # Skip aread if it's not async (e.g., in tests with mocks)
                        await response.aread()

                data = response.json()

                # Track original response format on first page
                if page_num == 1:
                    original_is_raw_list = isinstance(data, list)

                # Extract pagination info from headers or response body
                pagination_info = self._extract_pagination_info(response, data)

                if pagination_info:
                    current_page = pagination_info.get("page", page_num)
                    total_pages = pagination_info.get("total_pages")

                    # Extract the actual data items
                    if isinstance(data, list):
                        items = data
                    else:
                        items = data.get("data", [])
                    all_data.extend(items)

                    # Check max_items limit
                    if max_items is not None and len(all_data) >= max_items:
                        all_data = all_data[:max_items]  # Truncate to exact limit
                        self.logger.info(
                            "Reached max_items limit (%d), stopping pagination",
                            max_items,
                        )
                        break

                    # Check if we're done
                    # Break if we've reached the last known page or got an empty page
                    if (total_pages and current_page >= total_pages) or len(items) == 0:
                        break

                    self.logger.debug(
                        "Collected page %s/%s, items: %d, total so far: %d",
                        current_page,
                        total_pages or "?",
                        len(items),
                        len(all_data),
                    )
                else:
                    # No pagination info - return response preserving its shape
                    self.logger.info(
                        "No pagination info found, returning single-page response"
                    )
                    # Apply max_items truncation if set
                    if max_items is not None:
                        if isinstance(data, list) and len(data) > max_items:
                            truncated = json.dumps(data[:max_items]).encode()
                            headers = dict(response.headers)
                            headers.pop("content-encoding", None)
                            headers.pop("content-length", None)
                            return httpx.Response(
                                status_code=200,
                                headers=headers,
                                content=truncated,
                                request=request,
                            )
                        if isinstance(data, dict) and "data" in data:
                            items = data["data"]
                            if isinstance(items, list) and len(items) > max_items:
                                data["data"] = items[:max_items]
                                truncated = json.dumps(data).encode()
                                headers = dict(response.headers)
                                headers.pop("content-encoding", None)
                                headers.pop("content-length", None)
                                return httpx.Response(
                                    status_code=200,
                                    headers=headers,
                                    content=truncated,
                                    request=request,
                                )
                    return response

            except (json.JSONDecodeError, KeyError) as e:
                self.logger.warning("Failed to parse paginated response: %s", e)
                return response

        # Ensure we have a response at this point
        if response is None:
            msg = "No response available after pagination"
            raise RuntimeError(msg)

        # Create a combined response, preserving the original response shape
        if original_is_raw_list:
            # Original endpoint returned a raw JSON list - preserve that format
            combined_content = json.dumps(all_data).encode()
        else:
            # StatusPro wrapped list shape: {"data": [...], "meta": {...}}.
            # Generated response models (e.g. OrderListResponse.from_dict)
            # require `meta`, so synthesize one describing the combined result
            # as a single page containing every collected item.
            total_items = len(all_data)
            combined_data: dict[str, Any] = {
                "data": all_data,
                "meta": {
                    "current_page": 1,
                    "last_page": 1,
                    "per_page": total_items,
                    "total": total_items,
                    "from": 1 if total_items else None,
                    "to": total_items if total_items else None,
                },
            }
            # Telemetry about the underlying pagination walk; lands in
            # additional_properties on the parsed model.
            if total_pages:
                combined_data["pagination"] = {
                    "total_pages": total_pages,
                    "collected_pages": page_num,
                    "total_items": total_items,
                    "auto_paginated": True,
                }
            combined_content = json.dumps(combined_data).encode()

        # Remove content-encoding headers to avoid compression issues
        headers = dict(response.headers)
        headers.pop("content-encoding", None)
        headers.pop("content-length", None)  # Will be recalculated

        combined_response = httpx.Response(
            status_code=200,
            headers=headers,
            content=combined_content,
            request=request,
        )

        self.logger.info(
            "Auto-pagination complete: collected %d items from %d pages",
            len(all_data),
            page_num,
        )

        return combined_response

    def _normalize_pagination_values(
        self, pagination_info: dict[str, Any]
    ) -> dict[str, Any]:
        """Convert pagination values from strings to appropriate Python types.

        JSON parsing may return numeric values as strings (e.g., "41" instead of 41).
        String comparison produces incorrect results: "5" >= "41" is True because
        "5" > "4" lexicographically. This method ensures all numeric pagination
        fields are proper integers for correct comparisons.

        Additionally, boolean fields like first_page and last_page may come as
        string values ("true"/"false") and are converted to Python booleans.

        Args:
            pagination_info: Dictionary containing pagination metadata.

        Returns:
            Dictionary with numeric fields converted to integers and boolean
            fields converted to booleans.
        """
        # Fields that should be integers for pagination comparisons
        numeric_fields = [
            "page",
            "total_pages",
            "total_items",
            "limit",
            "offset",
            "count",
            "per_page",
            "current_page",
            "total_records",
        ]

        # Fields that should be booleans (API returns "true"/"false" strings)
        boolean_fields = [
            "first_page",
            "last_page",
        ]

        result = pagination_info.copy()

        # Convert numeric fields
        for field in numeric_fields:
            if field in result:
                value = result[field]
                # Convert string numbers to integers
                if isinstance(value, str):
                    try:
                        result[field] = int(value)
                    except ValueError:
                        self.logger.warning(
                            "Invalid pagination value for %s: %r, removing field",
                            field,
                            value,
                        )
                        # Remove invalid field so fallback values are used
                        del result[field]
                # Already an int or float - ensure it's int
                elif isinstance(value, float):
                    # Warn if float has a fractional part (unexpected for pagination)
                    if value != int(value):
                        self.logger.warning(
                            "Pagination value %s has fractional part: %r, truncating to %d",
                            field,
                            value,
                            int(value),
                        )
                    result[field] = int(value)
                # If it's already an int, leave it as is

        # Convert boolean fields ("true"/"false" strings to Python booleans)
        for field in boolean_fields:
            if field in result:
                value = result[field]
                if isinstance(value, str):
                    lower_value = value.lower()
                    if lower_value == "true":
                        result[field] = True
                    elif lower_value == "false":
                        result[field] = False
                    else:
                        self.logger.warning(
                            "Invalid boolean pagination value for %s: %r, removing field",
                            field,
                            value,
                        )
                        del result[field]
                elif not isinstance(value, bool):
                    # Unexpected type - convert truthy/falsy to bool
                    result[field] = bool(value)

        return result

    def _extract_pagination_info(
        self, response: httpx.Response, data: dict[str, Any]
    ) -> dict[str, Any] | None:
        """Extract pagination information from response headers or body.

        Note:
            All numeric pagination values (page, total_pages, total_items, etc.)
            are converted to integers to ensure correct comparisons. This is important
            because JSON parsing may return string values, and string comparison
            (e.g., "5" >= "41") produces incorrect results.
        """
        pagination_info: dict[str, Any] = {}

        # Check for X-Pagination header (JSON format)
        if "X-Pagination" in response.headers:
            try:
                header_data = json.loads(response.headers["X-Pagination"])
                # Validate that parsed JSON is a dictionary
                if not isinstance(header_data, dict):
                    self.logger.warning(
                        "X-Pagination header is not a JSON object: %r", header_data
                    )
                else:
                    # Convert numeric string values to integers to avoid string comparison bugs
                    # (e.g., "5" >= "41" is True in string comparison but should be False)
                    pagination_info = self._normalize_pagination_values(header_data)
                    # Only return early if we got valid pagination data
                    if pagination_info:
                        return pagination_info
            except json.JSONDecodeError:
                pass

        # Check for individual headers (with validation for malformed values)
        if "X-Total-Pages" in response.headers:
            try:
                pagination_info["total_pages"] = int(response.headers["X-Total-Pages"])
            except ValueError:
                self.logger.warning(
                    "Invalid X-Total-Pages header value: %s",
                    response.headers["X-Total-Pages"],
                )
        if "X-Current-Page" in response.headers:
            try:
                pagination_info["page"] = int(response.headers["X-Current-Page"])
            except ValueError:
                self.logger.warning(
                    "Invalid X-Current-Page header value: %s",
                    response.headers["X-Current-Page"],
                )

        # Check for pagination in response body
        if isinstance(data, dict):
            if "pagination" in data and isinstance(data["pagination"], dict):
                pagination_info.update(
                    {str(k): v for k, v in data["pagination"].items()}
                )
            elif "meta" in data and isinstance(data["meta"], dict):
                meta = data["meta"]
                # StatusPro shape: meta has current_page, last_page, per_page, total
                if "current_page" in meta or "last_page" in meta:
                    pagination_info.update({str(k): v for k, v in meta.items()})
                    # Map StatusPro field names to the internal canonical names so
                    # the downstream stop-condition check in _handle_paginated_request
                    # continues to use `page` / `total_pages`.
                    if "current_page" in meta:
                        pagination_info["page"] = meta["current_page"]
                    if "last_page" in meta:
                        pagination_info["total_pages"] = meta["last_page"]
                # Alternate nested pagination shape: meta.pagination
                elif "pagination" in meta and isinstance(meta["pagination"], dict):
                    pagination_info.update(
                        {str(k): v for k, v in meta["pagination"].items()}
                    )

        # Normalize all numeric values to ensure correct comparisons
        if pagination_info:
            pagination_info = self._normalize_pagination_values(pagination_info)

        return pagination_info if pagination_info else None


def ResilientAsyncTransport(
    max_retries: int = 5,
    max_pages: int = 100,
    logger: Logger | None = None,
    **kwargs: Any,
) -> RetryTransport:
    """
    Factory function that creates a chained transport with error logging,
    pagination, and retry capabilities.

    This function chains multiple transport layers:
    1. AsyncHTTPTransport (base HTTP transport)
    2. ErrorLoggingTransport (logs detailed 4xx errors)
    3. PaginationTransport (auto-collects paginated responses)
    4. RetryTransport (handles retries with Retry-After header support)

    Args:
        max_retries: Maximum number of retry attempts for failed requests. Defaults to 5.
        max_pages: Maximum number of pages to collect during auto-pagination. Defaults to 100.
        logger: Logger instance for capturing operations. If None, creates a default logger.
        **kwargs: Additional arguments passed to the base AsyncHTTPTransport.
            Common parameters include:
            - http2 (bool): Enable HTTP/2 support
            - limits (httpx.Limits): Connection pool limits
            - verify (bool | str | ssl.SSLContext): SSL certificate verification
            - cert (str | tuple): Client-side certificates
            - trust_env (bool): Trust environment variables for proxy configuration

    Returns:
        A RetryTransport instance wrapping all the layered transports.

    Note:
        When using a custom transport, parameters like http2, limits, and verify
        must be passed to this factory function (which passes them to the base
        AsyncHTTPTransport), not to the httpx.Client/AsyncClient constructor.

    Example:
        ```python
        transport = ResilientAsyncTransport(max_retries=3, max_pages=50)
        async with httpx.AsyncClient(transport=transport) as client:
            response = await client.get("https://api.example.com/items")
        ```
    """
    resolved_logger: Logger = (
        logger if logger is not None else logging.getLogger(__name__)
    )

    # Build the transport chain from inside out:
    # 1. Base AsyncHTTPTransport
    base_transport = AsyncHTTPTransport(**kwargs)

    # 2. Wrap with error logging
    error_logging_transport = ErrorLoggingTransport(
        wrapped_transport=base_transport,
        logger=resolved_logger,
    )

    # 3. Wrap with pagination
    pagination_transport = PaginationTransport(
        wrapped_transport=error_logging_transport,
        max_pages=max_pages,
        logger=resolved_logger,
    )

    # Finally wrap with retry logic (outermost layer)
    # Use RateLimitAwareRetry which:
    # - Retries ALL methods (including POST/PATCH) for 429 rate limiting
    # - Retries ONLY idempotent methods for server errors (502, 503, 504)
    retry = RateLimitAwareRetry(
        total=max_retries,
        backoff_factor=1.0,  # Exponential backoff: 1, 2, 4, 8, 16 seconds
        respect_retry_after_header=True,  # Honor server's Retry-After header
        status_forcelist=[
            429,
            502,
            503,
            504,
        ],  # Status codes that should trigger retries
        allowed_methods=[
            "HEAD",
            "GET",
            "PUT",
            "DELETE",
            "OPTIONS",
            "TRACE",
            "POST",
            "PATCH",
        ],
    )
    retry_transport = RetryTransport(
        transport=pagination_transport,
        retry=retry,
    )

    return retry_transport


class StatusProClient(AuthenticatedClient):
    """The pythonic StatusPro API client with automatic resilience and pagination.

    Inherits from ``AuthenticatedClient`` and can be passed directly to
    generated API methods without a ``.client`` property.

    Features:
    - Automatic retries on network errors and server errors (5xx)
    - Automatic rate-limit handling (parses ``Retry-After``, falls back to
      exponential backoff on 429 since StatusPro doesn't emit the header)
    - Auto-pagination for wrapped list endpoints (``{"data": [...], "meta": {...}}``)
    - Uses 100 items per page (StatusPro's max) by default
    - Raw-array endpoints (``/statuses``, ``/orders/{id}/viable-statuses``) are passed through
    - Rich logging and observability

    Auto-pagination behavior:
    - ON by default for GET requests with no ``page`` parameter
    - ``per_page`` defaults to 100; caller values are respected (capped at 100)
    - ANY explicit ``page`` param disables auto-pagination
    - Disabled per-request via ``extensions={"auto_pagination": False}``
    - ``max_pages`` constructor argument caps total pages collected
    - ``extensions={"max_items": N}`` caps total items collected

    Usage:
        async with StatusProClient() as client:
            from statuspro_public_api_client.api.orders import list_orders

            response = await list_orders.asyncio_detailed(client=client)

            # One specific page (disables auto-pagination)
            response = await list_orders.asyncio_detailed(
                client=client, page=2, per_page=25
            )
    """

    @staticmethod
    def _read_from_netrc(base_url: str) -> str | None:
        """
        Read API key from ~/.netrc file.

        Args:
            base_url: The base URL to extract the hostname from.

        Returns:
            The API key (password field) from netrc, or None if not found.

        Note:
            The password field in netrc is used to store the API token since
            StatusPro API uses bearer token authentication, not HTTP Basic Auth.
        """
        try:
            # Extract hostname from base_url - handle both full URLs and bare hostnames
            parsed = urlparse(base_url)
            host: str | None = None

            if parsed.hostname:
                # URL with scheme (e.g., "https://app.orderstatuspro.com/api/v1")
                host = parsed.hostname
            else:
                # Try parsing as URL without scheme (e.g., "app.orderstatuspro.com/api/v1")
                parsed_with_scheme = urlparse(f"https://{base_url}")
                if parsed_with_scheme.hostname:
                    host = parsed_with_scheme.hostname
                else:
                    # Final fallback: treat as bare hostname (e.g., "api.example.com")
                    # Extract just the hostname part before any path
                    host = base_url.split("/")[0] if base_url else None

            # If we couldn't extract a valid hostname, return None
            if not host:
                return None

            netrc_path = Path.home() / ".netrc"
            if not netrc_path.exists():
                return None

            # Warn if .netrc is readable by group or others (POSIX only)
            if os.name != "nt":
                mode = netrc_path.stat().st_mode
                if mode & 0o077:
                    import warnings

                    warnings.warn(
                        f"~/.netrc has insecure permissions ({oct(mode & 0o777)}). "
                        "This may expose your API key. Run: chmod 600 ~/.netrc",
                        stacklevel=2,
                    )

            auth = netrc.netrc(str(netrc_path))
            authenticators = auth.authenticators(host)

            if authenticators:
                # Return password field (which contains our API token)
                # netrc returns (login, account, password)
                _login, _account, password = authenticators
                return password
        except (FileNotFoundError, netrc.NetrcParseError, OSError):
            # Silently ignore netrc errors - it's an optional source
            pass

        return None

    def __init__(
        self,
        api_key: str | None = None,
        base_url: str | None = None,
        timeout: float = 30.0,
        max_retries: int = 5,
        max_pages: int = 100,
        logger: Logger | None = None,
        **httpx_kwargs: Any,
    ):
        """
        Initialize the StatusPro API client with automatic resilience features.

        Args:
            api_key: StatusPro API key. If None, will try to load from STATUSPRO_API_KEY env var,
                .env file, or ~/.netrc file (in that order).
            base_url: Base URL for the StatusPro API. Defaults to https://app.orderstatuspro.com/api/v1
            timeout: Request timeout in seconds. Defaults to 30.0.
            max_retries: Maximum number of retry attempts for failed requests. Defaults to 5.
            max_pages: Maximum number of pages to collect during auto-pagination. Defaults to 100.
            logger: Any object whose debug/info/warning/error methods accept
                (msg, *args, **kwargs) — the standard logging.Logger call convention
                (e.g. logging.Logger, structlog.BoundLogger). If None, creates a
                default stdlib logger.
            **httpx_kwargs: Additional arguments passed to the base AsyncHTTPTransport.
                Common parameters include:
                - http2 (bool): Enable HTTP/2 support
                - limits (httpx.Limits): Connection pool limits
                - verify (bool | str | ssl.SSLContext): SSL certificate verification
                - cert (str | tuple): Client-side certificates
                - trust_env (bool): Trust environment variables for proxy configuration
                - event_hooks (dict): Custom event hooks (will be merged with built-in hooks)

        Raises:
            ValueError: If no API key is provided via api_key param, STATUSPRO_API_KEY env var,
                .env file, or ~/.netrc file.

        Note:
            Transport-related parameters (http2, limits, verify, etc.) are correctly
            passed to the innermost AsyncHTTPTransport layer, ensuring they take effect
            even with the layered transport architecture.

        Example:
            >>> async with StatusProClient() as client:
            ...     # All API calls through client get automatic resilience
            ...     response = await some_api_method.asyncio_detailed(client=client)
        """
        load_dotenv()

        # Handle backwards compatibility: accept 'token' kwarg as alias for 'api_key'
        if "token" in httpx_kwargs:
            if api_key is not None:
                raise ValueError("Cannot specify both 'api_key' and 'token' parameters")
            api_key = httpx_kwargs.pop("token")

        # Determine base_url early so we can use it for netrc lookup
        base_url = (
            base_url
            or os.getenv("STATUSPRO_BASE_URL")
            or "https://app.orderstatuspro.com/api/v1"
        )

        # Setup credentials with priority: param > env (including .env) > netrc
        api_key = (
            api_key or os.getenv("STATUSPRO_API_KEY") or self._read_from_netrc(base_url)
        )

        if not api_key:
            raise ValueError(
                "API key required via: api_key param, STATUSPRO_API_KEY env var, "
                ".env file, or ~/.netrc"
            )

        self.logger: Logger = logger or logging.getLogger(__name__)
        self.max_pages = max_pages

        # Warn if SSL verification is disabled — risk of MITM attacks
        if httpx_kwargs.get("verify") is False:
            self.logger.warning(
                "SSL certificate verification is disabled (verify=False). "
                "This exposes the connection to MITM attacks. "
                "Only use this for local development."
            )

        # Domain helper instances (lazy-loaded via properties)
        self._orders: Orders | None = None
        self._statuses: Statuses | None = None
        self._api_namespace: ApiNamespace | None = None

        # Extract client-level parameters that shouldn't go to the transport
        # Event hooks for observability - start with our defaults
        event_hooks: dict[str, list[Callable[[httpx.Response], Awaitable[None]]]] = {
            "response": [
                self._capture_pagination_metadata,
                self._log_response_metrics,
            ]
        }

        # Extract and merge user hooks
        user_hooks = httpx_kwargs.pop("event_hooks", {})
        for event, hooks in user_hooks.items():
            # Normalize to list and add to existing or create new event
            hook_list = cast(
                list[Callable[[httpx.Response], Awaitable[None]]],
                hooks if isinstance(hooks, list) else [hooks],
            )
            if event in event_hooks:
                event_hooks[event].extend(hook_list)
            else:
                event_hooks[event] = hook_list

        # Check if user wants to override the transport entirely
        custom_transport = httpx_kwargs.pop("transport", None) or httpx_kwargs.pop(
            "async_transport", None
        )

        if custom_transport:
            # User provided a custom transport, use it as-is
            transport = custom_transport
        else:
            # Separate transport-specific kwargs from client-specific kwargs
            # Client-specific params that should NOT go to the transport
            client_only_params = ["headers", "cookies", "params", "auth"]
            client_kwargs = {
                k: httpx_kwargs.pop(k)
                for k in list(httpx_kwargs.keys())
                if k in client_only_params
            }

            # Create resilient transport with remaining transport-specific httpx_kwargs
            # These will be passed to the base AsyncHTTPTransport (http2, limits, verify, etc.)
            transport = ResilientAsyncTransport(
                max_retries=max_retries,
                max_pages=max_pages,
                logger=self.logger,
                **httpx_kwargs,  # Pass through http2, limits, verify, cert, trust_env, etc.
            )

            # Put client-specific params back into httpx_kwargs for the parent class
            httpx_kwargs.update(client_kwargs)

        # Initialize the parent AuthenticatedClient
        super().__init__(
            base_url=base_url,
            token=api_key,
            timeout=httpx.Timeout(timeout),
            httpx_args={
                "transport": transport,
                "event_hooks": event_hooks,
                **httpx_kwargs,  # Include any remaining client-level kwargs
            },
        )

    # Remove the client property since we inherit from AuthenticatedClient
    # Users can now pass the StatusProClient instance directly to API methods

    # Domain properties for ergonomic access
    @property
    def orders(self) -> "Orders":
        """Access order operations (list, lookup, get, update status, etc.)."""
        from .helpers.orders import Orders

        if self._orders is None:
            self._orders = Orders(self)
        return self._orders

    @property
    def statuses(self) -> "Statuses":
        """Access status catalog operations."""
        from .helpers.statuses import Statuses

        if self._statuses is None:
            self._statuses = Statuses(self)
        return self._statuses

    @property
    def api(self) -> ApiNamespace:
        """Thin CRUD wrappers for all API resources. Returns raw attrs models."""
        if self._api_namespace is None:
            self._api_namespace = ApiNamespace(self)
        return self._api_namespace

    # Event hooks for observability
    async def _capture_pagination_metadata(self, response: httpx.Response) -> None:
        """Capture and store pagination metadata from response headers."""
        if response.status_code == 200:
            x_pagination = response.headers.get("X-Pagination")
            if x_pagination:
                try:
                    pagination_info = json.loads(x_pagination)
                    self.logger.debug(f"Pagination metadata: {pagination_info}")
                    # Store pagination info for easy access
                    setattr(response, "pagination_info", pagination_info)  # noqa: B010
                except json.JSONDecodeError:
                    self.logger.warning(f"Invalid X-Pagination header: {x_pagination}")

    async def _log_response_metrics(self, response: httpx.Response) -> None:
        """Log response metrics for observability."""
        # Extract timing info if available (after response is read)
        try:
            if hasattr(response, "elapsed"):
                duration = response.elapsed.total_seconds()
            else:
                duration = 0.0
        except RuntimeError:
            # elapsed not available yet
            duration = 0.0

        self.logger.debug(
            f"Response: {response.status_code} {response.request.method} "
            f"{_sanitize_url(str(response.request.url))} ({duration:.2f}s)"
        )