Skip to content

Response

_response

Response parsing for PIP API responses (Arrow, JSON, CSV formats).

PIPResponse(url, status, content_type, content, response) dataclass

Full API response object returned when simplify=False.

Equivalent to pipr's pip_api S3 class.

Attributes:

Name Type Description
url str

The request URL as a string.

status int

HTTP status code.

content_type str

Content-Type header value.

content DataFrame

Parsed data as a pandas (or polars) DataFrame.

response Response

The underlying :class:httpx.Response.

parse_response(response, simplify=True, dataframe_type='pandas', is_raw=False)

Parse an HTTP response from the PIP API.

Dispatches to the appropriate format parser based on the Content-Type response header.

Parameters:

Name Type Description Default
response Response

Completed :class:httpx.Response.

required
simplify bool

If True (default), return a DataFrame directly. If False, return a :class:PIPResponse wrapper.

True
dataframe_type Literal['pandas', 'polars']

"pandas" (default) or "polars".

'pandas'
is_raw bool

If True, skip DataFrame conversion (for health-check, pip-info endpoints that return plain dicts/lists).

False

Returns:

Name Type Description
DataFrame | PIPResponse | dict | list

A DataFrame when simplify is True and is_raw is False.

A DataFrame | PIPResponse | dict | list

class:PIPResponse when simplify is False.

DataFrame | PIPResponse | dict | list

A dict or list when is_raw is True.

Raises:

Type Description
PIPError

If the response Content-Type is not supported.

Source code in src/povineq/_response.py
def parse_response(
    response: httpx.Response,
    simplify: bool = True,
    dataframe_type: Literal["pandas", "polars"] = "pandas",
    is_raw: bool = False,
) -> pd.DataFrame | PIPResponse | dict | list:
    """Parse an HTTP response from the PIP API.

    Dispatches to the appropriate format parser based on the
    ``Content-Type`` response header.

    Args:
        response: Completed :class:`httpx.Response`.
        simplify: If ``True`` (default), return a DataFrame directly.
            If ``False``, return a :class:`PIPResponse` wrapper.
        dataframe_type: ``"pandas"`` (default) or ``"polars"``.
        is_raw: If ``True``, skip DataFrame conversion (for health-check,
            pip-info endpoints that return plain dicts/lists).

    Returns:
        A DataFrame when *simplify* is ``True`` and *is_raw* is ``False``.
        A :class:`PIPResponse` when *simplify* is ``False``.
        A dict or list when *is_raw* is ``True``.

    Raises:
        PIPError: If the response Content-Type is not supported.
    """
    content_type = response.headers.get("content-type", "")

    if "application/vnd.apache.arrow.file" in content_type:
        table: pa.Table = _parse_arrow(response.content)
        # Arrow responses do not need decile pivoting (see pipr comment).
        # Rename columns directly on the Arrow Table (zero-copy) before
        # converting to the target type, avoiding a double numpy pass.
        if simplify and not is_raw:
            renamed_names = [COLUMN_RENAMES.get(name, name) for name in table.schema.names]
            table = table.rename_columns(renamed_names)
            return _to_target_type(table, dataframe_type)
        parsed: pd.DataFrame | dict | list = table.to_pandas()
        if isinstance(parsed, pd.DataFrame):
            parsed = rename_cols(
                parsed, list(COLUMN_RENAMES.keys()), list(COLUMN_RENAMES.values())
            )

    elif "application/json" in content_type:
        parsed = _parse_json(response.text, is_raw=is_raw)
        if is_raw:
            return parsed
        if simplify and isinstance(parsed, pd.DataFrame):
            parsed = _apply_post_processing(parsed)
            return _to_target_type(parsed, dataframe_type)

    elif "text/csv" in content_type:
        parsed = _parse_csv(response.text)
        if simplify and isinstance(parsed, pd.DataFrame):
            parsed = _apply_post_processing(parsed)
            return _to_target_type(parsed, dataframe_type)

    else:
        raise PIPError(f"Unsupported Content-Type: {content_type!r}")

    # simplify=False path
    if isinstance(parsed, pd.DataFrame):
        parsed = _apply_post_processing(parsed)
        parsed = _to_target_type(parsed, dataframe_type)

    return PIPResponse(
        url=str(response.url),
        status=response.status_code,
        content_type=content_type,
        content=parsed,  # type: ignore[arg-type]
        response=response,
    )