API reference

polars-api registers an api namespace on every Polars expression. Import the package once and the namespace becomes available on any expression that resolves to a URL string.

import polars as pl
import polars_api  # noqa: F401  — registers the `.api` namespace

Methods

Method	HTTP verb	Mode
get	GET	sync
aget	GET	async
post	POST	sync
apost	POST	async

All methods return a pl.Expr of dtype Utf8 containing the response body for each row. Use .str.json_decode() to parse JSON responses.

polars_api.Api

polars_api.api.Api

Source code in polars_api/api.py

@pl.api.register_expr_namespace("api")
class Api:
    def __init__(self, url: pl.Expr) -> None:
        self._url = url

    @staticmethod
    def _send_sync(
        client: httpx.Client,
        method: str,
        url: str,
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
        on_request: Optional[RequestHook],
        on_response: Optional[ResponseHook],
    ) -> httpx.Response:
        kwargs = _build_request_kwargs(params, body, data, headers, timeout)
        request = client.build_request(method, url, **kwargs)
        if on_request is not None:
            on_request(request)
        response = client.send(request)
        if on_response is not None:
            on_response(response)
        return response

    @staticmethod
    def _build_aio_kwargs(
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
    ) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}
        if params is not None:
            # aiohttp's params accept str/int/float values; coerce non-strings
            # so behaviour matches httpx.
            kwargs["params"] = {k: v if isinstance(v, str) else str(v) for k, v in params.items()}
        if headers is not None:
            kwargs["headers"] = headers
        if body is not None:
            kwargs["json"] = body
        if data is not None:
            kwargs["data"] = data
        if timeout is not None:
            kwargs["timeout"] = aiohttp.ClientTimeout(total=timeout)
        return kwargs

    @classmethod
    def _sync_one(
        cls,
        client: httpx.Client,
        method: str,
        url: str,
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        with_response_headers: bool,
        on_request: Optional[RequestHook],
        on_response: Optional[ResponseHook],
    ) -> dict[str, Any]:
        attempt = 0
        start = time.monotonic()
        while True:
            attempt_start = time.monotonic()
            try:
                response = cls._send_sync(
                    client,
                    method,
                    url,
                    params,
                    body,
                    data,
                    headers,
                    timeout,
                    on_request,
                    on_response,
                )
            except httpx.HTTPError as exc:
                if attempt < retries:
                    wait = backoff * (2**attempt) if backoff > 0 else 0.0
                    if wait > 0:
                        time.sleep(wait)
                    attempt += 1
                    continue
                elapsed_ms = (time.monotonic() - attempt_start) * 1000
                return _result_struct(
                    None,
                    0,
                    elapsed_ms,
                    f"{type(exc).__name__}: {exc}",
                    include_response_headers=with_response_headers,
                )

            status = response.status_code
            text = response.text
            resp_headers = _serialize_response_headers(response) if with_response_headers else None
            if response.is_success:
                elapsed_ms = (time.monotonic() - start) * 1000
                return _result_struct(
                    text,
                    status,
                    elapsed_ms,
                    None,
                    resp_headers,
                    include_response_headers=with_response_headers,
                )
            if attempt < retries and _is_retryable_status(status):
                wait = _retry_after_seconds(response)
                if wait is None:
                    wait = backoff * (2**attempt) if backoff > 0 else 0.0
                if wait > 0:
                    time.sleep(wait)
                attempt += 1
                continue
            elapsed_ms = (time.monotonic() - start) * 1000
            return _result_struct(
                text,
                status,
                elapsed_ms,
                f"HTTP {status}",
                resp_headers,
                include_response_headers=with_response_headers,
            )

    @classmethod
    async def _async_attempt(
        cls,
        session: aiohttp.ClientSession,
        method: str,
        url: str,
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
        attempt: int,
        retries: int,
        backoff: float,
        start: float,
        with_response_headers: bool,
        on_request: Optional[AsyncRequestHook],
        on_response: Optional[AsyncResponseHook],
    ) -> tuple[Optional[dict[str, Any]], Optional[float]]:
        attempt_start = time.monotonic()
        kwargs = cls._build_aio_kwargs(params, body, data, headers, timeout)
        try:
            if on_request is not None:
                on_request(method, url, kwargs)
            async with session.request(method, url, **kwargs) as response:
                if on_response is not None:
                    on_response(response)
                status = response.status
                text = await response.text()
                resp_headers = _serialize_aio_headers(response) if with_response_headers else None
                retry_after = _retry_after_seconds_aio(response)
        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
            if attempt < retries:
                wait = backoff * (2**attempt) if backoff > 0 else 0.0
                return None, wait
            elapsed_ms = (time.monotonic() - attempt_start) * 1000
            return _result_struct(
                None,
                0,
                elapsed_ms,
                f"{type(exc).__name__}: {exc}",
                include_response_headers=with_response_headers,
            ), None

        if 200 <= status < 300:
            elapsed_ms = (time.monotonic() - start) * 1000
            return _result_struct(
                text,
                status,
                elapsed_ms,
                None,
                resp_headers,
                include_response_headers=with_response_headers,
            ), None
        if attempt < retries and _is_retryable_status(status):
            wait = retry_after if retry_after is not None else (backoff * (2**attempt) if backoff > 0 else 0.0)
            return None, wait
        elapsed_ms = (time.monotonic() - start) * 1000
        return _result_struct(
            text,
            status,
            elapsed_ms,
            f"HTTP {status}",
            resp_headers,
            include_response_headers=with_response_headers,
        ), None

    @classmethod
    async def _async_one(
        cls,
        session: aiohttp.ClientSession,
        semaphore: Optional[asyncio.Semaphore],
        method: str,
        url: str,
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        with_response_headers: bool,
        on_request: Optional[AsyncRequestHook],
        on_response: Optional[AsyncResponseHook],
    ) -> dict[str, Any]:
        async def _go() -> dict[str, Any]:
            attempt = 0
            start = time.monotonic()
            while True:
                result, wait = await cls._async_attempt(
                    session,
                    method,
                    url,
                    params,
                    body,
                    data,
                    headers,
                    timeout,
                    attempt,
                    retries,
                    backoff,
                    start,
                    with_response_headers,
                    on_request,
                    on_response,
                )
                if result is not None:
                    return result
                if wait and wait > 0:
                    await asyncio.sleep(wait)
                attempt += 1

        if semaphore is None:
            return await _go()
        async with semaphore:
            return await _go()

    @classmethod
    def _send_sync_with_retries(
        cls,
        client: httpx.Client,
        method: str,
        url: str,
        params: Optional[dict[str, Any]],
        body: Optional[dict[str, Any]],
        data: Optional[dict[str, Any]],
        headers: Optional[dict[str, Any]],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        on_request: Optional[RequestHook],
        on_response: Optional[ResponseHook],
    ) -> Optional[httpx.Response]:
        """Like ``_send_sync`` but retries on failure. Returns the last response, or None
        if every attempt failed with a network error."""
        attempt = 0
        while True:
            try:
                response = cls._send_sync(
                    client,
                    method,
                    url,
                    params,
                    body,
                    data,
                    headers,
                    timeout,
                    on_request,
                    on_response,
                )
            except httpx.HTTPError:
                if attempt < retries:
                    wait = backoff * (2**attempt) if backoff > 0 else 0.0
                    if wait > 0:
                        time.sleep(wait)
                    attempt += 1
                    continue
                return None
            if response.is_success:
                return response
            if attempt < retries and _is_retryable_status(response.status_code):
                wait = _retry_after_seconds(response)
                if wait is None:
                    wait = backoff * (2**attempt) if backoff > 0 else 0.0
                if wait > 0:
                    time.sleep(wait)
                attempt += 1
                continue
            return None

    @classmethod
    def _sync_batch(
        cls,
        method: str,
        rows: list[tuple[str, Any, Any, Any, Any]],
        *,
        client: Optional[httpx.Client],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        cache: bool,
        with_response_headers: bool,
        on_request: Optional[RequestHook],
        on_response: Optional[ResponseHook],
    ) -> list[dict[str, Any]]:
        own_client = client is None
        cli: httpx.Client = httpx.Client() if own_client else client  # type: ignore[assignment]
        results: list[Optional[dict[str, Any]]] = [None] * len(rows)
        memo: dict[Any, dict[str, Any]] = {}
        try:
            for i, (url, params, body, data, headers) in enumerate(rows):
                key = None
                if cache:
                    key = (method, url, _hashable(params), _hashable(body), _hashable(data), _hashable(headers))
                    if key in memo:
                        results[i] = memo[key]
                        continue
                result = cls._sync_one(
                    cli,
                    method,
                    url,
                    params,
                    body,
                    data,
                    headers,
                    timeout,
                    retries,
                    backoff,
                    with_response_headers,
                    on_request,
                    on_response,
                )
                if cache and key is not None:
                    memo[key] = result
                results[i] = result
        finally:
            if own_client:
                cli.close()
        return [r for r in results if r is not None]

    @classmethod
    async def _async_many(
        cls,
        method: str,
        rows: list[tuple[str, Any, Any, Any, Any]],
        *,
        client: Optional[aiohttp.ClientSession],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        max_concurrency: Optional[int],
        cache: bool,
        with_response_headers: bool,
        on_request: Optional[AsyncRequestHook],
        on_response: Optional[AsyncResponseHook],
    ) -> list[dict[str, Any]]:
        semaphore = asyncio.Semaphore(max_concurrency) if max_concurrency else None

        # Dedupe identical rows up-front so we only fire one task per unique key.
        unique_indices: dict[Any, int] = {}
        order: list[int] = []  # index into rows for each unique task
        result_index: list[int] = [0] * len(rows)  # row -> position in tasks list
        for i, (url, params, body, data, headers) in enumerate(rows):
            if cache:
                key = (method, url, _hashable(params), _hashable(body), _hashable(data), _hashable(headers))
                if key in unique_indices:
                    result_index[i] = unique_indices[key]
                    continue
                unique_indices[key] = len(order)
                result_index[i] = len(order)
                order.append(i)
            else:
                result_index[i] = len(order)
                order.append(i)

        own_session = client is None
        if own_session:
            # Size the connection pool to the requested concurrency so we don't
            # bottleneck on aiohttp's default pool limit (100). 0 == unlimited.
            connector = aiohttp.TCPConnector(limit=max_concurrency or 0)
            sess = aiohttp.ClientSession(connector=connector)
        else:
            sess = client  # type: ignore[assignment]
        try:
            tasks = [
                cls._async_one(
                    sess,
                    semaphore,
                    method,
                    rows[idx][0],
                    rows[idx][1],
                    rows[idx][2],
                    rows[idx][3],
                    rows[idx][4],
                    timeout,
                    retries,
                    backoff,
                    with_response_headers,
                    on_request,
                    on_response,
                )
                for idx in order
            ]
            unique_results = await asyncio.gather(*tasks)
        finally:
            if own_session:
                await sess.close()
        return [unique_results[result_index[i]] for i in range(len(rows))]

    @staticmethod
    def _results_to_series(
        results: list[dict[str, Any]],
        *,
        with_metadata: bool,
        with_response_headers: bool,
        on_error: OnError,
    ) -> pl.Series:
        if with_metadata:
            return pl.Series(results, dtype=_metadata_dtype(with_response_headers))
        out: list[Optional[str]] = []
        for r in results:
            if r["error"] is None:
                out.append(_coerce_body(r["body"]))
            elif on_error == "raise":
                raise RuntimeError(r["error"])
            elif on_error == "return":
                out.append(_coerce_body(r["body"]))
            else:
                out.append(None)
        return pl.Series(out, dtype=pl.Utf8)

    @staticmethod
    def _rows_from_struct(s: pl.Series) -> list[tuple[str, Any, Any, Any, Any]]:
        urls = s.struct.field("url").to_list()
        params = s.struct.field("params").to_list()
        bodies = s.struct.field("body").to_list()
        data = s.struct.field("data").to_list()
        headers = s.struct.field("headers").to_list()
        return list(zip(urls, params, bodies, data, headers))

    def _build_headers_expr(
        self,
        headers: Optional[pl.Expr],
        auth: Optional[tuple[str, str]],
        bearer: Optional[Union[str, pl.Expr]],
        api_key: Optional[Union[str, pl.Expr]],
        api_key_header: str,
    ) -> Optional[pl.Expr]:
        extras: dict[str, pl.Expr] = {}
        if auth is not None:
            user, password = auth
            extras["Authorization"] = pl.lit(_basic_auth_header(user, password))
        if bearer is not None:
            bearer_expr = bearer if isinstance(bearer, pl.Expr) else pl.lit(bearer)
            extras["Authorization"] = pl.lit("Bearer ") + bearer_expr.cast(pl.Utf8)
        if api_key is not None:
            api_key_expr = api_key if isinstance(api_key, pl.Expr) else pl.lit(api_key)
            extras[api_key_header] = api_key_expr.cast(pl.Utf8)

        if not extras and headers is None:
            return None
        if not extras:
            return headers
        extra_exprs = [expr.alias(name) for name, expr in extras.items()]
        if headers is None:
            return pl.struct(*extra_exprs)
        return headers.struct.with_fields(*extra_exprs)

    def _input_struct(
        self,
        params: Optional[pl.Expr],
        body: Optional[pl.Expr],
        data: Optional[pl.Expr],
        headers: Optional[pl.Expr],
    ) -> pl.Expr:
        return pl.struct(
            self._url.alias("url"),
            (pl.lit(None) if params is None else params).alias("params"),
            (pl.lit(None) if body is None else body).alias("body"),
            (pl.lit(None) if data is None else data).alias("data"),
            (pl.lit(None) if headers is None else headers).alias("headers"),
        )

    def _sync_call(
        self,
        method: str,
        params: Optional[pl.Expr],
        body: Optional[pl.Expr],
        data: Optional[pl.Expr],
        headers: Optional[pl.Expr],
        *,
        client: Optional[httpx.Client],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        cache: bool,
        with_metadata: bool,
        with_response_headers: bool,
        on_error: OnError,
        on_request: Optional[RequestHook],
        on_response: Optional[ResponseHook],
    ) -> pl.Expr:
        return_dtype = _metadata_dtype(with_response_headers) if with_metadata else pl.Utf8
        return self._input_struct(params, body, data, headers).map_batches(
            lambda s: self._results_to_series(
                self._sync_batch(
                    method,
                    self._rows_from_struct(s),
                    client=client,
                    timeout=timeout,
                    retries=retries,
                    backoff=backoff,
                    cache=cache,
                    with_response_headers=with_response_headers,
                    on_request=on_request,
                    on_response=on_response,
                ),
                with_metadata=with_metadata,
                with_response_headers=with_response_headers,
                on_error=on_error,
            ),
            return_dtype=return_dtype,
        )

    def _async_call(
        self,
        method: str,
        params: Optional[pl.Expr],
        body: Optional[pl.Expr],
        data: Optional[pl.Expr],
        headers: Optional[pl.Expr],
        *,
        client: Optional[aiohttp.ClientSession],
        timeout: Optional[float],
        retries: int,
        backoff: float,
        max_concurrency: Optional[int],
        cache: bool,
        with_metadata: bool,
        with_response_headers: bool,
        on_error: OnError,
        on_request: Optional[AsyncRequestHook],
        on_response: Optional[AsyncResponseHook],
    ) -> pl.Expr:
        return_dtype = _metadata_dtype(with_response_headers) if with_metadata else pl.Utf8
        return self._input_struct(params, body, data, headers).map_batches(
            lambda s: self._results_to_series(
                _arun(
                    self._async_many(
                        method,
                        self._rows_from_struct(s),
                        client=client,
                        timeout=timeout,
                        retries=retries,
                        backoff=backoff,
                        max_concurrency=max_concurrency,
                        cache=cache,
                        with_response_headers=with_response_headers,
                        on_request=on_request,
                        on_response=on_response,
                    )
                ),
                with_metadata=with_metadata,
                with_response_headers=with_response_headers,
                on_error=on_error,
            ),
            return_dtype=return_dtype,
        )

    # ---- Public API: full request() / arequest() entry points ----

    def request(
        self,
        method: str,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        *,
        data: Optional[pl.Expr] = None,
        headers: Optional[pl.Expr] = None,
        client: Optional[httpx.Client] = None,
        timeout: Optional[float] = None,
        retries: int = 0,
        backoff: float = 0.0,
        cache: bool = False,
        with_metadata: bool = False,
        with_response_headers: bool = False,
        on_error: OnError = "null",
        on_request: Optional[RequestHook] = None,
        on_response: Optional[ResponseHook] = None,
        auth: Optional[tuple[str, str]] = None,
        bearer: Optional[Union[str, pl.Expr]] = None,
        api_key: Optional[Union[str, pl.Expr]] = None,
        api_key_header: str = "X-API-Key",
    ) -> pl.Expr:
        """Issue a synchronous HTTP request per row."""
        merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
        return self._sync_call(
            method.upper(),
            params,
            body,
            data,
            merged_headers,
            client=client,
            timeout=timeout,
            retries=retries,
            backoff=backoff,
            cache=cache,
            with_metadata=with_metadata,
            with_response_headers=with_response_headers,
            on_error=on_error,
            on_request=on_request,
            on_response=on_response,
        )

    def arequest(
        self,
        method: str,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        *,
        data: Optional[pl.Expr] = None,
        headers: Optional[pl.Expr] = None,
        client: Optional[aiohttp.ClientSession] = None,
        timeout: Optional[float] = None,
        retries: int = 0,
        backoff: float = 0.0,
        max_concurrency: Optional[int] = None,
        cache: bool = False,
        with_metadata: bool = False,
        with_response_headers: bool = False,
        on_error: OnError = "null",
        on_request: Optional[AsyncRequestHook] = None,
        on_response: Optional[AsyncResponseHook] = None,
        auth: Optional[tuple[str, str]] = None,
        bearer: Optional[Union[str, pl.Expr]] = None,
        api_key: Optional[Union[str, pl.Expr]] = None,
        api_key_header: str = "X-API-Key",
    ) -> pl.Expr:
        """Issue concurrent asynchronous HTTP requests across the batch."""
        merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
        return self._async_call(
            method.upper(),
            params,
            body,
            data,
            merged_headers,
            client=client,
            timeout=timeout,
            retries=retries,
            backoff=backoff,
            max_concurrency=max_concurrency,
            cache=cache,
            with_metadata=with_metadata,
            with_response_headers=with_response_headers,
            on_error=on_error,
            on_request=on_request,
            on_response=on_response,
        )

    # ---- Verb wrappers (sync) ----

    def get(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue a synchronous GET per row."""
        return self.request("GET", params, None, timeout=timeout, **kwargs)

    def post(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue a synchronous POST per row."""
        return self.request("POST", params, body, timeout=timeout, **kwargs)

    def put(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue a synchronous PUT per row."""
        return self.request("PUT", params, body, timeout=timeout, **kwargs)

    def patch(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue a synchronous PATCH per row."""
        return self.request("PATCH", params, body, timeout=timeout, **kwargs)

    def delete(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue a synchronous DELETE per row."""
        return self.request("DELETE", params, None, timeout=timeout, **kwargs)

    def head(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue a synchronous HEAD per row."""
        return self.request("HEAD", params, None, timeout=timeout, **kwargs)

    # ---- Verb wrappers (async) ----

    def aget(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue concurrent asynchronous GETs across the batch."""
        return self.arequest("GET", params, None, timeout=timeout, **kwargs)

    def apost(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue concurrent asynchronous POSTs across the batch."""
        return self.arequest("POST", params, body, timeout=timeout, **kwargs)

    def aput(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue concurrent asynchronous PUTs across the batch."""
        return self.arequest("PUT", params, body, timeout=timeout, **kwargs)

    def apatch(
        self,
        params: Optional[pl.Expr] = None,
        body: Optional[pl.Expr] = None,
        timeout: Optional[float] = None,
        **kwargs: Any,
    ) -> pl.Expr:
        """Issue concurrent asynchronous PATCHes across the batch."""
        return self.arequest("PATCH", params, body, timeout=timeout, **kwargs)

    def adelete(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue concurrent asynchronous DELETEs across the batch."""
        return self.arequest("DELETE", params, None, timeout=timeout, **kwargs)

    def ahead(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
        """Issue concurrent asynchronous HEADs across the batch."""
        return self.arequest("HEAD", params, None, timeout=timeout, **kwargs)

    # ---- Pagination ----

    def paginate(
        self,
        params: Optional[pl.Expr] = None,
        *,
        method: str = "GET",
        max_pages: int = 10,
        next_url: Optional[NextUrl] = None,
        headers: Optional[pl.Expr] = None,
        client: Optional[httpx.Client] = None,
        timeout: Optional[float] = None,
        retries: int = 0,
        backoff: float = 0.0,
        on_request: Optional[RequestHook] = None,
        on_response: Optional[ResponseHook] = None,
        auth: Optional[tuple[str, str]] = None,
        bearer: Optional[Union[str, pl.Expr]] = None,
        api_key: Optional[Union[str, pl.Expr]] = None,
        api_key_header: str = "X-API-Key",
    ) -> pl.Expr:
        """Synchronously paginate per row, following Link: rel="next" by default.

        Returns a column of List[Utf8] — one list of response bodies per starting
        URL. Pipe through `.list.eval(pl.element().str.json_decode())` and `.explode(...)`
        to flatten paginated rows back into the DataFrame.

        Pass `next_url=lambda response: ...` to extract the next URL from a custom
        location (e.g. a JSON field) instead of the Link header.
        """
        merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
        params_expr = pl.lit(None) if params is None else params
        headers_expr = pl.lit(None) if merged_headers is None else merged_headers
        extractor = next_url or (lambda r: _parse_link_next(r.headers.get("link")))
        verb = method.upper()

        def _follow_links(cli: httpx.Client, url: str, p: Any, h: Any) -> list[str]:
            bodies: list[str] = []
            current_url, current_params = url, p
            for _ in range(max_pages):
                response = self._send_sync_with_retries(
                    cli,
                    verb,
                    current_url,
                    current_params,
                    None,
                    None,
                    h,
                    timeout,
                    retries,
                    backoff,
                    on_request,
                    on_response,
                )
                if response is None:
                    break
                bodies.append(response.text)
                nxt = extractor(response)
                if not nxt:
                    break
                current_url = nxt
                current_params = None  # next URL already encodes its own query string
            return bodies

        def _paginate_batch(s: pl.Series) -> pl.Series:
            urls = s.struct.field("url").to_list()
            param_list = s.struct.field("params").to_list()
            header_list = s.struct.field("headers").to_list()
            own_client = client is None
            cli: httpx.Client = httpx.Client() if own_client else client  # type: ignore[assignment]
            try:
                out = [_follow_links(cli, url, p, h) for url, p, h in zip(urls, param_list, header_list)]
                return pl.Series(out, dtype=pl.List(pl.Utf8))
            finally:
                if own_client:
                    cli.close()

        return pl.struct(
            self._url.alias("url"),
            params_expr.alias("params"),
            headers_expr.alias("headers"),
        ).map_batches(_paginate_batch, return_dtype=pl.List(pl.Utf8))

request(method, params=None, body=None, *, data=None, headers=None, client=None, timeout=None, retries=0, backoff=0.0, cache=False, with_metadata=False, with_response_headers=False, on_error='null', on_request=None, on_response=None, auth=None, bearer=None, api_key=None, api_key_header='X-API-Key')

Issue a synchronous HTTP request per row.

Source code in polars_api/api.py

def request(
    self,
    method: str,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    *,
    data: Optional[pl.Expr] = None,
    headers: Optional[pl.Expr] = None,
    client: Optional[httpx.Client] = None,
    timeout: Optional[float] = None,
    retries: int = 0,
    backoff: float = 0.0,
    cache: bool = False,
    with_metadata: bool = False,
    with_response_headers: bool = False,
    on_error: OnError = "null",
    on_request: Optional[RequestHook] = None,
    on_response: Optional[ResponseHook] = None,
    auth: Optional[tuple[str, str]] = None,
    bearer: Optional[Union[str, pl.Expr]] = None,
    api_key: Optional[Union[str, pl.Expr]] = None,
    api_key_header: str = "X-API-Key",
) -> pl.Expr:
    """Issue a synchronous HTTP request per row."""
    merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
    return self._sync_call(
        method.upper(),
        params,
        body,
        data,
        merged_headers,
        client=client,
        timeout=timeout,
        retries=retries,
        backoff=backoff,
        cache=cache,
        with_metadata=with_metadata,
        with_response_headers=with_response_headers,
        on_error=on_error,
        on_request=on_request,
        on_response=on_response,
    )

arequest(method, params=None, body=None, *, data=None, headers=None, client=None, timeout=None, retries=0, backoff=0.0, max_concurrency=None, cache=False, with_metadata=False, with_response_headers=False, on_error='null', on_request=None, on_response=None, auth=None, bearer=None, api_key=None, api_key_header='X-API-Key')

Issue concurrent asynchronous HTTP requests across the batch.

Source code in polars_api/api.py

def arequest(
    self,
    method: str,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    *,
    data: Optional[pl.Expr] = None,
    headers: Optional[pl.Expr] = None,
    client: Optional[aiohttp.ClientSession] = None,
    timeout: Optional[float] = None,
    retries: int = 0,
    backoff: float = 0.0,
    max_concurrency: Optional[int] = None,
    cache: bool = False,
    with_metadata: bool = False,
    with_response_headers: bool = False,
    on_error: OnError = "null",
    on_request: Optional[AsyncRequestHook] = None,
    on_response: Optional[AsyncResponseHook] = None,
    auth: Optional[tuple[str, str]] = None,
    bearer: Optional[Union[str, pl.Expr]] = None,
    api_key: Optional[Union[str, pl.Expr]] = None,
    api_key_header: str = "X-API-Key",
) -> pl.Expr:
    """Issue concurrent asynchronous HTTP requests across the batch."""
    merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
    return self._async_call(
        method.upper(),
        params,
        body,
        data,
        merged_headers,
        client=client,
        timeout=timeout,
        retries=retries,
        backoff=backoff,
        max_concurrency=max_concurrency,
        cache=cache,
        with_metadata=with_metadata,
        with_response_headers=with_response_headers,
        on_error=on_error,
        on_request=on_request,
        on_response=on_response,
    )

get(params=None, timeout=None, **kwargs)

Issue a synchronous GET per row.

Source code in polars_api/api.py

def get(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue a synchronous GET per row."""
    return self.request("GET", params, None, timeout=timeout, **kwargs)

post(params=None, body=None, timeout=None, **kwargs)

Issue a synchronous POST per row.

Source code in polars_api/api.py

def post(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue a synchronous POST per row."""
    return self.request("POST", params, body, timeout=timeout, **kwargs)

put(params=None, body=None, timeout=None, **kwargs)

Issue a synchronous PUT per row.

Source code in polars_api/api.py

def put(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue a synchronous PUT per row."""
    return self.request("PUT", params, body, timeout=timeout, **kwargs)

patch(params=None, body=None, timeout=None, **kwargs)

Issue a synchronous PATCH per row.

Source code in polars_api/api.py

def patch(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue a synchronous PATCH per row."""
    return self.request("PATCH", params, body, timeout=timeout, **kwargs)

delete(params=None, timeout=None, **kwargs)

Issue a synchronous DELETE per row.

Source code in polars_api/api.py

def delete(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue a synchronous DELETE per row."""
    return self.request("DELETE", params, None, timeout=timeout, **kwargs)

head(params=None, timeout=None, **kwargs)

Issue a synchronous HEAD per row.

Source code in polars_api/api.py

def head(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue a synchronous HEAD per row."""
    return self.request("HEAD", params, None, timeout=timeout, **kwargs)

aget(params=None, timeout=None, **kwargs)

Issue concurrent asynchronous GETs across the batch.

Source code in polars_api/api.py

def aget(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue concurrent asynchronous GETs across the batch."""
    return self.arequest("GET", params, None, timeout=timeout, **kwargs)

apost(params=None, body=None, timeout=None, **kwargs)

Issue concurrent asynchronous POSTs across the batch.

Source code in polars_api/api.py

def apost(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue concurrent asynchronous POSTs across the batch."""
    return self.arequest("POST", params, body, timeout=timeout, **kwargs)

aput(params=None, body=None, timeout=None, **kwargs)

Issue concurrent asynchronous PUTs across the batch.

Source code in polars_api/api.py

def aput(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue concurrent asynchronous PUTs across the batch."""
    return self.arequest("PUT", params, body, timeout=timeout, **kwargs)

apatch(params=None, body=None, timeout=None, **kwargs)

Issue concurrent asynchronous PATCHes across the batch.

Source code in polars_api/api.py

def apatch(
    self,
    params: Optional[pl.Expr] = None,
    body: Optional[pl.Expr] = None,
    timeout: Optional[float] = None,
    **kwargs: Any,
) -> pl.Expr:
    """Issue concurrent asynchronous PATCHes across the batch."""
    return self.arequest("PATCH", params, body, timeout=timeout, **kwargs)

adelete(params=None, timeout=None, **kwargs)

Issue concurrent asynchronous DELETEs across the batch.

Source code in polars_api/api.py

def adelete(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue concurrent asynchronous DELETEs across the batch."""
    return self.arequest("DELETE", params, None, timeout=timeout, **kwargs)

ahead(params=None, timeout=None, **kwargs)

Issue concurrent asynchronous HEADs across the batch.

Source code in polars_api/api.py

def ahead(self, params: Optional[pl.Expr] = None, timeout: Optional[float] = None, **kwargs: Any) -> pl.Expr:
    """Issue concurrent asynchronous HEADs across the batch."""
    return self.arequest("HEAD", params, None, timeout=timeout, **kwargs)

paginate(params=None, *, method='GET', max_pages=10, next_url=None, headers=None, client=None, timeout=None, retries=0, backoff=0.0, on_request=None, on_response=None, auth=None, bearer=None, api_key=None, api_key_header='X-API-Key')

Synchronously paginate per row, following Link: rel="next" by default.

Returns a column of List[Utf8] — one list of response bodies per starting URL. Pipe through .list.eval(pl.element().str.json_decode()) and .explode(...) to flatten paginated rows back into the DataFrame.

Pass next_url=lambda response: ... to extract the next URL from a custom location (e.g. a JSON field) instead of the Link header.

Source code in polars_api/api.py

def paginate(
    self,
    params: Optional[pl.Expr] = None,
    *,
    method: str = "GET",
    max_pages: int = 10,
    next_url: Optional[NextUrl] = None,
    headers: Optional[pl.Expr] = None,
    client: Optional[httpx.Client] = None,
    timeout: Optional[float] = None,
    retries: int = 0,
    backoff: float = 0.0,
    on_request: Optional[RequestHook] = None,
    on_response: Optional[ResponseHook] = None,
    auth: Optional[tuple[str, str]] = None,
    bearer: Optional[Union[str, pl.Expr]] = None,
    api_key: Optional[Union[str, pl.Expr]] = None,
    api_key_header: str = "X-API-Key",
) -> pl.Expr:
    """Synchronously paginate per row, following Link: rel="next" by default.

    Returns a column of List[Utf8] — one list of response bodies per starting
    URL. Pipe through `.list.eval(pl.element().str.json_decode())` and `.explode(...)`
    to flatten paginated rows back into the DataFrame.

    Pass `next_url=lambda response: ...` to extract the next URL from a custom
    location (e.g. a JSON field) instead of the Link header.
    """
    merged_headers = self._build_headers_expr(headers, auth, bearer, api_key, api_key_header)
    params_expr = pl.lit(None) if params is None else params
    headers_expr = pl.lit(None) if merged_headers is None else merged_headers
    extractor = next_url or (lambda r: _parse_link_next(r.headers.get("link")))
    verb = method.upper()

    def _follow_links(cli: httpx.Client, url: str, p: Any, h: Any) -> list[str]:
        bodies: list[str] = []
        current_url, current_params = url, p
        for _ in range(max_pages):
            response = self._send_sync_with_retries(
                cli,
                verb,
                current_url,
                current_params,
                None,
                None,
                h,
                timeout,
                retries,
                backoff,
                on_request,
                on_response,
            )
            if response is None:
                break
            bodies.append(response.text)
            nxt = extractor(response)
            if not nxt:
                break
            current_url = nxt
            current_params = None  # next URL already encodes its own query string
        return bodies

    def _paginate_batch(s: pl.Series) -> pl.Series:
        urls = s.struct.field("url").to_list()
        param_list = s.struct.field("params").to_list()
        header_list = s.struct.field("headers").to_list()
        own_client = client is None
        cli: httpx.Client = httpx.Client() if own_client else client  # type: ignore[assignment]
        try:
            out = [_follow_links(cli, url, p, h) for url, p, h in zip(urls, param_list, header_list)]
            return pl.Series(out, dtype=pl.List(pl.Utf8))
        finally:
            if own_client:
                cli.close()

    return pl.struct(
        self._url.alias("url"),
        params_expr.alias("params"),
        headers_expr.alias("headers"),
    ).map_batches(_paginate_batch, return_dtype=pl.List(pl.Utf8))