graflo.data_source.api_response

API response envelope parsing and optional first-response auto-detection.

Attributes

CARRY_TOKEN_CANDIDATES = ('results_id', 'scroll_id', 'pit_id', 'search_id') module-attribute

CURSOR_CANDIDATES = ('next_cursor', 'cursor', 'next_page_token') module-attribute

HAS_MORE_CANDIDATES = ('has_more', 'hasMore') module-attribute

NEXT_OFFSET_CANDIDATES = ('next_offset', 'nextOffset') module-attribute

OFFSET_CANDIDATES = ('offset', 'skip') module-attribute

RECORDS_CANDIDATES = ('results', 'data', 'items', 'records', 'entries', 'rows') module-attribute

TOTAL_COUNT_CANDIDATES = ('count', 'total', 'total_count') module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

APIBodyError

Bases: RuntimeError

A response the server answered successfully whose body reports a failure.

Attributes:

Name	Type	Description
code		The body's error code at error_path, or None if it has none.

Source code in graflo/data_source/api_response.py

class APIBodyError(RuntimeError):
    """A response the server answered successfully whose body reports a failure.

    Attributes:
        code: The body's error code at ``error_path``, or ``None`` if it has none.
    """

    def __init__(self, code: str | None, detail: str) -> None:
        super().__init__(
            f"API body reports failure ({code or 'no error code'}): {detail}"
        )
        self.code = code

Attributes

code = code instance-attribute

Methods:

__init__(code, detail)

Source code in graflo/data_source/api_response.py

def __init__(self, code: str | None, detail: str) -> None:
    super().__init__(
        f"API body reports failure ({code or 'no error code'}): {detail}"
    )
    self.code = code

ResolvedApiResponse

Bases: ApiResponseStructure

Response paths resolved after optional auto-detection on the first body.

Source code in graflo/data_source/api_response.py

class ResolvedApiResponse(ApiResponseStructure):
    """Response paths resolved after optional auto-detection on the first body."""

    @classmethod
    def resolve(
        cls,
        config: ApiResponseStructure,
        body: dict[str, Any] | list[Any],
    ) -> ResolvedApiResponse:
        detected: dict[str, str] = {}
        if config.auto_detect and isinstance(body, dict):
            detected = detect_response_shape(body)
            if detected:
                logger.info("Auto-detected API response paths: %s", detected)

        return cls(
            records_path=config.records_path or detected.get("records_path"),
            total_count_path=config.total_count_path
            or detected.get("total_count_path"),
            offset_path=config.offset_path or detected.get("offset_path"),
            next_offset_path=config.next_offset_path
            or detected.get("next_offset_path"),
            has_more_path=config.has_more_path or detected.get("has_more_path"),
            cursor_path=config.cursor_path or detected.get("cursor_path"),
            ok_path=config.ok_path,
            error_path=config.error_path,
            batch_metadata_paths=dict(config.batch_metadata_paths),
            auto_detect=False,
        )

Methods:

resolve(config, body) classmethod

Source code in graflo/data_source/api_response.py

@classmethod
def resolve(
    cls,
    config: ApiResponseStructure,
    body: dict[str, Any] | list[Any],
) -> ResolvedApiResponse:
    detected: dict[str, str] = {}
    if config.auto_detect and isinstance(body, dict):
        detected = detect_response_shape(body)
        if detected:
            logger.info("Auto-detected API response paths: %s", detected)

    return cls(
        records_path=config.records_path or detected.get("records_path"),
        total_count_path=config.total_count_path
        or detected.get("total_count_path"),
        offset_path=config.offset_path or detected.get("offset_path"),
        next_offset_path=config.next_offset_path
        or detected.get("next_offset_path"),
        has_more_path=config.has_more_path or detected.get("has_more_path"),
        cursor_path=config.cursor_path or detected.get("cursor_path"),
        ok_path=config.ok_path,
        error_path=config.error_path,
        batch_metadata_paths=dict(config.batch_metadata_paths),
        auto_detect=False,
    )

Functions:

check_body(body, resolved)

Raise :class:APIBodyError when the body says the request failed.

A body fails when ok_path is set and its value is not True, or when error_path is set and its value is non-empty.

Source code in graflo/data_source/api_response.py

def check_body(body: object, resolved: ApiResponseStructure) -> None:
    """Raise :class:`APIBodyError` when the body says the request failed.

    A body fails when ``ok_path`` is set and its value is not ``True``, or when
    ``error_path`` is set and its value is non-empty.
    """
    error = get_at_path(body, resolved.error_path)
    code = str(error) if error not in (None, "", False) else None
    if resolved.ok_path is not None:
        flag = get_at_path(body, resolved.ok_path)
        if flag is not True:
            raise APIBodyError(code, f"{resolved.ok_path}={flag!r}")
    if code is not None:
        raise APIBodyError(code, f"{resolved.error_path}={error!r}")

detect_carry_params(body)

Infer session-token carry map from known response field names.

Returns a query-param-name -> response-path map. Looks at top-level object keys first, then at 0.<key> when the body is a non-empty list of objects (e.g. a list-envelope API that returns a session token). Does not treat singular result_id as a carry token.

Source code in graflo/data_source/api_response.py

def detect_carry_params(body: dict[str, Any] | list[Any]) -> dict[str, str]:
    """Infer session-token carry map from known response field names.

    Returns a query-param-name -> response-path map. Looks at top-level object
    keys first, then at ``0.<key>`` when the body is a non-empty list of
    objects (e.g. a list-envelope API that returns a session token). Does not
    treat singular ``result_id`` as a carry token.
    """
    detected: dict[str, str] = {}
    for prefix, envelope in _carry_search_envelopes(body):
        for key in CARRY_TOKEN_CANDIDATES:
            if key in detected:
                continue
            if key in envelope and _is_carry_value(envelope[key]):
                detected[key] = f"{prefix}{key}"
    return detected

detect_response_shape(body)

Infer unset response path keys from a top-level response envelope.

Source code in graflo/data_source/api_response.py

def detect_response_shape(body: dict[str, Any]) -> dict[str, str]:
    """Infer unset response path keys from a top-level response envelope."""
    detected: dict[str, str] = {}

    records_path = _detect_records_path(body)
    if records_path is not None:
        detected["records_path"] = records_path

    for field, candidates in (
        ("next_offset_path", NEXT_OFFSET_CANDIDATES),
        ("total_count_path", TOTAL_COUNT_CANDIDATES),
        ("offset_path", OFFSET_CANDIDATES),
        ("has_more_path", HAS_MORE_CANDIDATES),
        ("cursor_path", CURSOR_CANDIDATES),
    ):
        key = _detect_top_level_key(body, candidates)
        if key is not None:
            detected[field] = key

    return detected

extract_records(body, resolved)

Extract record dicts from a parsed JSON body.

Source code in graflo/data_source/api_response.py

def extract_records(
    body: dict[str, Any] | list[Any],
    resolved: ResolvedApiResponse,
) -> list[dict[str, Any]]:
    """Extract record dicts from a parsed JSON body."""
    if resolved.records_path is None:
        if isinstance(body, list):
            return [_as_record(item) for item in body]
        raise ValueError(
            "API response is an object envelope but response.records_path is not "
            "configured. Set response.records_path or enable response.auto_detect."
        )

    data = get_at_path(body, resolved.records_path)
    if data is None:
        return []
    if isinstance(data, list):
        return [_as_record(item) for item in data]
    if isinstance(data, dict):
        return [cast(dict[str, Any], data)]
    return []

get_batch_metadata(body, resolved)

Read batch-level metadata from the response envelope.

Source code in graflo/data_source/api_response.py

def get_batch_metadata(
    body: dict[str, Any] | list[Any],
    resolved: ResolvedApiResponse,
) -> dict[str, Any]:
    """Read batch-level metadata from the response envelope."""
    if not resolved.batch_metadata_paths:
        return {}

    metadata: dict[str, Any] = {}
    for annotation_key, response_path in resolved.batch_metadata_paths.items():
        value = get_at_path(body, response_path)
        if value is not None:
            metadata[annotation_key] = value
    return metadata

has_more_pages(body, resolved, items, *, strategy)

Return whether another HTTP page should be fetched.

Source code in graflo/data_source/api_response.py

def has_more_pages(
    body: dict[str, Any] | list[Any],
    resolved: ResolvedApiResponse,
    items: list[dict[str, Any]],
    *,
    strategy: str,
) -> bool:
    """Return whether another HTTP page should be fetched."""
    if resolved.has_more_path is not None:
        return bool(get_at_path(body, resolved.has_more_path))

    if resolved.next_offset_path is not None:
        return get_at_path(body, resolved.next_offset_path) is not None

    if resolved.total_count_path is not None and resolved.offset_path is not None:
        count = get_at_path(body, resolved.total_count_path)
        offset = get_at_path(body, resolved.offset_path)
        if count is not None and offset is not None:
            return _as_int(offset) + len(items) < _as_int(count)

    if strategy == "cursor" and resolved.cursor_path is not None:
        cursor = get_at_path(body, resolved.cursor_path)
        return cursor is not None and str(cursor) != ""

    return len(items) > 0

next_cursor_value(body, resolved)

Read the next cursor token from the response when configured.

Source code in graflo/data_source/api_response.py

def next_cursor_value(
    body: dict[str, Any] | list[Any],
    resolved: ResolvedApiResponse,
) -> str | None:
    """Read the next cursor token from the response when configured."""
    if resolved.cursor_path is None:
        return None

    value = get_at_path(body, resolved.cursor_path)
    if value is None:
        return None
    token = str(value)
    return token if token else None

next_offset_value(body, resolved)

Read the next offset from the response when configured.

Source code in graflo/data_source/api_response.py

def next_offset_value(
    body: dict[str, Any] | list[Any],
    resolved: ResolvedApiResponse,
) -> int | None:
    """Read the next offset from the response when configured."""
    if resolved.next_offset_path is None:
        return None

    value = get_at_path(body, resolved.next_offset_path)
    if value is None:
        return None
    return _as_int(value)

update_carried_params(body, carry_params, carried)

Refresh carried query params from the latest response body.

Source code in graflo/data_source/api_response.py

def update_carried_params(
    body: dict[str, Any] | list[Any],
    carry_params: dict[str, str],
    carried: dict[str, Any],
) -> dict[str, Any]:
    """Refresh carried query params from the latest response body."""
    if not carry_params:
        return dict(carried)

    updated = dict(carried)
    for param, path in carry_params.items():
        value = get_at_path(body, path)
        if _is_carry_value(value):
            updated[param] = value
    return updated