Metadata-Version: 2.4
Name: spark-rest-api-reader
Version: 0.3.0
Summary: A generic Spark Data Source for reading GET REST APIs into DataFrames, batch or streaming.
Author-email: Liam McElhaney <liam.mcelhaney122@gmail.com>
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/liam-mcelhaney122/spark-rest-api-reader
Project-URL: Documentation, https://liam-mcelhaney122.github.io/spark-rest-api-reader/
Project-URL: Repository, https://github.com/liam-mcelhaney122/spark-rest-api-reader
Project-URL: Issues, https://github.com/liam-mcelhaney122/spark-rest-api-reader/issues
Project-URL: Changelog, https://github.com/liam-mcelhaney122/spark-rest-api-reader/blob/main/CHANGELOG.md
Keywords: spark,pyspark,databricks,rest-api,data-source,etl,ingestion,streaming
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: NOTICE
Requires-Dist: httpx>=0.27
Requires-Dist: tenacity>=8.2
Provides-Extra: docs
Requires-Dist: mkdocs-material>=9; extra == "docs"
Provides-Extra: dev
Requires-Dist: pyspark>=4.0.0; extra == "dev"
Requires-Dist: pyarrow>=11.0; extra == "dev"
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: pytest-mock>=3.11; extra == "dev"
Requires-Dist: ruff>=0.6; extra == "dev"
Requires-Dist: mypy>=1.10; extra == "dev"
Requires-Dist: build>=1.2; extra == "dev"
Dynamic: license-file

# spark-rest-api-reader

[![CI](https://github.com/liam-mcelhaney122/spark-rest-api-reader/actions/workflows/ci.yml/badge.svg)](https://github.com/liam-mcelhaney122/spark-rest-api-reader/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/spark-rest-api-reader)](https://pypi.org/project/spark-rest-api-reader/)
[![Python versions](https://img.shields.io/pypi/pyversions/spark-rest-api-reader)](https://pypi.org/project/spark-rest-api-reader/)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue)](LICENSE)

**Full documentation:** <https://liam-mcelhaney122.github.io/spark-rest-api-reader/>

A generic Spark [Python Data Source](https://spark.apache.org/docs/latest/api/python/tutorial/sql/python_data_source.html)
for pulling data out of any **GET** REST API — returning JSON, CSV, or XML —
and into a DataFrame, batch or streaming, without writing a bespoke connector
for every API.

Requires **Spark 4.0+ / Databricks Runtime 15.4+** (the Python Data Source
API). Not tied to Databricks otherwise — it works with any Spark 4.0+ cluster.

## Why

Every project ends up writing the same HTTP-get + pagination + retry +
auth-header boilerplate to land a REST API into Bronze. This wraps that
boilerplate once, behind Spark options, so pulling a new API is a matter of
configuring `auth_type`/`pagination_type`, not writing new Python.

Built on [`httpx`](https://www.python-httpx.org/) for the HTTP client and
[`tenacity`](https://tenacity.readthedocs.io/) for retry/backoff.

## Install

```bash
pip install spark-rest-api-reader
```

On a Databricks cluster, install it as a cluster library (PyPI source:
`spark-rest-api-reader`) or per-notebook:

```python
%pip install spark-rest-api-reader
```

For air-gapped clusters, build the wheel yourself and upload it as a
cluster library:

```bash
pip install build
python -m build   # produces dist/spark_rest_api_reader-*.whl
```

`pyspark`/`pyarrow` are intentionally **not** pinned as install dependencies —
the cluster already provides them, and pinning risks conflicting with the
runtime's own version. For local development/tests, install `.[dev]`.

## Quick start

```python
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

from spark_rest_api_reader import RestApiDataSource

spark.dataSource.register(RestApiDataSource)

items_schema = StructType(
    [
        StructField("id", IntegerType()),
        StructField("name", StringType()),
        StructField("status", StringType()),
    ]
)

df = (
    spark.read.format("rest_api")
    .schema(items_schema)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "bearer")
    .option("token", dbutils.secrets.get("api-scope", "api-token"))
    .option("pagination_type", "page_number")
    .load()
)

df.write.format("delta").mode("append").saveAsTable("main.bronze.items_raw")
```

`spark.dataSource.register` only needs to run once per session (e.g. once at
the top of a job/notebook).

**Secrets:** resolve `dbutils.secrets.get(...)` (or any other secret lookup)
on the driver *before* passing the value into `.option(...)`. Options are
plain strings that get shipped to executors as part of the reader — do not
put a `dbutils` call anywhere reachable from executor code, since `dbutils`
isn't available there.

## How it's organized

- **Auth** (`auth_type`) — how the request is authenticated.
- **Pagination** (`pagination_type`) — how a *single partition* walks through
  multiple pages of results until the API says "no more."
- **Partitioning** (`partitions` / `pages`) — how the overall pull is split
  into independent units of work that Spark can fan out across executors.
- **Response format** (`response_format`) — how the response body is parsed:
  `json` (default), `csv`, or `xml`. Everything downstream (records
  extraction, schemas, pagination stop-conditions) works the same on all
  three.
- **Records extraction** (`records_path`, `flatten`) — how to find the list
  of records inside the parsed response body, and whether to flatten nested
  objects into dotted column names.
- **Schema** — supply a `StructType` via `.schema(...)` (or the `schema`
  option), or set `infer_schema=true` to sample and infer one, Auto
  Loader-style.
- **Streaming** — an incremental variant driven by a cursor field in the
  response (e.g. `updated_at`), via `spark.readStream`.

Partitioning and pagination are independent: e.g. you can partition by
region (`partitions`) and have each region's partition separately paginate
through everything on its own (`pagination_type`), or auto-generate one
partition per page number (`pages`) for maximum parallelism when the API
doesn't need per-partition pagination at all.

## Option reference

### Request basics

| Option | Default | Description |
|---|---|---|
| `url` | *required* | Base URL of the GET endpoint. |
| `headers` | `{}` | JSON object of static request headers. |
| `query_params` | `{}` | JSON object of static query-string params sent on every request. |
| `timeout` | `30` | Per-request timeout in seconds. |
| `max_retries` | `3` | Retries on transient failures (connection errors + `retry_status_codes`). |
| `backoff_factor` | `0.5` | Exponential backoff factor between retries. |
| `retry_status_codes` | `[429,500,502,503,504]` | JSON array of status codes to retry (entries are coerced to integers; a non-numeric entry raises a `ConfigError`). `Retry-After` is honored when present, regardless of status code. |
| `verify_ssl` | `true` | Set `false` to skip TLS verification (not recommended). |
| `ca_bundle_path` | — | Path to a custom CA bundle; overrides `verify_ssl` when set. |
| `proxies` | `{}` | JSON object, e.g. `{"https": "http://proxy:8080"}`. Only one proxy is used per client (prefers `https`, then `http`, then any key given) — httpx applies a single proxy to all requests rather than mounting per scheme. |
| `requests_per_second` | — | Client-side throttle applied per partition. |

### Auth (`auth_type`)

| `auth_type` | Extra options |
|---|---|
| `none` (default) | — |
| `basic` | `username`, `password` |
| `bearer` | `token` |
| `api_key` | `api_key_value`, `api_key_name` (default `X-API-Key`), `api_key_location` (`header` \| `query`, default `header`) |
| `header` | `auth_headers` — JSON object merged into request headers verbatim; the escape hatch for custom schemes |
| `oauth2_client_credentials` | `oauth_token_url`, `oauth_client_id`, `oauth_client_secret`, `oauth_scope`, `oauth_audience`, `oauth_extra_params` (JSON), `oauth_client_auth_mode` (`basic` \| `body`, default `basic`), `oauth_token_field` (default `access_token`), `oauth_expires_in_field` (default `expires_in`) |

OAuth2 tokens are fetched lazily per executor process and cached until ~30
seconds before expiry, then refreshed automatically.

### Response format

| Option | Default | Description |
|---|---|---|
| `response_format` | `json` | How the response body is parsed: `json`, `csv`, or `xml`. |
| `csv_delimiter` | `,` | CSV only — the field delimiter. |
| `xml_attribute_prefix` | `_` | XML only — prefix for keys derived from XML attributes (matching Spark's XML source convention). |

- **`csv`** — a header row is required; each data row becomes one record,
  every value a string (coerce via the schema, or let `infer_schema` sniff
  the types). Empty cells become `null`. Incompatible with `records_path`
  (the body already *is* the record list) and with `pagination_type=cursor`
  (no envelope to read a next-cursor from).
- **`xml`** — the element tree maps onto nested dicts/lists: repeated
  elements become arrays, attributes become `{prefix}`-prefixed keys, and an
  element with both attributes and text keeps its text under `{prefix}VALUE`.
  Dot paths (`records_path`, `next_cursor_path`, `stream_cursor_field`) and
  nested-struct schemas then work exactly as for JSON; paths are relative to
  *inside* the root element (e.g. `items.item` for
  `<response><items><item>...`).

### Records extraction

| Option | Default | Description |
|---|---|---|
| `records_path` | — | Dot path to the record list inside the parsed body (JSON/XML), e.g. `envelope.payload`. If unset, tries the body itself (if it's a list) or common wrapper keys (`results`, `data`, `items`, `records`, `value`), else treats the whole body as one record. |
| `flatten` | `false` | Flattens nested objects into dotted column names (e.g. `address.city`). Lists are left as-is. |

### Schema

Schemas follow the **structs pattern**: nested objects are declared and
returned as real `StructType`/`ArrayType` columns — build them as
`StructType` objects, not DDL strings, and never model nested data as
JSON-string columns.

**Manually defined** (preferred for production jobs):

```python
from pyspark.sql.types import ArrayType, IntegerType, StringType, StructField, StructType

schema = StructType(
    [
        StructField("id", IntegerType()),
        StructField("address", StructType([StructField("city", StringType())])),
        StructField("tags", ArrayType(StringType())),
    ]
)

df = spark.read.format("rest_api").schema(schema).option("url", ...).load()
```

| Option | Description |
|---|---|
| `.schema(StructType(...))` | Preferred — pass the `StructType` directly on the `DataFrameReader`. |
| `schema` | Same, as an option: pass `your_struct_type.json()` (Spark's JSON schema representation, also what `df.schema.json()` produces). Parsed via `StructType.fromJson`, which — unlike `StructType.fromDDL` — is pure Python and needs no active `SparkContext`; that matters because `DataSource.schema()` runs in an isolated Python worker process with no driver context. A DDL string (`"id int, name string"`, auto-detected by the missing leading `{`) is still accepted for back-compat. |
| `infer_schema` | `true` to sample the first page and infer a schema from the sampled values — see below. |

**Inferred** (Auto Loader-style, for all three response formats):

| Option | Default | Description |
|---|---|---|
| `infer_schema` | `false` | Sample the first page and infer the schema from the sampled records. |
| `infer_sample_size` | `100` | Max records sampled for inference. |
| `infer_column_types` | `true` | Infer real leaf types. Set `false` for Auto Loader's conservative default — every leaf a `string` — while still preserving the full nested struct/array shape (nested data is never collapsed to a JSON string, in either mode). For untyped formats (`csv`/`xml`, where every raw value is a string), `true` additionally sniffs string values into `long`/`double`/`boolean`/`date`/`timestamp`. |
| `schema_hints` | — | Auto Loader-style overrides applied on top of the inferred schema: comma-separated `"<column path> <type>"` entries, e.g. `"version int, user_info.dob date, tags array<int>"`. Dotted paths descend through structs and array-of-struct elements; a hinted top-level column that wasn't inferred is appended. Only valid together with `infer_schema=true` (with an explicit schema, fold the type into the schema itself). |

Inference merges field sets across sampled records, widens mixed numeric
samples to `double` (and mixed date/timestamp to `timestamp`), and falls
back to `string` only on genuine type conflicts.

Nested response objects map onto nested `StructType`/`ArrayType` schema
fields automatically. ISO-8601 timestamp/date strings are parsed for
`TimestampType`/`DateType` columns.

**Type coercion:** values are best-effort coerced to match the declared
column type — a stringified number/boolean (`"123"`, `"true"`) coerces to
`Integer`/`Boolean`, and a non-string value in a `StringType` column is
stringified. This matters because REST APIs commonly stringify numbers to
dodge JS float-precision issues. A value that genuinely can't be coerced
(e.g. an object where a scalar is expected) raises `SchemaTypeError` naming
the exact field path and both types involved, instead of silently reaching
Spark's Arrow conversion and failing there with an opaque, unattributed
error several layers removed from the actual problem.

### Schema drift

The schema is fixed once resolved — it is not re-inferred mid-batch or
mid-stream. By default, fields the API returns that aren't in the schema are
silently dropped (and fields the schema expects but the API omits become
`null`). Two options change that:

| Option | Default | Description |
|---|---|---|
| `rescued_data_column` | — | Name of a `StringType` column (must be declared in the schema) that captures any top-level record keys not otherwise mapped, as a JSON string. `null` when nothing drifted. With `infer_schema=true`, this column is added to the inferred schema automatically; with a DDL/JSON `schema`, you must declare it yourself or resolution fails with a clear error. |
| `fail_on_new_fields` | `false` | Raise immediately the first time a record has a top-level key outside the schema, instead of silently dropping it. Takes precedence over `rescued_data_column` if both are set. |

Drift detection is **top-level only** — new keys appearing inside an array
element aren't (currently) surfaced by either option. One nuance: with
`flatten=true`, nested dict keys are promoted to top-level dotted names
*before* drift detection runs, so nested-*dict* drift actually is caught in
that combination — only drift nested inside an array stays invisible either
way.

### In-band errors

Some APIs report failure inside a successful HTTP response — ArcGIS, for
example, answers `200 OK` with `{"error": {...}}` in the body.
`raise_for_status()` never sees those; untreated, they surface as confusing
downstream symptoms ("zero records to infer from" at schema inference, or
pagination silently ending early and truncating the data). Two composable
mechanisms catch them; every page of every request is checked, including the
schema-inference sample:

| Option | Default | Description |
|---|---|---|
| `error_path` | — | Dot path into the parsed body (JSON/XML). Any response where the path resolves to a non-null value fails the read with a `ResponseError` carrying that value, e.g. `.option("error_path", "error")` for ArcGIS. |

For checks a dot path can't express, attach a **`response_validator`**
function to a `RestApiDataSource` subclass and register the subclass under
its own name. The validator receives the parsed body of each page and
returns an error message to fail the read (or `None` to accept it); raising
its own exception also fails the read. When both are configured,
`error_path` is checked first.

```python
from spark_rest_api_reader import RestApiDataSource

class ArcGisDataSource(RestApiDataSource):
    @classmethod
    def name(cls):
        return "arcgis_rest"

    @staticmethod
    def response_validator(body):
        if isinstance(body, dict) and "error" in body:
            return f"ArcGIS error: {body['error']}"

spark.dataSource.register(ArcGisDataSource)

df = spark.read.format("arcgis_rest").option("url", ...).load()
```

It must be a subclass attribute rather than an `.option()` because option
values are plain strings, and because data source code runs in isolated
worker processes: a subclass defined in a notebook is cloudpickled **by
value** at `register()` time, so the attached function travels to every
worker — module-level state (a registry, a global) would not.

### Pagination (`pagination_type`) — within one partition

| `pagination_type` | Extra options |
|---|---|
| `none` (default) | One request per partition. |
| `page_number` | `page_param` (default `page`), `start_page` (default `1`), `page_size_param`, `page_size`. Stops when a page comes back empty, checked via `records_path` the same way records are actually extracted (so a custom envelope's emptiness is read correctly, not just the top-level body/common wrapper keys) — also stops on a non-JSON response body (e.g. a 204 on the true last page) rather than continuing to increment. |
| `offset_limit` | `offset_param` (default `offset`), `limit_param` (default `limit`), `limit` (default `100`), `start_offset` (default `0`). Stops when a page returns fewer than `limit` records, again counted via `records_path`. |
| `cursor` | `cursor_param` (default `cursor`), `next_cursor_path` (**required** — dot path to the next-cursor value in the response body; JSON/XML only). Stops only when the path resolves to `null` or `""` — a falsy-but-real value like `0` or `false` is a legitimate cursor and does **not** stop pagination. |
| `link_header` | Follows the standard `Link: <url>; rel="next"` response header. Stops when absent. |

`max_pages_per_partition` (default `10000`) is a safety valve against
runaway pagination loops.

### Partitioning — how the pull fans out across Spark

| Option | Description |
|---|---|
| `partitions` | JSON array of objects; one partition per object, each merged into that partition's query params. A `_url` key inside an object overrides the base URL entirely (e.g. per-resource-id endpoints). Each partition still runs the configured `pagination_type` to walk all of its own pages. Every entry must be a JSON object — a non-object entry raises a `ConfigError` naming the offending index. |
| `pages` | Integer — auto-generates one partition per page number (`1..pages`, using `page_param`), for maximum parallelism against APIs where pages can be fetched independently. Mutually exclusive with in-partition pagination (each partition fetches exactly its one page). |
| *(neither set)* | Single partition; `pagination_type` walks all pages sequentially. |

### Streaming (`spark.readStream`)

| Option | Description |
|---|---|
| `stream_cursor_field` | Dot path to the high-water-mark field in each returned record (e.g. `updated_at`, or `meta.updated_at` for a nested field). |
| `stream_cursor_param` | Query param name sent upstream with the last-seen cursor value (e.g. `updated_since`). |
| `stream_initial_value` | Starting cursor value for the very first micro-batch (default `""`). |
| `stream_cursor_is_numeric` | `false` (default). Set `true` if `stream_cursor_field` is a raw (non-zero-padded) number — see below. |

Each micro-batch pulls everything from `stream_cursor_field > last value`
using whatever `pagination_type` is configured, all within a single
partition. Simple and correct; if you need intra-batch parallelism, prefer
the batch reader with `partitions` on a schedule instead.

**Cursor comparison:** by default, "greater than" is a **string** comparison
— correct for ISO-8601 timestamps and zero-padded numeric IDs, but wrong for
raw (non-zero-padded) numeric cursors once they cross a digit-count boundary
(`"9" > "10"` is `True` lexicographically). Set `stream_cursor_is_numeric=true`
for a numeric cursor field to compare correctly instead.

**Restart/checkpoint safety requires pyspark >= 4.2.** This reader implements
the admission-control `latestOffset(start, limit)` signature, which is the
only one that actually receives the restored checkpoint offset after a
query restart (driver restart, job retry, redeploy, etc.) — pyspark added
this in 4.2.0. On an older pyspark/DBR where only the legacy zero-arg
`latestOffset()` convention exists, the reader still works, but the cursor
can only be tracked in an in-memory attribute that resets on every fresh
process: a restart resumes from `stream_initial_value` again, reprocessing
everything already ingested, rather than crashing. If your streaming jobs
need to survive restarts without full reprocessing, confirm the cluster's
pyspark version is >= 4.2 before relying on this in production.

## Examples

- `examples/batch_example.py` — batch read into Delta, with commented
  alternatives for every pagination style (`page_number`, `offset_limit`,
  `cursor`, `link_header`), auto page-range/explicit-partition fan-out,
  `records_path`/`flatten` for custom envelopes, schema drift handling
  (`rescued_data_column`/`fail_on_new_fields`), passing a `StructType`
  through the `schema` option, `infer_schema` with `schema_hints`, and
  `csv`/`xml` response formats.
- `examples/auth_examples.py` — every `auth_type` (`none`, `basic`,
  `bearer`, `api_key` as a header or query param, `header` for custom
  schemes, `oauth2_client_credentials`) as an independent, copy-pasteable
  snippet.
- `examples/streaming_example.py` — incremental streaming pull driven by a
  cursor field, with Spark handling checkpointing between micro-batches.

## Development

```bash
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest
```

Tests spin up a real local `SparkSession` plus a real local HTTP server (not
mocks) so that requests issued from PySpark's separate worker processes are
exercised end-to-end.

See [CONTRIBUTING.md](CONTRIBUTING.md) for the full development guide, and
[CHANGELOG.md](CHANGELOG.md) for release history.

## License

[Apache-2.0](LICENSE)
