Axiom Python SDK
axiom-py is the official Python SDK for interacting with the Axiom API, allowing users to ingest, query, and manage observability data at scale. It provides both synchronous and asynchronous clients. The current version is 0.10.0 and it maintains an active release cadence with ongoing development.
Warnings
- breaking Version v0.9.0 removed the 'aggregation operation enum'. If you were using this enum for defining aggregation operations, your code will break.
- gotcha When using edge endpoints for ingestion and query operations, you must use API tokens (prefixed with `xaat-`) and pass the edge configuration explicitly to the client constructor. Personal tokens are not supported for edge endpoints.
- gotcha Axiom client initialization requires an API token and optionally an organization ID. These can be passed directly to the constructor or set via `AXIOM_TOKEN` and `AXIOM_ORG_ID` environment variables. Failing to provide these will result in authentication errors.
Install
-
pip install axiom-py
Imports
- Client
from axiom_py import Client
- AsyncClient
from axiom_py import AsyncClient
- AxiomHandler
from axiom_py.logging import AxiomHandler
- AxiomProcessor
from axiom_py.structlog import AxiomProcessor
Quickstart
import os
from axiom_py import Client
import asyncio
AXIOM_TOKEN = os.environ.get('AXIOM_TOKEN', 'YOUR_AXIOM_TOKEN')
AXIOM_ORG_ID = os.environ.get('AXIOM_ORG_ID', 'YOUR_AXIOM_ORG_ID')
DATASET_NAME = os.environ.get('AXIOM_DATASET', 'your-dataset-name')
async def main():
if not AXIOM_TOKEN or not AXIOM_ORG_ID or not DATASET_NAME:
print("Please set AXIOM_TOKEN, AXIOM_ORG_ID, and AXIOM_DATASET environment variables or replace placeholders.")
return
client = Client(token=AXIOM_TOKEN, org_id=AXIOM_ORG_ID)
# Ingest events
try:
response = client.ingest_events(
dataset=DATASET_NAME,
events=[
{"service": "my-app", "level": "info", "message": "User logged in"},
{"service": "my-app", "level": "debug", "message": "Processing data"}
]
)
print(f"Ingest response: {response}")
except Exception as e:
print(f"Error during ingest: {e}")
# Query data
try:
query_result = client.query(f"['{DATASET_NAME}'] | limit 2")
print(f"Query result matches: {len(query_result.matches)}")
for match in query_result.matches:
print(match.data)
except Exception as e:
print(f"Error during query: {e}")
if __name__ == "__main__":
asyncio.run(main())