advancedView Python source

Read a REST API as a table

This example reads 487 support tickets from a paginated JSON API as a DataFrame, saves them once, and aggregates them by status. It uses a Python data source with one partition per API page, and the engine reads partitions as separate tasks, so pages are fetched concurrently.

Run it

Run all commands below from the repository root (sail-examples/):

uv run python/rest-api-as-a-table/main.py

A small local service stands in for the API (GET /tickets?page=N, 50 tickets per page), so the example runs offline.

Output

The newest capture appears first. Older output stays here when a result changes.

Current output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-05, Python 3.12.14.

+-------+------------+
|tickets|distinct_ids|
+-------+------------+
|    487|         487|
+-------+------------+

+-------+-------+------------+
| status|tickets|avg_priority|
+-------+-------+------------+
| closed|    122|         2.0|
|   open|    122|         1.0|
|pending|    121|         4.0|
| solved|    122|         3.0|
+-------+-------+------------+

Python code

main.py150 lines

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
# ]
# ///
"""Read a paginated REST API as a table with a Python data source.

`TicketsDataSource` uses the Spark Python Data Source API: `partitions()` asks the API how many
pages there are, and `read()` fetches one page per partition, so pages load in parallel. A
small local service stands in for the API so the example runs offline.
"""

import atexit
import json
import shutil
import tempfile
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition

PAGE_SIZE = 50
TOTAL_TICKETS = 487
STATUSES = ["open", "pending", "solved", "closed"]


def ticket(i: int) -> dict:
    return {
        "id": i,
        "subject": f"Ticket {i}",
        "status": STATUSES[(i * 7) % len(STATUSES)],
        "priority": 1 + (i * 13) % 4,
    }


class StandInApi:
    """GET /tickets?page=N returns one page of tickets and the page count."""

    def __init__(self):
        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                page = int(parse_qs(urlparse(self.path).query).get("page", ["1"])[0])
                start = (page - 1) * PAGE_SIZE
                items = [ticket(i) for i in range(start, min(start + PAGE_SIZE, TOTAL_TICKETS))]
                pages = -(-TOTAL_TICKETS // PAGE_SIZE)
                body = json.dumps({"page": page, "pages": pages, "items": items}).encode()
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

            def log_message(self, *args):
                pass

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.url = f"http://127.0.0.1:{self.server.server_port}/tickets"
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    def stop(self):
        self.server.shutdown()
        self.server.server_close()


def temp_dir(prefix: str) -> str:
    """A temporary directory that is removed when the process exits."""
    path = tempfile.mkdtemp(prefix=prefix)
    atexit.register(shutil.rmtree, path, ignore_errors=True)
    return path


def get_page(url: str, page: int) -> dict:
    with urllib.request.urlopen(f"{url}?page={page}", timeout=30) as response:
        return json.load(response)


class PagePartition(InputPartition):
    def __init__(self, page: int):
        super().__init__(page)
        self.page = page


class TicketsReader(DataSourceReader):
    def __init__(self, url: str):
        self.url = url

    def partitions(self):
        pages = get_page(self.url, 1)["pages"]
        return [PagePartition(page) for page in range(1, pages + 1)]

    def read(self, partition):
        for item in get_page(self.url, partition.page)["items"]:
            yield (item["id"], item["subject"], item["status"], item["priority"])


class TicketsDataSource(DataSource):
    @classmethod
    def name(cls):
        return "tickets_api"

    def schema(self):
        return "id long, subject string, status string, priority int"

    def reader(self, schema):
        return TicketsReader(self.options["url"])


def run(spark: SparkSession) -> dict[str, DataFrame]:
    api = StandInApi()
    try:
        spark.dataSource.register(TicketsDataSource)
        tickets = spark.read.format("tickets_api").option("url", api.url).load()
        # Read the API once: save the rows, then run every query on the saved copy.
        saved_path = temp_dir("tickets-")
        tickets.write.mode("overwrite").parquet(saved_path)
    finally:
        api.stop()

    saved = spark.read.parquet(saved_path)
    by_status = saved.groupBy("status").agg(
        F.count("*").alias("tickets"), F.round(F.avg("priority"), 3).alias("avg_priority")
    )
    totals = saved.agg(F.count("*").alias("tickets"), F.countDistinct("id").alias("distinct_ids"))
    return {
        "by_status": by_status.orderBy("status"),
        "first_tickets": saved.orderBy("id").limit(5),
        "totals": totals,
    }


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    results = run(spark)
    results["totals"].show()
    results["by_status"].show()
    spark.stop()
    server.stop()

How it works

TicketsDataSource subclasses pyspark.sql.datasource.DataSource. It declares the schema and returns a reader for the url option.
partitions() asks the API for page 1 and returns one InputPartition per page.
read(partition) fetches its page and yields one tuple per ticket. Each partition runs as its own task, so several pages are fetched at the same time on both Sail and Spark.
spark.dataSource.register(...) makes the source available as spark.read.format("tickets_api"). After that, it is an ordinary DataFrame: filter it, join it, write it to a table.
Save once. Every action on a data source DataFrame reads the API again, so the example writes the rows to Parquet and runs its queries on the saved copy.

Use a real API

Replace get_page with a call to your API. Options are part of the query plan and can appear in plan output and logs, so do not pass secrets as options: read credentials inside read() from the environment of the process that runs the reader, or from a secret manager. If the API is rate limited, return fewer, larger partitions.