Python SDK

Official Python client for the Hotdata HTTP API — typed, Pydantic-validated, and generated from the OpenAPI spec.

Install

pip install hotdata

Supported Python versions are listed on GitHub.

For Apache Arrow result support (faster, more memory-efficient for large result sets):

pip install 'hotdata[arrow]'

Authentication

import hotdata

configuration = hotdata.Configuration(
    api_key="YOUR_API_KEY",
    workspace_id="YOUR_WORKSPACE_ID",
)

host defaults to https://api.hotdata.dev. Override it if you target another environment.

Quickstart

import hotdata

configuration = hotdata.Configuration(
    api_key="YOUR_API_KEY",
    workspace_id="YOUR_WORKSPACE_ID",
)

# Every query is scoped to an instant database — see below for how to get an id.
database_id = "dbid6lguax1dxn9y1xj5gxnameyywl"

with hotdata.ApiClient(configuration) as api_client:
    query_api = hotdata.QueryApi(api_client)
    response = query_api.query(
        hotdata.QueryRequest(sql="SELECT 1 AS ok"),
        x_database_id=database_id,
    )
    print(response)

A database scope is not optional. A database is the only window into catalogs, so a query that names none is rejected:

BadRequestException: (400)
{"error":{"code":"BAD_REQUEST","message":"a database is required: set the
 X-Database-Id header or the database_id body field"}}

The same scope is required on stored results and query runs. Get an id from hotdata databases list, from the dashboard, or from DatabasesApi below.

Execute SQL

with hotdata.ApiClient(configuration) as api_client:
    query_api = hotdata.QueryApi(api_client)

    # Synchronous query
    response = query_api.query(
        hotdata.QueryRequest(sql="SELECT * FROM orders LIMIT 10"),
        x_database_id=database_id,
    )

    # Async query — returns a query run ID for polling
    response = query_api.query(
        hotdata.QueryRequest(
            sql="SELECT * FROM large_table",
            var_async=True,
        ),
        x_database_id=database_id,
    )
    run_id = response.query_run_id

    # Try sync first, fall back to async after 3 s
    response = query_api.query(
        hotdata.QueryRequest(
            sql="SELECT * FROM orders",
            var_async=True,
            async_after_ms=3000,
        ),
        x_database_id=database_id,
    )

The database_id body field is the equivalent of the x_database_id header. Send exactly one of the two — if both are present and they disagree, the request is a 400:

response = query_api.query(
    hotdata.QueryRequest(
        sql="SELECT * FROM default.public.orders LIMIT 5",
        database_id=database_id,
    )
)

Instant databases

with hotdata.ApiClient(configuration) as api_client:
    db_api = hotdata.DatabasesApi(api_client)

    # Create a database and declare tables
    created = db_api.create_database(
        hotdata.CreateDatabaseRequest(
            name="sales",
            expires_at="24h",
            schemas=[
                hotdata.DatabaseDefaultSchemaDecl(
                    name="public",
                    tables=[
                        hotdata.DatabaseDefaultTableDecl(name="orders"),
                        hotdata.DatabaseDefaultTableDecl(name="customers"),
                    ],
                )
            ],
        )
    )
    print(created.id)  # e.g. "dbid6lguax1dxn9y1xj5gxnameyywl"

    # List all databases
    listing = db_api.list_databases()
    for db in listing.databases:
        print(db.id, db.name)

    # Get a specific database
    detail = db_api.get_database(created.id)
    print(detail.default_connection_id)

    # Delete a database — teardown, run this after the sections below
    db_api.delete_database(created.id)

Load parquet into a managed table

Upload a parquet file and load it into a declared table:

with hotdata.ApiClient(configuration) as api_client:
    uploads_api = hotdata.UploadsApi(api_client)
    connections_api = hotdata.ConnectionsApi(api_client)

    # Upload the file directly to storage: the SDK opens an upload session,
    # PUTs the bytes, and finalizes it in one call.
    upload = uploads_api.upload_file(
        "orders.parquet",
        content_type="application/parquet",
    )

    # Load into the declared table
    # (the generated client names the schema parameter var_schema)
    result = connections_api.load_managed_table(
        connection_id=detail.default_connection_id,
        var_schema="public",
        table="orders",
        load_managed_table_request=hotdata.LoadManagedTableRequest(
            mode="replace",
            upload_id=upload.upload_id,
        ),
    )
    print(result.row_count, result.table_name)

Apache Arrow results

Fetch results as an Arrow table instead of JSON — faster and more memory-efficient for large result sets:

from hotdata import ApiClient, Configuration
from hotdata.arrow import ResultsApi

with ApiClient(configuration) as client:
    results = ResultsApi(client)

    # Buffered — returns a pyarrow.Table
    table = results.get_result_arrow(result_id, x_database_id=database_id)

    # Streaming — yields batches without materializing the full table
    with results.stream_result_arrow(result_id, x_database_id=database_id) as reader:
        for batch in reader:
            print(batch.to_pandas())

Both methods take x_database_id — the database the result belongs to — and accept offset and limit for pagination. They raise hotdata.arrow.ResultNotReadyError if the result is still pending — poll results.get_result(result_id, x_database_id=database_id) until status == "ready" first.

Workspaces

with hotdata.ApiClient(configuration) as api_client:
    workspaces_api = hotdata.WorkspacesApi(api_client)

    listing = workspaces_api.list_workspaces()
    for ws in listing.workspaces:
        print(ws.public_id, ws.name)

Query run history

with hotdata.ApiClient(configuration) as api_client:
    runs_api = hotdata.QueryRunsApi(api_client)
    results_api = hotdata.ResultsApi(api_client)

    # List recent runs — scoped to a database (X-Database-Id is required)
    listing = runs_api.list_query_runs(x_database_id=database_id, limit=20)
    for run in listing.query_runs:
        print(run.id, run.status, run.execution_time_ms)

    # Fetch stored result rows — results are scoped to the database
    # they were queried in
    result = results_api.get_result(run.result_id, x_database_id=database_id)

Error handling

from hotdata.rest import ApiException

try:
    response = query_api.query(
        hotdata.QueryRequest(sql="SELECT * FROM missing_table"),
        x_database_id=database_id,
    )
except ApiException as e:
    print(f"API error {e.status}: {e.reason}")
    print(e.body)

API classes

See also

• hotdata on PyPI
• sdk-python on GitHub
• Rust SDK — the same API from Rust
• API Reference — Full HTTP API documentation
• Quick Start — CLI and workspace setup