pyamselect

Python client library and CLI for AirMettle Select. Run SQL queries directly against data in your Azure Blob Storage — CSV, JSON, and Parquet — without moving or ingesting it.

• pyamselect
  • Installation
  • Quick Start
    • 1. Prepare Your Data
    • 2. Run a Query
    • 3. Query from Python
  • Building Requests
    • Dataclasses vs Raw Dicts
    • CSV Convenience Factory
    • JSON and Parquet Input
    • Request JSON Format
  • Client API
    • Sync Client
      • Streaming Events
    • Async Client
      • Async Streaming
    • Configuration
      • Timeouts
      • Error Handling
  • CLI Reference
    • query
    • prepare
    • Common Options
  • License

Installation

pip install pyamselect

Requires Python 3.10+. Dependencies (httpx[http2], crc32c) are installed automatically.

Verify the install:

amselect --version

Quick Start

You need the following, provided with your AirMettle Select subscription:

• Your subscription ID and API key
• The query endpoint (host and port) and the metadata service endpoint (URL)
• Data in an Azure storage account that is registered under your subscription

1. Prepare Your Data

Before an object can be queried, the service generates query metadata for it. Run this once per object:

amselect prepare \
    --endpoint "https://<metadata-endpoint>" \
    --blob-url "https://<account>.blob.core.windows.net/<container>/<blob>" \
    --subscription-id "<subscription-id>" \
    --api-key "<api-key>"

The service accesses the blob with the access key stored in the storage account's bucket registration, so the storage account must be registered under your subscription first.

The command prints a JSON result and waits (up to 5 minutes) for the operation to complete. Pass --overwrite to regenerate metadata after an object has changed, or --no-wait to return as soon as an asynchronous prepare is accepted.

2. Run a Query

amselect query -H "<query-host>" -r '{
  "select_request": {
    "expression": "SELECT * FROM Object WHERE age > 30",
    "container": "<container>",
    "blob": "<blob>",
    "storage_account": "<account>",
    "subscription_id": "<subscription-id>",
    "api_key": "<api-key>",
    "input_options": {"type": "csv", "value": {"csv_header_config": "use"}},
    "output_options": {"type": "csv", "value": {"recordDelimiter": "\n", "fieldDelimiter": ","}}
  }
}'

Results stream to stdout; pass -o results.csv to write them to a file instead. Larger requests can be kept in a file and passed with --request-file request.json.

3. Query from Python

from pyamselect import AMSelectClient, SelectRequest, CSVInputOptions, CSVOutputOptions

request = SelectRequest(
    expression="SELECT * FROM Object WHERE age > 30",
    container="<container>",
    blob="<blob>",
    storage_account="<account>",
    subscription_id="<subscription-id>",
    api_key="<api-key>",
    input_options=CSVInputOptions(csv_header_config="use", csv_field_delimiter=","),
    output_options=CSVOutputOptions(),
)

with AMSelectClient("<query-host>") as client:
    result = client.select_to_string(request)
    print(result)

An async client with the same API is also available — see Async Client.

Building Requests

Dataclasses vs Raw Dicts

Input/output options can be passed as typed dataclasses (IDE autocomplete, validation) or raw dicts (the same JSON structure sent on the wire).

# Dataclass
from pyamselect import CSVInputOptions
input_options = CSVInputOptions(csv_header_config="use", csv_field_delimiter=",")

# Equivalent raw dict
input_options = {"type": "csv", "value": {"csv_header_config": "use", "csv_field_delimiter": ","}}

CSV Convenience Factory

request = SelectRequest.csv_query(
    expression="SELECT name, age FROM Object",
    container="mycontainer",
    blob="people.csv",
    storage_account="myaccount",
    subscription_id="sub-123",
    api_key="key-456",
)

JSON and Parquet Input

from pyamselect import JSONInputOptions, JSONOutputOptions, ParquetInputOptions, CSVOutputOptions

# JSON lines
request = SelectRequest(
    expression="""SELECT t."Timestamp", t.devname FROM Object as t WHERE t.windows_event_id='4624'""",
    container="mycontainer",
    blob="events.jsonl",
    storage_account="myaccount",
    input_options=JSONInputOptions(json_type="lines"),
    output_options=JSONOutputOptions(record_delimiter="\n"),
)

# Parquet
request = SelectRequest(
    expression="SELECT col1, col2 FROM Object",
    container="mycontainer",
    blob="data.parquet",
    storage_account="myaccount",
    input_options=ParquetInputOptions(),
    output_options=CSVOutputOptions(),
)

Request JSON Format

The request JSON sent to the service has the following format:

{
  "select_request": {
    "expression": "SELECT * FROM Object",
    "expression_type": "sql",
    "container": "mycontainer",
    "blob": "myfile.csv",
    "storage_account": "myaccount",
    "subscription_id": "sub-123",
    "api_key": "key-456",
    "input_options": {
      "type": "csv",
      "value": { "csv_header_config": "use", "csv_field_delimiter": "," }
    },
    "output_options": {
      "type": "csv",
      "value": { "recordDelimiter": "\n", "fieldDelimiter": "," }
    }
  }
}

The select_request envelope is optional when using SelectRequest.from_json() -- bare inner objects are accepted too.

Client API

Both clients share the same methods: select(), select_to_string(), select_to_file(), and select_to_stream().

Sync Client

with AMSelectClient("<query-host>") as client:
    # Collect all data as a string
    result = client.select_to_string(request)

    # Write directly to a file
    client.select_to_file(request, "output.csv")

    # Write to any writable binary stream
    client.select_to_stream(request, stream)

Streaming Events

For full control over the event stream (data, stats, continue events):

from pyamselect import AMSelectClient, EventType

with AMSelectClient("<query-host>") as client:
    for event in client.select(request):
        if event.event_type == EventType.DATA:
            print(event.payload.decode("utf-8"), end="")
        elif event.event_type == EventType.STATS:
            print(f"Stats: {event.payload}")
        elif event.event_type == EventType.END:
            print("Done.")

Async Client

import asyncio
from pyamselect import AsyncAMSelectClient

async def main():
    async with AsyncAMSelectClient("<query-host>") as client:
        result = await client.select_to_string(request)

        await client.select_to_file(request, "output.csv")

        await client.select_to_stream(request, stream)

asyncio.run(main())

Async Streaming

import asyncio
from pyamselect import AsyncAMSelectClient, EventType

async def main():
    async with AsyncAMSelectClient("<query-host>") as client:
        async for event in client.select(request):
            if event.event_type == EventType.DATA:
                print(event.payload.decode("utf-8"), end="")

asyncio.run(main())

Configuration

These options apply to both AMSelectClient and AsyncAMSelectClient.

Timeouts

with AMSelectClient("<query-host>", connect_timeout=10.0, read_timeout=300.0) as client:
    result = client.select_to_string(request)

Error Handling

from pyamselect import AMSelectHTTPError, AMSelectStreamError, AMSelectConnectionError

with AMSelectClient("<query-host>") as client:
    try:
        result = client.select_to_string(request)
    except AMSelectConnectionError as e:
        print(f"Connection failed: {e}")
    except AMSelectHTTPError as e:
        print(f"HTTP {e.status_code}: {e.body}")
    except AMSelectStreamError as e:
        print(f"Query error {e.error_code}: {e.error_message}")

CLI Reference

Installing the package provides the amselect command with two subcommands: query and prepare. Query options can also be passed directly to amselect without the query subcommand. All commands exit non-zero on failure.

amselect --version     # Show version
amselect --buildinfo   # Show detailed build information

query

Runs a select request against the query endpoint and streams the results.

# Inline request (results to stdout)
amselect query -H "<query-host>" -r '{"select_request": {...}}'

# Request from a file
amselect query -H "<query-host>" --request-file request.json

# Write results to a file
amselect query -H "<query-host>" --request-file request.json -o results.csv

-H, --host              Query endpoint host (required)
-P, --port              Query endpoint port (default: 443)
-r, --request           JSON request string
    --request-file      Path to JSON request file
-o, --output            Output file path (default: stdout)

prepare

Generates the query metadata for an object via the metadata service. An object must be prepared before it can be queried, and re-prepared (with --overwrite) after its contents change.

amselect prepare \
    --endpoint "https://<metadata-endpoint>" \
    --blob-url "https://<account>.blob.core.windows.net/<container>/<blob>" \
    --subscription-id "<subscription-id>" \
    --api-key "<api-key>"

The service accesses the blob with the access key stored in the storage account's bucket registration; requests for storage accounts not registered under the subscription are rejected. The result is printed as JSON. By default the command waits (up to 5 minutes) for an asynchronous prepare to finish.

    --endpoint          Metadata service base URL (required)
    --blob-url          Full URL of the blob to prepare
    --subscription-id   Subscription ID (required with --blob-url)
    --api-key           Subscription API key (required with --blob-url)
-r, --request           JSON request string (alternative to --blob-url)
    --request-file      Path to JSON request file (alternative to --blob-url)
    --overwrite         Regenerate metadata if it already exists
    --no-wait           Return as soon as an asynchronous prepare is accepted

Common Options

Available on both subcommands:

    --ca-cert           Path to CA certificate bundle
    --insecure-skip-tls-verify
                        Disable TLS certificate verification
    --connect-timeout   Connection timeout in seconds (default: 60)
    --read-timeout      Read timeout in seconds (default: 60)
-v, --verbose           Increase verbosity (-v=INFO, -vv=DEBUG)

License

MIT — see the LICENSE file. Use of the AirMettle Select service itself is governed by your service agreement.