Core Components¶

def analyze_url(
    self,
    url: str,
    *,
    method: HTTPMethod = HTTPMethod.GET,
    content: bytes | None = None,
    headers: Mapping[str, str] | None = None,
) -> list[StepMetrics]:
    """Analyze URL with optional redirect following.

    Performs HTTP request(s) and collects comprehensive metrics.
    If follow_redirects is enabled and server returns 3xx with Location,
    continues following redirects up to max_redirects.

    Args:
        url: Initial URL to analyze. Must be valid HTTP/HTTPS URL.
        method: HTTP method to use (GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS).
        content: Optional request body as bytes.
        headers: Optional mapping of request headers applied to every step.

    Returns:
        List of StepMetrics, one per request in the chain. Each step contains
        timing, network, and response information. Returns at least one step
        even if request fails. When a redirect points at a URL that cannot
        be requested (malformed authority, invalid port, missing host or a
        non-http(s) scheme), the redirect step is kept as is and the chain
        ends with an extra failed step for that target: it carries the
        target URL and a network-kind error but no timing or response, just
        like a hop whose connection failed.

    Examples:
        Basic usage without redirects:
            >>> analyzer = HTTPTapAnalyzer()
            >>> steps = analyzer.analyze_url("https://example.com")
            >>> print(f"Total time: {steps[0].timing.total_ms}ms")
            Total time: 234.5ms

        Following redirect chain:
            >>> analyzer = HTTPTapAnalyzer(follow_redirects=True)
            >>> steps = analyzer.analyze_url("http://example.com")
            >>> for i, step in enumerate(steps, 1):
            ...     print(f"Step {i}: {step.response.status}")
            Step 1: 301
            Step 2: 200

    """
    steps: list[StepMetrics] = []
    current_url = url
    redirect_count = 0
    deadline = time.monotonic() + self.timeout

    while True:
        step_number = len(steps) + 1
        step = self._analyze_single_request(
            current_url,
            step_number,
            deadline=deadline,
            method=method,
            content=content,
            headers=headers,
        )
        steps.append(step)
        # Only the redacted Location is displayed and exported; the real one
        # is followed so its userinfo can still authenticate.
        location = step.response.location
        if location:
            step.response.location = redact_url_credentials(location)

        # Check if we should follow redirect
        if not self.follow_redirects:
            break

        if step.has_error:
            # Stop on error
            break

        if step.is_redirect:
            # Follow redirect
            if location:
                if redirect_count == self.max_redirects:
                    step.note = f"{REDIRECT_LIMIT_NOTE} ({self.max_redirects})"
                    step.redirect_limit_reached = True
                    break
                next_method = _redirect_method(step.response.status or 0, method)
                next_url, target_error = _resolve_redirect_target(current_url, location)
                if target_error is not None:
                    steps.append(_invalid_redirect_step(next_url, step_number + 1, next_method, target_error))
                    break
                headers = _redirect_headers(
                    headers,
                    keep_credentials=_is_same_origin_or_https_upgrade(current_url, next_url),
                    keep_body_headers=next_method == method,
                )
                if next_method != method:
                    content = None
                method = next_method
                current_url = next_url
                redirect_count += 1
            else:
                # No Location header despite 3xx status
                break
        else:
            # Not a redirect, we're done
            break

    return steps

def to_dict(self) -> dict[str, Any]:
    """Convert step metrics to dictionary for JSON export.

    Returns:
        Dictionary containing all step information organized by category.

    """
    return {
        "url": self.url,
        "step_number": self.step_number,
        "request": {
            "method": self.request_method,
            "headers": self.request_headers,
            "body_bytes": self.request_body_bytes,
        },
        "timing": self.timing.to_dict(),
        "network": self.network.to_dict(),
        "response": self.response.to_dict(),
        "error": self.error,
        "note": self.note,
        "redirect_limit_reached": self.redirect_limit_reached,
        "proxy": self.proxied_via,
    }

def calculate_derived(self) -> None:
    """Calculate derived timing metrics.

    This method computes timing values that are derived from the raw
    measurements. It should be called after all raw timing values
    (dns_ms, connect_ms, tls_ms, ttfb_ms, total_ms) are populated.

    Computes:
        wait_ms: Time server spent processing the request before sending
            the first byte of the response. Calculated as:
            max(0, ttfb_ms - (dns_ms + connect_ms + tls_ms))

        xfer_ms: Time spent transferring the response body after headers
            were received. Calculated as:
            max(0, total_ms - ttfb_ms)

    Note:
        Both values are clamped to 0 to handle edge cases where timing
        measurements may have slight inconsistencies due to measurement
        precision or system clock adjustments.

    Examples:
        >>> timing = TimingMetrics(
        ...     dns_ms=10.0,
        ...     connect_ms=20.0,
        ...     tls_ms=50.0,
        ...     ttfb_ms=100.0,
        ...     total_ms=150.0
        ... )
        >>> timing.calculate_derived()
        >>> print(f"Wait: {timing.wait_ms}ms, Transfer: {timing.xfer_ms}ms")
        Wait: 20.0ms, Transfer: 50.0ms

    """
    self.wait_ms = max(
        0.0,
        self.ttfb_ms - (self.dns_ms + self.connect_ms + self.tls_ms),
    )
    self.xfer_ms = max(0.0, self.total_ms - self.ttfb_ms)

def to_dict(self) -> dict[str, float | bool]:
    """Convert timing metrics to dictionary.

    Returns:
        Dictionary with all timing metrics.

    """
    return {
        "dns_ms": self.dns_ms,
        "connect_ms": self.connect_ms,
        "tls_ms": self.tls_ms,
        "ttfb_ms": self.ttfb_ms,
        "total_ms": self.total_ms,
        "wait_ms": self.wait_ms,
        "xfer_ms": self.xfer_ms,
        "is_estimated": self.is_estimated,
    }

def to_dict(self) -> dict[str, Any]:
    """Convert network info to dictionary.

    Returns:
        Dictionary with all network information.

    """
    return {
        "ip": self.ip,
        "ip_family": self.ip_family,
        "http_version": self.http_version,
        "tls_version": self.tls_version,
        "tls_cipher": self.tls_cipher,
        "cert_cn": self.cert_cn,
        "cert_days_left": self.cert_days_left,
        "cert_sans": self.cert_sans,
        "cert_issuer": self.cert_issuer,
        "cert_serial": self.cert_serial,
        "cert_not_before": self.cert_not_before.isoformat() if self.cert_not_before else None,
        "cert_not_after": self.cert_not_after.isoformat() if self.cert_not_after else None,
        "tls_verified": self.tls_verified,
        "tls_custom_ca": self.tls_custom_ca,
        "proxy_url": self.proxy_url,
        "proxy_source": self.proxy_source,
    }

def to_dict(self) -> dict[str, Any]:
    """Convert response info to dictionary.

    Returns:
        Dictionary with all response information.

    """
    return {
        "status": self.status,
        "bytes": self.bytes,
        "content_type": self.content_type,
        "server": self.server,
        "date": self.date.isoformat() if self.date else None,
        "location": self.location,
        "headers": self.headers,
    }