Important

Version v0.15.0 removes the legacy result format, query_legacy() and the QueryLegacy types. APL queries return tabular results and you read them with result.tables[0].events(). Read MIGRATING.md.

axiom-py

Install

pip install axiom-py

Synchronous Client

import axiom_py

client = axiom_py.Client()

client.ingest_events(dataset="DATASET_NAME", events=[{"foo": "bar"}, {"bar": "baz"}])
result = client.query(r"['DATASET_NAME'] | where foo == 'bar' | limit 100")
for row in result.tables[0].events():
    print(row)

Edge Ingestion

For improved data locality, you can configure the client to use regional edge endpoints for ingest and query operations. All other API operations continue to use the main Axiom API endpoint.

import axiom_py

# Using a regional edge domain
edge_client = axiom_py.Client(
    token="xaat-your-api-token",
    edge="eu-central-1.aws.edge.axiom.co"
)

# Or using an explicit edge URL
edge_client = axiom_py.Client(
    token="xaat-your-api-token",
    edge_url="https://custom-edge.example.com"
)

Note: Edge endpoints require API tokens (xaat-), not personal tokens. Edge configuration must be passed explicitly when creating the client.

Metrics Queries (MPL)

To query OTel metrics using MPL (Metrics Processing Language), configure the client with your edge endpoint and use mpl_query:

import axiom_py
from axiom_py import MplOptions
from datetime import datetime, timedelta, timezone

client = axiom_py.Client(
    token="xaat-your-api-token",
    edge="us-east-1.aws.edge.axiom.co"
)

end = datetime.now(timezone.utc)
start = end - timedelta(hours=1)

result = client.mpl_query(
    "`my-metrics`:`http.server.duration` | align to 5m using avg",
    opts=MplOptions(start_time=start, end_time=end),
)

for series in result.series:
    print(series.metric, series.tags, series.data)

Asynchronous Client

The library also provides an async client for use with asyncio:

import asyncio
from axiom_py import AsyncClient

async def main():
    async with AsyncClient() as client:
        # Ingest events
        await client.ingest_events(
            dataset="DATASET_NAME",
            events=[{"foo": "bar"}, {"bar": "baz"}]
        )

        # Query data
        result = await client.query(r"['DATASET_NAME'] | where foo == 'bar' | limit 100")
        for row in result.tables[0].events():
            print(row)

asyncio.run(main())

Concurrent Operations

The async client enables efficient concurrent operations:

import asyncio
from axiom_py import AsyncClient

async def main():
    async with AsyncClient() as client:
        # Ingest to multiple datasets concurrently
        await asyncio.gather(
            client.ingest_events("dataset1", [{"event": "data1"}]),
            client.ingest_events("dataset2", [{"event": "data2"}]),
            client.ingest_events("dataset3", [{"event": "data3"}]),
        )

asyncio.run(main())

Documentation

Read documentation on axiom.co/docs/guides/python.

License

MIT