response

response

preferred response class for all API requests

RequestMetadata dataclass

RequestMetadata(
    url: str,
    headers: dict = dict(),
    body: str | None = None,
    params: dict | None = None,
)

to_dict

to_dict(auth_headers: list[str] | None = None) -> dict

returns dict representation of RequestMetadata

Source code in src/crew_dcs/client/response.py

def to_dict(self, auth_headers: list[str] | None = None) -> dict:
    """returns dict representation of RequestMetadata"""

    return {
        "url": self.url,
        "headers": {
            k: v for k, v in self.headers.items() if k not in (auth_headers or [])
        },
        "body": self.body,
        "params": self.params,
    }

ResponseGetData dataclass

ResponseGetData(
    status: int,
    response: dict[str, Any] | str | list[Any] | bytes,
    is_success: bool,
    request_metadata: RequestMetadata | None = None,
    additional_information: dict | None = None,
    response_headers: dict[str, str] | None = None,
    elapsed: float | None = None,
    raw_response: Response | None = None,
)

preferred response class for all API Requests

is_dry_run property

is_dry_run: bool

True when this response was SIMULATED, not sent.

A dry run returns status=200, is_success=True (client/get_data.py), so is_success alone cannot tell a simulated write apart from a real one — a caller checking only is_success would report a destructive operation as done when nothing was sent. Check this before believing a write landed.

from_httpx_response classmethod

from_httpx_response(
    res: Response,
    request_metadata: RequestMetadata | None = None,
    additional_information: dict | None = None,
    raw_response: Response | None = None,
) -> ResponseGetData

returns ResponseGetData from httpx.Response

Source code in src/crew_dcs/client/response.py

@classmethod
def from_httpx_response(
    cls,
    res: httpx.Response,
    request_metadata: RequestMetadata | None = None,
    additional_information: dict | None = None,
    raw_response: httpx.Response | None = None,
) -> "ResponseGetData":
    """returns ResponseGetData from httpx.Response"""

    # Capture response headers and timing
    response_headers = dict(res.headers) if res.headers else None
    # Try to get elapsed time, but handle case where response hasn't been read yet
    # (e.g., cached responses that are reconstructed)
    # Note: Accessing .elapsed on an unread response raises RuntimeError
    elapsed = None
    try:
        if hasattr(res, "elapsed") and res.elapsed:
            elapsed = res.elapsed.total_seconds()
    except RuntimeError:
        # Response hasn't been read yet (common with cached responses)
        elapsed = None

    # Check if response is successful
    ok = 200 <= res.status_code <= 399

    if ok:
        content_type = res.headers.get("Content-Type", "")

        # Try to parse as JSON if content type indicates it
        response = res.text  # Default to text
        if "application/json" in content_type:
            try:  # noqa: SIM105
                response = res.json()
            except ValueError:
                pass  # Keep as text if JSON parse fails

        stored_raw = (raw_response or res) if raw_response is not None else None
        return cls(
            status=res.status_code,
            response=response,
            is_success=True,
            additional_information=additional_information,
            request_metadata=request_metadata,
            response_headers=response_headers,
            elapsed=elapsed,
            raw_response=stored_raw,
        )

    # Error responses: prefer JSON body when available so callers can inspect API error payload
    content_type = res.headers.get("Content-Type", "")
    if "application/json" in content_type:
        try:
            response_payload = res.json()
        except ValueError:
            response_payload = (
                res.reason_phrase
                if hasattr(res, "reason_phrase")
                else res.text or "Unknown reason"
            )
    else:
        response_payload = (
            res.text
            if res.text
            else (
                res.reason_phrase
                if hasattr(res, "reason_phrase")
                else "Unknown reason"
            )
        )
    stored_raw = (raw_response or res) if raw_response is not None else None
    return cls(
        status=res.status_code,
        response=response_payload,
        is_success=False,
        request_metadata=request_metadata,
        additional_information=additional_information,
        response_headers=response_headers,
        elapsed=elapsed,
        raw_response=stored_raw,
    )

from_looper classmethod

from_looper(
    res: ResponseGetData, array: list
) -> ResponseGetData

Create ResponseGetData with array response (immutable).

Parameters:

Name	Type	Description	Default
res	ResponseGetData	Original ResponseGetData from last request	required
array	list	Complete aggregated array from looper	required

Returns:

Type	Description
ResponseGetData	New ResponseGetData instance with array response

Source code in src/crew_dcs/client/response.py

@classmethod
def from_looper(
    cls,
    res: "ResponseGetData",
    array: list,
) -> "ResponseGetData":
    """Create ResponseGetData with array response (immutable).

    Args:
        res: Original ResponseGetData from last request
        array: Complete aggregated array from looper

    Returns:
        New ResponseGetData instance with array response
    """
    if not res.is_success:
        return res

    # Create new instance instead of mutating
    return cls(
        status=res.status,
        response=array,  # New array response
        is_success=res.is_success,
        request_metadata=res.request_metadata,
        additional_information=res.additional_information,
        response_headers=res.response_headers,
        elapsed=res.elapsed,
        raw_response=res.raw_response,
    )

get_cache_headers

get_cache_headers() -> dict[str, str]

Extract cache-related headers from response.

Returns:

Type	Description
dict[str, str]	Dictionary of cache-related headers

Source code in src/crew_dcs/client/response.py

def get_cache_headers(self) -> dict[str, str]:
    """Extract cache-related headers from response.

    Returns:
        Dictionary of cache-related headers
    """
    if not self.response_headers:
        return {}

    cache_header_names = [
        "cache-control",
        "expires",
        "etag",
        "last-modified",
        "age",
        "date",
        "vary",
        "pragma",
    ]

    return {
        k: v
        for k, v in self.response_headers.items()
        if k.lower() in cache_header_names
    }

to_dict

to_dict(is_exclude_response: bool = True) -> dict

returns dict representation of ResponseGetData

Source code in src/crew_dcs/client/response.py

def to_dict(self, is_exclude_response: bool = True) -> dict:
    """returns dict representation of ResponseGetData"""
    return {
        "status": self.status,
        "response": None if is_exclude_response else self.response,
        "is_success": self.is_success,
        "request_metadata": (
            self.request_metadata.to_dict() if self.request_metadata else None
        ),
        "additional_information": self.additional_information,
        "response_headers": self.response_headers,
        "elapsed": self.elapsed,
        "raw_response": None,  # Don't serialize raw response
    }

find_ip

find_ip(html: str, html_tag: str = 'p') -> str | None

Extract IP address from HTML content.

Parameters:

Name	Type	Description	Default
html	str	HTML content to search	required
html_tag	str	HTML tag to search within (default: "p")	'p'

Returns:

Type	Description
str | None	IP address string if found, None otherwise

Source code in src/crew_dcs/client/response.py

def find_ip(html: str, html_tag: str = "p") -> str | None:
    """Extract IP address from HTML content.

    Args:
        html: HTML content to search
        html_tag: HTML tag to search within (default: "p")

    Returns:
        IP address string if found, None otherwise
    """
    ip_address_regex = r"(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})"
    soup = BeautifulSoup(html, "html.parser")

    tag = soup.find(html_tag)
    if not tag:
        return None

    matches = re.findall(ip_address_regex, str(tag))
    return matches[0] if matches else None