fastflowtransform.api.http

get

get(url, *, params=None, headers=None, ttl=None, paginator=None, timeout=None)

Raw GET with optional FS cache and simple pagination. If paginator is provided, it should return {"next_request": {"url": "...", "params": {...}}} or None. When pagination is active the result is a list of response bodies; otherwise a single bytes object is returned.

Source code in src/fastflowtransform/api/http.py

def get(
    url: str,
    *,
    params: dict | None = None,
    headers: dict | None = None,
    ttl: int | None = None,
    paginator: Callable[[str, dict | None, Any], dict | None] | None = None,
    timeout: float | None = None,
) -> bytes | list[bytes]:
    """
    Raw GET with optional FS cache and simple pagination.
    If paginator is provided, it should return
    {"next_request": {"url": "...", "params": {...}}} or None.
    When pagination is active the result is a list of response bodies; otherwise
    a single bytes object is returned.
    """
    if not _domain_ok(url):
        raise RuntimeError(f"HTTP domain not allowed by FF_HTTP_ALLOWED_DOMAINS: {url}")

    ttl = _DEF_TTL if ttl is None else ttl
    headers = dict(headers or {})

    if paginator is None:
        body, _ = _request_with_cache("GET", url, params, headers, ttl, timeout)
        return body

    pages = _collect_pages(
        "GET",
        url,
        params,
        headers,
        ttl,
        timeout,
        paginator,
        keep_payload=False,
        payload_factory=_maybe_json_payload,
    )
    return [body for body, _, _ in pages]

get_json

get_json(url, *, params=None, headers=None, ttl=None, paginator=None, timeout=None)

GET returning parsed JSON. If paginator is provided, it follows pages via callback.

Source code in src/fastflowtransform/api/http.py

def get_json(
    url: str,
    *,
    params: dict | None = None,
    headers: dict | None = None,
    ttl: int | None = None,
    paginator: Callable[[str, dict | None, Any], dict | None] | None = None,
    timeout: float | None = None,
) -> Any:
    """GET returning parsed JSON. If paginator is provided, it follows pages via callback."""
    ttl = _DEF_TTL if ttl is None else ttl
    headers = dict(headers or {})
    pages = _collect_pages(
        "GET",
        url,
        params,
        headers,
        ttl,
        timeout,
        paginator,
        keep_payload=True,
        payload_factory=_json_payload,
    )
    payloads = [payload for _, _, payload in pages]
    return payloads[0] if paginator is None else payloads

get_df

get_df(url, *, params=None, headers=None, ttl=None, paginator=None, json_path=None, record_path=None, meta=None, dtype=None, timeout=None, normalize=False, output='pandas', session=None)

GET JSON and normalize into a DataFrame using pandas.json_normalize. If paginator is provided, concatenates pages over the same normalization logic.

Parameters

record_path : Sequence[str] | None Path to the list in the JSON to be normalized. meta : Sequence[str | Sequence[str]] | None Columns to include as metadata (top-level keys or nested paths). output : {"pandas","spark","bigframes"} Controls the returned frame type. "pandas" (default) yields a pandas DataFrame. "spark" materialises a pyspark.sql.DataFrame using the provided session (or an active/builder session). "bigframes" returns a BigFrames DataFrame (requires bigframes). session : Any | None Optional backend handle. For Spark, pass a SparkSession; otherwise the active session or a new one is used.

Source code in src/fastflowtransform/api/http.py

def get_df(
    url: str,
    *,
    params: dict | None = None,
    headers: dict | None = None,
    ttl: int | None = None,
    paginator: Callable[[str, dict | None, Any], dict | None] | None = None,
    json_path: list[str] | None = None,
    record_path: Sequence[str] | None = None,
    meta: MetaArgIn | None = None,
    dtype: dict[str, str] | None = None,
    timeout: float | None = None,
    normalize: bool = False,
    output: OutputBackend = "pandas",
    session: Any | None = None,
) -> Any:
    """
    GET JSON and normalize into a DataFrame using pandas.json_normalize.
    If `paginator` is provided, concatenates pages over the same normalization logic.

    Parameters
    ----------
    record_path : Sequence[str] | None
        Path to the list in the JSON to be normalized.
    meta : Sequence[str | Sequence[str]] | None
        Columns to include as metadata (top-level keys or nested paths).
    output : {"pandas","spark","bigframes"}
        Controls the returned frame type. "pandas" (default) yields a pandas DataFrame.
        "spark" materialises a pyspark.sql.DataFrame using the provided session
        (or an active/builder session).
        "bigframes" returns a BigFrames DataFrame (requires `bigframes`).
    session : Any | None
        Optional backend handle. For Spark, pass a SparkSession;
        otherwise the active session or a new one is used.
    """

    def _extract(obj: Any) -> Any:
        """Follow json_path (if provided) into nested JSON."""
        cur = obj
        for k in json_path or []:
            cur = cur.get(k) if isinstance(cur, dict) else None
        return cur

    def _coerce_meta(m: MetaArgIn) -> MetaParamOut:
        """
        Return a value whose static type is exactly:
            str | list[str | list[str]] | None
        """
        if m is None:
            return None
        # Build a list whose element type is (str | list[str])
        out: list[MetaEntry] = []
        for elem in m:
            if isinstance(elem, str):
                out.append(elem)  # str
            else:
                out.append(list(elem))  # Sequence[str] -> list[str]
        return out  # list[MetaEntry] == list[str | list[str]]

    def _to_df(js: Any) -> pd.DataFrame:
        base = _extract(js)
        base = base if base is not None else js
        if record_path:
            rp = list(record_path) if record_path else None
            meta_param = _coerce_meta(meta)
            df = pd.json_normalize(base, record_path=rp, meta=meta_param)
        # if it's a list of dicts
        elif isinstance(base, list):
            df = pd.json_normalize(base, sep=".") if normalize else pd.DataFrame(base)
        else:
            df = pd.json_normalize(base, sep=".") if normalize else pd.json_normalize(base)
        if dtype:
            # Use DataFrame.astype with a mapping to avoid Series.astype overload issues.
            try:
                df = df.astype(cast(Any, dict(dtype)), copy=False)
            except Exception:
                # Best-effort fallback, still via DataFrame.astype (no Series.astype)
                for col, dt in dtype.items():
                    with suppress(Exception):
                        df = df.astype({col: cast(Any, dt)}, copy=False)
        return df

    def _finalize(pdf: pd.DataFrame) -> Any:
        mode = (output or "pandas").lower()
        if mode == "pandas":
            return pdf
        if mode == "spark":
            try:
                from pyspark.sql import SparkSession  # noqa: PLC0415
            except Exception as exc:  # pragma: no cover - pyspark optional dependency
                raise RuntimeError(
                    "get_df(..., output='spark') requires pyspark to be installed."
                ) from exc
            spark = session
            if spark is None:
                spark = SparkSession.getActiveSession()
            if spark is None:
                spark = SparkSession.builder.getOrCreate()
            return spark.createDataFrame(pdf)
        if mode == "bigframes":
            try:
                import bigframes.pandas as bpd  # noqa: PLC0415
            except Exception as exc:  # pragma: no cover - bigframes optional dependency
                raise RuntimeError(
                    "get_df(..., output='bigframes') requires the 'bigframes' package."
                ) from exc
            return bpd.DataFrame(pdf)
        raise ValueError(
            f"Unsupported output backend '{output}' (expected pandas|spark|bigframes)."
        )

    if paginator is None:
        js = get_json(url, params=params, headers=headers, ttl=ttl, timeout=timeout)
        return _finalize(_to_df(js))

    pages = get_json(
        url, params=params, headers=headers, ttl=ttl, paginator=paginator, timeout=timeout
    )
    frames = []
    for js in pages if isinstance(pages, list) else [pages]:
        frames.append(_to_df(js))
    if not frames:
        return _finalize(pd.DataFrame())
    return _finalize(pd.concat(frames, ignore_index=True))