Alloy Python SDK

Python SDK package for hosted Alloy APIs.

Install

Requires Python 3.12. The SDK quality workflow tests Linux and macOS on Python 3.12.

Install from PyPI:

pip install alloy-sdk

For local validation, install from this repository checkout:

uv pip install ./packages/python/alloy-sdk

Storage

Use alloy.storage to upload local files into Mesh Storage, list uploaded files, and download files by key. The SDK handles the transfer details.

from alloy import storage

with storage.connect() as store:
    upload = store.upload_folder(
        "local/run-001",
        path="flights/run-001",
        overwrite=False,
    )

print(upload.prefix)
# uploads/sdk-uploads/flights/run-001/

Single-file uploads use the same folder-style path:

from alloy import storage

with storage.connect() as store:
    store.upload_file("local/run-001/run.mcap", path="flights/run-001", overwrite=False)

path is a Mesh folder under uploads/sdk-uploads/; do not include uploads/, sdk-uploads/, a bucket name, or a trailing slash. If path is omitted, Alloy generates a dated folder such as 2026-06-24/<uuid>.

Read helpers accept either an SDK upload path or an existing Mesh prefix. Downloads use exact Mesh keys:

from alloy import storage

with storage.connect() as store:
    listing = store.list_files(path="flights/run-001")
    for file in listing:
        print(file.key, file.size)

    store.download_file(
        "uploads/sdk-uploads/flights/run-001/run.mcap",
        "run.mcap",
    )

Async clients mirror the sync methods:

from alloy import storage

async with storage.async_connect() as store:
    await store.upload_folder(
        "local/run-001",
        path="flights/run-001",
        overwrite=False,
    )

The SDK does not expose delete, move, rename, or overwrite operations. overwrite=True is rejected; delete/replace workflows must go through Mesh deletion semantics.

Storage Security Model

Storage helpers request short-lived, path-scoped R2 credentials from the Alloy data API and use them only for the transfer. Do not log UploadSession, ReadSession, UploadResult, or raw credential values, and do not hand temporary credentials to untrusted code. Upload sessions are scoped to uploads/sdk-uploads/<path>/; raw object-store credentials can still perform S3-compatible writes inside that temporary scope until expiry, so the SDK keeps overwrite=False fixed and does not expose delete, move, rename, or overwrite helpers.

Hosted SQL

Hosted SQL has one public transport: HTTPS POST /mesh/query returning Apache Arrow IPC stream bytes. The SDK keeps the public API small:

• fetch* for small row-oriented results.
• fetch_rows for capped Python rowsets with columns, CSV output, and truncation metadata.
• query for an Arrow-backed result object.
• query_arrow, query_pandas, query_polars, and query_df for direct formats.
• stream for batch iteration.

Connect

The SDK reads ALLOY_DATA_URL and ALLOY_API_KEY.

export ALLOY_DATA_URL="https://data.usealloy.ai"
export ALLOY_API_KEY="ak_..."

from alloy import sql

with sql.connect() as db:
    count = db.fetchval("SELECT count(*) FROM alloy.fleet.diagnostics")

Pass explicit values when you do not want environment-based config:

from alloy import sql

with sql.connect(
    base_url="https://data.usealloy.ai",
    api_key="ak_...",
) as db:
    rows = db.fetch(
        """
        SELECT topic, count(*) AS n
        FROM alloy.fleet.diagnostics
        GROUP BY topic
        """
    )

Use the data API host provided by Alloy for your environment. Common hosts are https://data.usealloy.ai for production, https://data-adnav.usealloy.ai for Adnav production, and https://data-dev.usealloy.ai for development testing.

Small Row Results

Use fetch, fetchrow, and fetchval when you want ordinary Python values for a bounded result set. This is the most compact shape for scripts, dashboards, backend routes, and MCP tools.

from alloy import sql

p = sql.param

with sql.connect() as db:
    count = db.fetchval("SELECT count(*) FROM alloy.fleet.diagnostics")

    latest = db.fetchrow(
        """
        SELECT d.topic, d.created_at
        FROM alloy.fleet.diagnostics AS d
        JOIN alloy.mesh.file_meta AS fm
          ON fm.file_id = d.file_id
         AND fm.key = 'alloy.mission_id'
        WHERE fm.value = $mission_id
        ORDER BY d.created_at DESC
        LIMIT 1
        """,
        params={"mission_id": p.utf8("2U8NUGFHGifdCt7tdcnwTd")},
    )

    rows = db.fetch(
        """
        SELECT d.topic, count(*) AS n
        FROM alloy.fleet.diagnostics AS d
        JOIN alloy.mesh.file_meta AS fm
          ON fm.file_id = d.file_id
         AND fm.key = 'alloy.mission_id'
        WHERE fm.value = $mission_id
        GROUP BY d.topic
        ORDER BY n DESC
        LIMIT 100
        """,
        params={"mission_id": p.utf8("2U8NUGFHGifdCt7tdcnwTd")},
    )

Use fetch_rows when the caller needs result metadata, CSV output, or an explicit client-side cap in one object:

from alloy import sql

with sql.connect() as db:
    rowset = db.fetch_rows(
        """
        SELECT topic, created_at, payload
        FROM alloy.fleet.diagnostics
        ORDER BY created_at DESC
        LIMIT 1000
        """,
        max_rows=1000,
    )

print(rowset.columns)
print(rowset.row_count)
print(rowset.to_csv())

if rowset.truncated:
    print("Result was capped locally; add or tighten LIMIT in SQL.")

max_rows is a client materialization cap. It avoids building unbounded Python row lists, but it does not rewrite the SQL, reduce hosted SQL work, or reduce the bytes the endpoint sends. Use SQL LIMIT for performance and max_rows as a final guardrail at the SDK boundary.

Arrow-Backed Results

Use query when you want metadata plus flexible conversion. The sync result is script-friendly and exposes ClickHouse-style row helpers.

from alloy import sql

with sql.connect() as db:
    result = db.query("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")

result.column_names
result.column_types
result.row_count
result.result_rows
result.first_row
result.first_item

for row in result.named_results():
    print(row["topic"])

arrow_table = result.to_arrow()
pandas_df = result.to_pandas()
polars_df = result.to_polars()
duckdb_conn = result.to_duckdb(table_name="diagnostics")

Direct format helpers are shortcuts over the same query path:

with sql.connect() as db:
    table = db.query_arrow("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")
    df = db.query_df("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")
    pandas_df = db.query_pandas("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")
    polars_df = db.query_polars("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")

Async Usage

The async client mirrors the sync client. Use the same method names with await. Async HTTP is native, Arrow IPC decode runs off the event loop, and expensive result materialization is async-only.

from alloy import sql

async with sql.async_connect() as db:
    count = await db.fetchval("SELECT count(*) FROM alloy.fleet.diagnostics")

    latest = await db.fetchrow(
        """
        SELECT fm.value AS mission_id, d.topic, d.created_at
        FROM alloy.fleet.diagnostics AS d
        LEFT JOIN alloy.mesh.file_meta AS fm
          ON fm.file_id = d.file_id
         AND fm.key = 'alloy.mission_id'
        ORDER BY d.created_at DESC
        LIMIT 1
        """
    )

    rows = await db.fetch("SELECT topic FROM alloy.fleet.diagnostics LIMIT 100")

    rowset = await db.fetch_rows(
        "SELECT topic, created_at FROM alloy.fleet.diagnostics LIMIT 1000",
        max_rows=1000,
    )
    csv_text = await rowset.to_csv()
    if rowset.truncated:
        raise RuntimeError("Hosted SQL result exceeded the local row cap")

    result = await db.query("SELECT * FROM alloy.fleet.diagnostics LIMIT 100")
    table = result.to_arrow()
    df = await result.to_pandas()
    polars_df = await result.to_polars()

Streaming Batches

Use stream for batch-oriented reads. The sync stream decodes directly from the HTTP response. The async stream uses async HTTP and decodes Arrow IPC batches in a worker thread; the stream reads the response body before decoding batches.

from alloy import sql

with sql.connect() as db:
    with db.stream("SELECT * FROM alloy.fleet.diagnostics") as stream:
        for batch in stream:
            print(batch.num_rows)

from alloy import sql

async with sql.async_connect() as db:
    async with db.stream("SELECT * FROM alloy.fleet.diagnostics") as stream:
        async with stream.reader() as reader:
            print(reader.schema.names)
            async for batch in reader:
                print(batch.num_rows)

Params

Use named typed params instead of formatting values into SQL strings. Params are values only: not table names, column names, identifiers, clauses, or SQL fragments. The SDK never interpolates params into SQL client-side.

from alloy import sql

p = sql.param

with sql.connect() as db:
    rows = db.fetch(
        """
        SELECT d.created_at, d.payload
        FROM alloy.fleet.diagnostics AS d
        JOIN alloy.mesh.file_meta AS fm
          ON fm.file_id = d.file_id
         AND fm.key = 'alloy.mission_id'
        WHERE fm.value = $mission_id
          AND d.created_at >= $since
          AND d.score >= $min_score
        ORDER BY d.created_at DESC
        LIMIT 100
        """,
        params={
            "mission_id": p.utf8("2U8NUGFHGifdCt7tdcnwTd"),
            "since": p.timestamp_us("2026-06-01T00:00:00Z"),
            "min_score": p.float64(0.8),
        },
    )

Params are named only: SQL uses $mission_id, and the params key is "mission_id". Positional placeholders like $1 are not supported. Supported types are utf8, bool, int64, float64, date32, timestamp_us, and binary. int64 is encoded as a decimal string on the wire so values are not rounded by JavaScript gateways; use sql.param.int64(...) instead of hand-writing that payload.

Local DuckDB

Local DuckDB helpers run over the downloaded Arrow result. They do not push work back to hosted SQL.

with sql.connect() as db:
    result = db.query(
        """
        SELECT topic, count(*) AS n
        FROM alloy.fleet.diagnostics
        GROUP BY topic
        """
    )
    summary = result.local_sql("SELECT sum(n) FROM result")

async with sql.async_connect() as db:
    result = await db.query(
        """
        SELECT topic, count(*) AS n
        FROM alloy.fleet.diagnostics
        GROUP BY topic
        """
    )
    summary = await result.local_sql("SELECT sum(n) FROM result")

Hosted SQL Limits

Hosted SQL uses HTTPS POST /mesh/query, bearer API-key auth, Arrow IPC stream output, and named typed SQL value params via params={...}.

The SDK does not expose public Flight SQL/gRPC, Postgres wire protocol, ODBC/JDBC BI plug-and-play, cursors, sessions, or stream resume.

• If a stream fails partway through, rerun the query.
• Pointer-follow is for projected payload fields. Filtering on pointered payload fields is not a performance contract.
• Nested filters are evaluated correctly, but arbitrary nested predicate pushdown is separate performance hardening work.