# Python SDK

Source: https://www.hotdata.dev/docs/python-sdk
Site index: https://www.hotdata.dev/llms.txt

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

## Install

```bash
pip install hotdata
```

Supported Python versions are listed on [GitHub](https://github.com/hotdata-dev/sdk-python).

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

```bash
pip install 'hotdata[arrow]'
```

## Authentication

```python
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

```python
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`](#instant-databases) below.

## Execute SQL

```python
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:

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

## Instant databases

```python
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:

```python
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:

```python
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

```python
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

```python
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

```python
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

| Class | Description |
|-------|-------------|
| `QueryApi` | Execute SQL queries |
| `DatabasesApi` | Create, list, and delete instant databases |
| `ConnectionsApi` | Manage connections and load managed tables |
| `WorkspacesApi` | List and create workspaces |
| `InformationSchemaApi` | List tables and columns |
| `QueryRunsApi` | Inspect query run history |
| `ResultsApi` | Retrieve stored query results |
| `UploadsApi` | Upload files for managed table loads |
| `IndexesApi` | Create and list indexes (BM25, vector) |
| `JobsApi` | Monitor background jobs |

## See also

- [hotdata on PyPI](https://pypi.org/project/hotdata)
- [sdk-python on GitHub](https://github.com/hotdata-dev/sdk-python)
- [Rust SDK](/docs/rust-sdk) — the same API from Rust
- [API Reference](/docs/api-reference) — Full HTTP API documentation
- [Quick Start](/docs/quick-start) — CLI and workspace setup