alloy-sdk 0.1.0

Python SDK for Alloy hosted SQL and Mesh Storage APIs

Alloy Python SDK

Public-ready Python SDK package for hosted Alloy APIs.

Install

Requires Python 3.12. The first public SDK release intentionally supports Python 3.12 only; the SDK quality workflow tests Linux and macOS on Python 3.12.

After the PyPI release is live:

pip install alloy-sdk

Before the PyPI release, install from this repository checkout for validation:

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 in v1. 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 in v1.

Hosted SQL

Hosted SQL v1 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 v1 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; v1 still 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 in v1. 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")

V1 Contract

• Supported: HTTPS POST /mesh/query, bearer API-key auth, Arrow IPC stream output.
• Supported: PyArrow-first results, with Pandas, Polars, and DuckDB helpers.
• Supported: named typed SQL value params via params={...}.
• Not supported in v1: 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.

Release Validation

The production release flow is documented in RELEASE.md. Public SDK releases are shipped by merging the bot-created release/sdk-vX.Y.Z -> release/sdk release PR. That merge publishes through PyPI Trusted Publishing after the protected pypi environment is approved.

Humans set only the release line in alloy/_version.py, such as 0.1. The release PR workflow stamps the generated branch with the next available patch version, such as 0.1.0 or 0.1.1.

Before marking a release candidate ready, run the local SDK quality gate from packages/python/alloy-sdk:

uv sync --all-extras --frozen
uv run ruff format --check .
uv run ruff check .
uv run ty check
uv run pip-audit --local --progress-spinner off
uv run pytest -q
rm -rf dist
uv build --out-dir dist
uv run python scripts/validate_release.py
uv run python scripts/validate_release_scope.py --base origin/main --head HEAD
uv run python scripts/validate_distribution.py dist
uv run twine check dist/*

Then smoke-test the built wheel in a clean environment:

smoke_venv=$(mktemp -d)
uv venv "$smoke_venv" --python 3.12
uv pip install --python "$smoke_venv/bin/python" dist/*.whl
"$smoke_venv/bin/python" scripts/smoke_installed_wheel.py
rm -rf "$smoke_venv"

For live release validation, set ALLOY_DATA_URL and ALLOY_API_KEY, then run:

uv run python scripts/live_smoke.py

The live smoke runs SQL plus sync and async Storage upload/list/download, exact-key conflict, and cross-prefix denial checks. It writes tiny objects under uploads/sdk-uploads/sdk-live-smoke/ and leaves them in place because SDK v1 intentionally has no delete helper.