### Install axiom-py

Source: https://github.com/axiomhq/axiom-py/blob/main/README.md

Install the axiom-py library using pip. This is the first step before using any of its features.

```sh
pip install axiom-py
```

--------------------------------

### Asynchronous Client Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Example of initializing and using the asynchronous Axiom client within an async context manager.

```python
from axiom_py import AsyncClient

async with AsyncClient(token="xaat-your-token") as client:
    # Use client here
    pass
```

--------------------------------

### Complete Example: Axiom Python Annotations

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/annotations.md

This example demonstrates the full lifecycle of an annotation using the Axiom Python SDK. It covers creating, listing, updating, retrieving, and deleting annotations. Ensure you have authenticated with your Axiom token.

```python
import axiom_py
from axiom_py import AnnotationCreateRequest, AnnotationUpdateRequest
from datetime import datetime, timedelta, timezone

client = axiom_py.Client(token="xaat-your-token")

# Create a deployment annotation
deploy_time = datetime.now(timezone.utc)
deploy_req = AnnotationCreateRequest(
    datasets=["production-logs", "production-metrics"],
    time=deploy_time,
    title="Production Release v1.5.0",
    description="Released new user authentication feature",
    url="https://github.com/org/repo/releases/v1.5.0",
    type="deployment"
)

annotation = client.annotations.create(deploy_req)
print(f"Created annotation: {annotation.id}")

# Get annotations from the last 7 days
start = datetime.now(timezone.utc) - timedelta(days=7)
end = datetime.now(timezone.utc)
recent = client.annotations.list(
    datasets=["production-logs"],
    start=start,
    end=end
)

print(f"Found {len(recent)} annotations in the last week:")
for ann in recent:
    print(f"  {ann.time}: {ann.title} ({ann.type})")

# Update the annotation with additional info
update_req = AnnotationUpdateRequest(
    description="Released new user authentication feature. Rollback procedure available at wiki/rollback/v1.5.0"
)

updated = client.annotations.update(annotation.id, update_req)
print(f"Updated: {updated.description}")

# Later: retrieve the annotation
retrieved = client.annotations.get(annotation.id)
print(f"Retrieved: {retrieved.title}")

# Finally: delete the annotation
client.annotations.delete(annotation.id)
```

--------------------------------

### Complete Dataset Operations with Axiom Py

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

This example demonstrates a full lifecycle of dataset operations including creation, listing, retrieval, updating, trimming old data, and deletion. Ensure you have your Axiom token and have installed the axiom-py library.

```python
import axiom_py
from datetime import timedelta

# Create a client
client = axiom_py.Client(token="xaat-your-token")

# Create a new dataset
new_ds = client.datasets.create(
    name="production-logs",
    description="Logs from production environment"
)
print(f"Created: {new_ds.id}")

# List all datasets
all_datasets = client.datasets.get_list()
print(f"Total datasets: {len(all_datasets)}")

# Get a specific dataset
ds = client.datasets.get(new_ds.id)
print(f"Dataset: {ds.name}")

# Update the description
updated = client.datasets.update(
    new_ds.id,
    "Updated production logs dataset"
)

# Trim old data (keep only last 30 days)
client.datasets.trim(new_ds.id, timedelta(days=30))

# Delete the dataset
client.datasets.delete(new_ds.id)
```

--------------------------------

### Synchronous Client Examples

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Demonstrates various ways to initialize the synchronous Axiom client, including using environment variables, explicit credentials, and custom endpoints.

```python
import axiom_py

# Using environment variables (AXIOM_TOKEN, AXIOM_ORG_ID)
client = axiom_py.Client()
```

```python
# Explicit credentials
client = axiom_py.Client(
    token="xaat-your-api-token",
    org_id="org-123"
)
```

```python
# With edge endpoint
client = axiom_py.Client(
    token="xaat-your-api-token",
    edge="eu-central-1.aws.edge.axiom.co"
)
```

```python
# With full edge URL
client = axiom_py.Client(
    token="xaat-your-api-token",
    edge_url="https://eu-central-1.aws.edge.axiom.co"
)
```

```python
# Custom API endpoint
client = axiom_py.Client(
    token="xaat-your-api-token",
    url="https://custom-axiom-api.example.com"
)
```

--------------------------------

### AsyncAxiomHandler Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Demonstrates how to set up and use the AsyncAxiomHandler for sending logs asynchronously to Axiom.

```APIDOC
## AsyncAxiomHandler Example

### Description
This example shows the complete setup for using the `AsyncAxiomHandler` to send logs asynchronously to Axiom. It includes creating an `AsyncClient`, configuring the handler with a dataset and flush interval, attaching it to a logger, and logging messages.

### Usage
```python
import asyncio
import logging
from axiom_py import AsyncClient, AsyncAxiomHandler

async def main():
    # Create async client
    async with AsyncClient(token="xaat-your-token") as client:
        # Create and add async handler
        handler = AsyncAxiomHandler(
            client=client,
            dataset="async-app-logs",
            interval=2  # Flush every 2 seconds
        )
        
        # Configure logger
        logger = logging.getLogger("myapp")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        
        # Log some messages
        logger.debug("Application initialized")
        logger.info("Processing started")
        
        # Do some work...
        await asyncio.sleep(1)
        
        logger.info("Processing completed")
        
        # Explicitly flush and close handler
        await handler.close()

asyncio.run(main())
```
```

--------------------------------

### Install Pre-commit Hooks

Source: https://github.com/axiomhq/axiom-py/blob/main/CLAUDE.md

Set up pre-commit hooks for the project. These hooks run automatically before each commit to ensure code quality.

```bash
# Install pre-commit hooks
uv run pre-commit install
```

--------------------------------

### MPL Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Executes a query using Metrics Processing Language (MPL) for OTel metrics. Requires specifying start and end times and potentially an edge endpoint.

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

--------------------------------

### Complete Axiom Client Configuration Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Instantiate the Axiom client with full configuration, set up a logging handler, and configure ingest and query options. Use these configured components for ingesting events and performing queries.

```python
import logging
import axiom_py
from axiom_py import AplOptions, IngestOptions
from datetime import datetime, timedelta, timezone

# Create client with full configuration
client = axiom_py.Client(
    token="xaat-your-api-token",
    org_id="your-org-id",
    url="https://api.axiom.co",
    edge="eu-central-1.aws.edge.axiom.co"
)

# Configure logging handler
handler = axiom_py.AxiomHandler(
    client=client,
    dataset="application-logs",
    interval=2  # Flush every 2 seconds
)

logger = logging.getLogger()
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Configure ingest options
ingest_opts = IngestOptions(
    timestamp_field="ts",
    timestamp_format="2006-01-02T15:04:05Z07:00"
)

# Configure query options
query_opts = AplOptions(
    start_time=datetime.now(timezone.utc) - timedelta(hours=24),
    end_time=datetime.now(timezone.utc),
    limit=1000
)

# Use configured components
status = client.ingest_events(
    dataset="events",
    events=[{"ts": "2024-01-01T00:00:00Z", "data": "value"}],
    opts=ingest_opts
)

result = client.query("['events'] | limit 100", opts=query_opts)
logger.info(f"Ingested {status.ingested} events")
```

--------------------------------

### Quick Example: Ingest and Query Events

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Demonstrates basic usage of the Axiom Python client for ingesting events into a dataset and then querying those events using APL.

```python
import axiom_py

# Create client
client = axiom_py.Client()

# Ingest events
status = client.ingest_events(
    dataset="my-dataset",
    events=[{"message": "test", "level": "info"}]
)

# Query data
result = client.query("['my-dataset'] | where level == 'info'")
print(f"Found {len(result.matches)} matches")
```

--------------------------------

### Async Logging Handler Setup

Source: https://github.com/axiomhq/axiom-py/blob/main/features/ASYNC_IMPLEMENTATION.md

Illustrates how to set up and use the `AsyncAxiomHandler` for asynchronous logging. Remember to close the handler to ensure all logs are flushed before exiting.

```python
import asyncio
import logging
from axiom_py import AsyncClient, AsyncAxiomHandler

async def main():
    async with AsyncClient() as client:
        handler = AsyncAxiomHandler(client, "logs")
        logger = logging.getLogger()
        logger.addHandler(handler)

        logger.info("This log will be sent to Axiom")

        # Important: flush before exit
        await handler.close()

asyncio.run(main())
```

--------------------------------

### Complete User Information Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/users.md

Demonstrates how to initialize the Axiom client with a personal token and retrieve comprehensive user information, including name, email, role, user ID, and role ID.

```python
import axiom_py

# Create client with personal token
client = axiom_py.Client(token="xapt-your-personal-token")

# Get current user info
user = client.users.current()

if user:
    print(f"Name: {user.name}")
    print(f"Email: {user.email}")
    print(f"Role: {user.role.name}")
    print(f"User ID: {user.id}")
    print(f"Role ID: {user.role.id}")
else:
    print("Token is an organization token, not a personal token")
```

--------------------------------

### Complete Token Management Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/tokens.md

Demonstrates the full lifecycle of token management: creating a token with specific dataset ingest permissions, listing all tokens, retrieving token details, regenerating an existing token with new expiry dates, and finally deleting the token.

```python
import axiom_py
from axiom_py.tokens import (
    CreateTokenRequest,
    TokenOrganizationCapabilities,
    TokenDatasetCapabilities,
    RegenerateTokenRequest
)
from datetime import datetime, timedelta, timezone

client = axiom_py.Client(token="xaat-existing-token")

# Create a token with limited dataset permissions
req = CreateTokenRequest(
    name="app-ingest-token",
    description="Read-only token for application event ingestion",
    datasetCapabilities={
        "production-logs": TokenDatasetCapabilities(
            ingest=["create"],  # Can only ingest
        ),
    }
)

new_token = client.tokens.create(req)
print(f"Token created: {new_token.token}")
print(f"Token ID: {new_token.id}")

# List all tokens
all_tokens = client.tokens.list()
print(f"Total tokens: {len(all_tokens)}")

# Get token details
token_info = client.tokens.get(new_token.id)
print(f"Token name: {token_info.name}")

# Regenerate a token
regen_req = RegenerateTokenRequest(
    existingTokenExpiresAt=datetime.now(timezone.utc),
    newTokenExpiresAt=datetime.now(timezone.utc) + timedelta(days=90)
)

regenerated = client.tokens.regenerate(new_token.id, regen_req)
print(f"Token regenerated: {regenerated.id}")

# Delete a token
client.tokens.delete(new_token.id)
print("Token deleted")
```

--------------------------------

### Async Axiom Handler Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Demonstrates setting up an asynchronous Axiom handler with the standard logging library. Configure the logger, add the handler, and log messages. Ensure to explicitly close the handler for proper flushing.

```python
import asyncio
import logging
from axiom_py import AsyncClient, AsyncAxiomHandler

async def main():
    # Create async client
    async with AsyncClient(token="xaat-your-token") as client:
        # Create and add async handler
        handler = AsyncAxiomHandler(
            client=client,
            dataset="async-app-logs",
            interval=2  # Flush every 2 seconds
        )
        
        # Configure logger
        logger = logging.getLogger("myapp")
        logger.addHandler(handler)
        logger.setLevel(logging.DEBUG)
        
        # Log some messages
        logger.debug("Application initialized")
        logger.info("Processing started")
        
        # Do some work...
        await asyncio.sleep(1)
        
        logger.info("Processing completed")
        
        # Explicitly flush and close handler
        await handler.close()

asyncio.run(main())
```

--------------------------------

### Async Axiom Processor Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Demonstrates how to set up and use the AsyncAxiomProcessor with structlog for asynchronous, structured logging. This includes configuring the client, processor, and logger, logging events, and manually flushing and closing the processor.

```python
import asyncio
import structlog
from axiom_py import AsyncClient, AsyncAxiomProcessor

async def main():
    async with AsyncClient(token="xaat-your-token") as client:
        # Create async processor
        processor = AsyncAxiomProcessor(
            client=client,
            dataset="async-structured-logs",
            interval=1
        )
        
        # Configure structlog
        structlog.configure(
            processors=[
                processor,
                structlog.processors.JSONRenderer(),
            ],
            logger_factory=structlog.PrintLoggerFactory(),
        )
        
        log = structlog.get_logger()
        
        # Log events (buffered, async flushed)
        log.info("request_received", method="POST", path="/api/data")
        log.info("request_completed", status=200, duration_ms=125)
        
        # Do async work...
        await asyncio.sleep(1)
        
        # Flush logs
        await processor.flush()
        
        # Close processor
        await processor.close()

asyncio.run(main())
```

--------------------------------

### Manage Datasets with Axiom Python Client

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/client.md

Perform CRUD operations on datasets using the Axiom Python client. Examples include getting, listing, creating, updating, and deleting datasets.

```python
# Get a dataset
ds = client.datasets.get("dataset-id")

# List all datasets
all_ds = client.datasets.get_list()

# Create a dataset
new_ds = client.datasets.create(
    name="my-dataset",
    description="Dataset for application logs"
)

# Update dataset description
updated = client.datasets.update("dataset-id", "New description")

# Delete a dataset
client.datasets.delete("dataset-id")
```

--------------------------------

### MPL Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Demonstrates how to perform an MPL (Metrics Processing Language) query asynchronously using the Axiom AsyncClient. This is suitable for querying time-series metrics data.

```python
from axiom_py import AsyncClient, MplOptions
from datetime import datetime, timedelta, timezone
import asyncio

async def main():
    async with AsyncClient(edge="us-east-1.aws.edge.axiom.co") as client:
        end = datetime.now(timezone.utc)
        start = end - timedelta(hours=1)
        
        result = await client.mpl_query(
            "`otel-metrics`:`http.server.duration` | align to 1m using avg",
            opts=MplOptions(start_time=start, end_time=end)
        )
        
        for series in result.series:
            print(f"Metric: {series.metric}")
            print(f"Tags: {series.tags}")
            print(f"Data: {series.data}")

asyncio.run(main())
```

--------------------------------

### APL Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Demonstrates how to perform a simple APL query and an APL query with a specified time range using the Axiom Python client. Ensure you have your authentication token set.

```python
from axiom_py import Client, AplOptions
from datetime import datetime, timedelta, timezone

client = Client(token="xaat-your-token")

# Simple APL query
result = client.query(
    "['my-dataset'] | where level == 'error' | limit 100"
)

# APL query with time range
end = datetime.now(timezone.utc)
start = end - timedelta(hours=24)

result = client.query(
    "['my-dataset'] | stats count() by service | limit 50",
    opts=AplOptions(
        start_time=start,
        end_time=end,
        limit=50
    )
)

print(f"Status: {result.status.elapsedTime}ms")
for entry in result.matches or []:
    print(entry.data)
```

--------------------------------

### Manage Annotations with Axiom Python

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Create, list, get, update, and delete annotations. Ensure the necessary import statements are included.

```python
from axiom_py import AnnotationCreateRequest, AnnotationUpdateRequest
from datetime import datetime, timezone

# Create
req = AnnotationCreateRequest(
    datasets=["dataset"],
    time=datetime.now(timezone.utc),
    title="Deployment v1.0",
    type="deployment"
)
ann = client.annotations.create(req)

```

```python
# List
anns = client.annotations.list(datasets=["dataset"])

```

```python
# Get
ann = client.annotations.get("annotation-id")

```

```python
# Update
upd = AnnotationUpdateRequest(title="Updated title")
ann = client.annotations.update("annotation-id", upd)

```

```python
# Delete
client.annotations.delete("annotation-id")

```

--------------------------------

### Asynchronously List All Datasets

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Retrieve a list of all available datasets asynchronously. This is useful for getting an overview of your data.

```python
async with AsyncClient() as client:
    datasets = await client.datasets.get_list()
```

--------------------------------

### Legacy Structured Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Constructs and executes a query using the legacy structured query format with time range and filters. Requires importing specific query classes.

```python
from axiom_py.query import QueryLegacy, Filter, FilterOperation

query = QueryLegacy(
    startTime=start,
    endTime=end,
    filter=Filter(op=FilterOperation.EQUAL, field="level", value="error"),
    limit=100
)
result = client.query_legacy("dataset-id", query, opts)
```

--------------------------------

### Manage Tokens with Axiom Python

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

List, create, get, and delete API tokens. The actual token string is available in the `token.token` attribute after creation.

```python
from axiom_py.tokens import CreateTokenRequest, TokenOrganizationCapabilities

# List
tokens = client.tokens.list()

```

```python
# Create
req = CreateTokenRequest(
    name="token-name",
    description="optional"
)
token = client.tokens.create(req)
print(token.token)  # The actual token string

```

```python
# Get
token = client.tokens.get("token-id")

```

```python
# Delete
client.tokens.delete("token-id")

```

--------------------------------

### Configure structlog with AxiomProcessor

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Set up structlog to send logs to Axiom using AxiomProcessor. This example configures the processor to send to a specific dataset and also renders logs as JSON locally. Ensure you have a valid Axiom token and dataset name.

```python
import structlog
import axiom_py

# Create client
client = axiom_py.Client(token="xaat-your-token")

# Create processor
processor = axiom_py.AxiomProcessor(
    client=client,
    dataset="structured-logs",
    interval=2
)

# Configure structlog
structlog.configure(
    processors=[
        processor,  # Send to Axiom
        structlog.processors.JSONRenderer(),  # Also output JSON locally
    ],
    logger_factory=structlog.PrintLoggerFactory(),
)

log = structlog.get_logger()

# Log structured events
log.info("user_login", user_id=123, ip="192.168.1.1")
log.warning("rate_limit_exceeded", endpoint="/api/data", limit=1000)
log.error("database_error", code="CONN_TIMEOUT", retry_count=3)
```

--------------------------------

### MplOptions

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Options for configuring MPL (Metrics Processing Language) queries, including required start and end times, and optional parameters.

```APIDOC
## MplOptions

Options for MPL (Metrics Processing Language) queries.

```python
@dataclass
class MplOptions:
    start_time: datetime                    # Query start time (required)
    end_time: datetime                      # Query end time (required)
    params: Optional[Dict[str, str]] = None # Additional query parameters
    query_options: Optional[Dict[str, str]] = None  # API-specific options
    nocache: bool = False                   # Bypass query cache
```

**Fields:**

| Field | Type | Description |
|-------|------|-------------|
| `start_time` | `datetime` | Required start time for the query range. |
| `end_time` | `datetime` | Required end time for the query range. |
| `params` | `Dict[str, str]` | Additional parameters passed to MetricsDB. |
| `query_options` | `Dict[str, str]` | Query options forwarded to the API payload. |
| `nocache` | `bool` | If True, bypass the query cache. |
```

--------------------------------

### MplOptions for MPL Queries

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Configure MPL queries with required start and end times, and optional parameters for MetricsDB or API query options. Use `nocache=True` to bypass the cache.

```python
from dataclasses import dataclass
from typing import Optional, Dict
from datetime import datetime

@dataclass
class MplOptions:
    start_time: datetime                    # Query start time (required)
    end_time: datetime                      # Query end time (required)
    params: Optional[Dict[str, str]] = None # Additional MetricsDB parameters
    query_options: Optional[Dict[str, str]] = None  # API query options
    nocache: bool = False                   # Bypass cache
```

```python
from datetime import datetime, timedelta, timezone

opts = MplOptions(
    start_time=datetime.now(timezone.utc) - timedelta(hours=1),
    end_time=datetime.now(timezone.utc),
    nocache=True
)

result = client.mpl_query(
    "`metrics`:`http.duration` | align to 1m using avg",
    opts=opts
)
```

--------------------------------

### List Annotations (Async)

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/annotations.md

Asynchronously list annotations, with optional filtering by dataset, start time, and end time. Requires an active AsyncClient session.

```python
async with AsyncClient() as client:
    annotations = await client.annotations.list(
        datasets=["logs"],
        start=datetime.now(timezone.utc) - timedelta(days=7),
        end=datetime.now(timezone.utc)
    )
```

--------------------------------

### Legacy Structured Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Shows how to construct and execute a legacy structured query with filters, aggregations, grouping, and ordering using the Axiom Python client. This method is useful for complex queries that benefit from structured definition.

```python
from axiom_py.query import QueryLegacy, Filter, FilterOperation, Aggregation, Order
from axiom_py import QueryOptions, QueryKind
from datetime import datetime, timedelta

client = Client(token="xaat-your-token")

end = datetime.utcnow()
start = end - timedelta(hours=1)

query = QueryLegacy(
    startTime=start,
    endTime=end,
    filter=Filter(
        op=FilterOperation.EQUAL,
        field="level",
        value="error"
    ),
    aggregations=[
        Aggregation(op="count", alias="error_count")
    ],
    groupBy=["service"],
    order=[Order(field="error_count", desc=True)],
    limit=100
)

opts = QueryOptions(saveAsKind=QueryKind.ANALYTICS)
result = client.query_legacy("dataset-id", query, opts)

print(f"Rows examined: {result.status.rowsExamined}")
for entry in result.matches:
    print(f"{entry._time}: {entry.data}")
```

--------------------------------

### Configure Client with Edge Endpoint

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/errors.md

Provides examples of configuring the Axiom client with an edge endpoint, either by specifying the edge host or the full edge URL. This is required for MPL queries.

```python
client = axiom_py.Client(
    token="xaat-your-token",
    edge="eu-central-1.aws.edge.axiom.co"
)
```

```python
client = axiom_py.Client(
    token="xaat-your-token",
    edge_url="https://eu-central-1.aws.edge.axiom.co"
)
```

--------------------------------

### Querying

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Provides examples for querying data using Axiom's APL (Axiom Processing Language) and MPL (Metrics Processing Language), including options for time ranges and limits.

```APIDOC
## Querying

### APL Query
```python
from axiom_py import AplOptions
from datetime import datetime, timedelta, timezone

result = client.query(
    "['dataset'] | where level == 'error' | limit 100",
    opts=AplOptions(
        start_time=datetime.now(timezone.utc) - timedelta(hours=1),
        end_time=datetime.now(timezone.utc),
        limit=100
    )
)

for entry in result.matches:
    print(entry.data)
```

### Metrics (MPL)
```python
from axiom_py import MplOptions

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

for series in result.series:
    print(f"{series.metric}: {series.data}")
```
```

--------------------------------

### APL Query Example

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Executes a query using Axiom Processing Language (APL) to filter events by status and count them by service. Assumes a client object is already initialized.

```python
result = client.query("['dataset'] | where status == 'error' | stats count() by service")
```

--------------------------------

### Metrics Query with MPL

Source: https://github.com/axiomhq/axiom-py/blob/main/README.md

Query OTel metrics using MPL by configuring the client with an edge endpoint and using the 'mpl_query' method. The example queries 'http.server.duration' over the last hour.

```python
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 Ingest and Query

Source: https://github.com/axiomhq/axiom-py/blob/main/README.md

Utilize the asynchronous client for non-blocking I/O operations. This example shows ingesting events and querying data within an asyncio event loop.

```python
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")
        print(f"Found {len(result.matches)} matches")

asyncio.run(main())
```

--------------------------------

### Get Dataset

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Retrieves a specific dataset by its ID.

```APIDOC
## Get Dataset

### Description
Retrieves a specific dataset by its ID.

### Method
`GET`

### Endpoint
`/v1/datasets/{datasetId}`

### Parameters
#### Path Parameters
- **datasetId** (string) - Required - The ID of the dataset to retrieve.
```

--------------------------------

### Get Annotation

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/annotations.md

Retrieves a specific annotation by its ID.

```APIDOC
## GET /v1/annotations/{annotationId}

### Description
Retrieves a specific annotation by its ID.

### Method
GET

### Endpoint
/v1/annotations/{annotationId}

### Parameters
#### Path Parameters
- **annotationId** (string) - Required - The ID of the annotation to retrieve.

### Response
#### Success Response (200)
- **id** (string) - The unique identifier of the annotation.
- **time** (datetime) - The timestamp of the annotation.
- **title** (string) - The title of the annotation.
- **description** (string) - The description of the annotation.
- **url** (string) - The URL associated with the annotation.
- **type** (string) - The type of the annotation.

#### Response Example
```json
{
  "id": "annotation-id-123",
  "time": "2023-10-27T10:00:00Z",
  "title": "Production Release v1.5.0",
  "description": "Released new user authentication feature",
  "url": "https://github.com/org/repo/releases/v1.5.0",
  "type": "deployment"
}
```
```

--------------------------------

### Client Initialization

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Demonstrates how to initialize the Axiom client, both synchronously and asynchronously. The client can read credentials from environment variables or accept them directly.

```APIDOC
## Client Initialization

### Synchronous
```python
from axiom_py import Client

client = Client()  # Reads AXIOM_TOKEN, AXIOM_ORG_ID from env
client = Client(token="xaat-ப்படாத", org_id="org-id")
client = Client(edge="eu-central-1.aws.edge.axiom.co")  # With edge
```

### Asynchronous
```python
from axiom_py import AsyncClient

async with AsyncClient(token="xaat-ப்படாத") as client:
    # Use client
    pass
```
```

--------------------------------

### Get Token Details

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/tokens.md

Retrieves detailed information about a specific API token using its ID.

```APIDOC
## GET /api/v1/tokens/{tokenId}

### Description
Retrieves details for a specific API token.

### Method
GET

### Endpoint
/api/v1/tokens/{tokenId}

### Parameters
#### Path Parameters
- **tokenId** (string) - Required - The unique identifier of the token to retrieve.

### Response
#### Success Response (200)
- **id** (string) - The unique identifier of the token.
- **token** (string) - The API token string (often masked or partial in responses).
- **name** (string) - The name of the token.
- **description** (string) - The description of the token.
- **createdAt** (string) - The timestamp when the token was created.
- **expiresAt** (string) - The timestamp when the token will expire.
- **datasetCapabilities** (object) - The dataset capabilities assigned to the token.
- **organizationCapabilities** (object) - The organization capabilities assigned to the token.
```

--------------------------------

### Get a Dataset by ID

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Retrieve a specific dataset using its unique identifier. Raises an AxiomError if the dataset is not found.

```python
dataset = client.datasets.get("my-dataset-id")
print(f"Dataset: {dataset.name} ({dataset.id})")
print(f"Description: {dataset.description}")
```

--------------------------------

### Interval

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/types.md

Represents a time bucket within a time-series aggregation, including its start and end times and any grouped results.

```APIDOC
## Interval

### Description
A time bucket in a time-series.

### Fields
- **startTime** (datetime) - Interval start
- **endTime** (datetime) - Interval end
- **groups** (Optional[List[EntryGroup]]) - Grouped results
```

--------------------------------

### Initialize Asynchronous Axiom Client

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Instantiate the asynchronous Axiom client using an async context manager. Requires explicit token.

```python
from axiom_py import AsyncClient

async with AsyncClient(token="xaat-ப்படாத") as client:
    # Use client
    pass
```

--------------------------------

### Get Current Authenticated User

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/client.md

Retrieve information about the current authenticated user. This functionality is only available when using personal access tokens.

```python
# Get current authenticated user (only with personal tokens)
user = client.users.current()
if user:
    print(f"Logged in as {user.name}")
```

--------------------------------

### Synchronous Client Initialization and Usage

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Initializes a synchronous Axiom client and demonstrates ingesting events and performing a query. Requires a valid token.

```python
import axiom_py

client = axiom_py.Client(token="xaat-your-token")
status = client.ingest_events("dataset", events)
result = client.query("['dataset'] | limit 100")
```

--------------------------------

### Handle Dataset Operations Errors

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/errors.md

Catch specific AxiomError statuses (404 for get, 409 for create) when performing dataset operations.

```python
try:
    dataset = client.datasets.get("nonexistent-id")
except axiom_py.AxiomError as e:
    if e.status == 404:
        print("Dataset not found")
    else:
        raise

try:
    ds = client.datasets.create(name="my-dataset")
except axiom_py.AxiomError as e:
    if e.status == 409:
        print("Dataset already exists")
    else:
        raise
```

--------------------------------

### Initialize AsyncClient

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/async-client.md

Instantiate the asynchronous Axiom client. It's recommended to use a context manager for proper resource handling. Ensure you provide a valid API token.

```python
import asyncio
from axiom_py import AsyncClient

async def main():
    async with AsyncClient(token="xaat-your-api-token") as client:
        # Use client here
        pass

asyncio.run(main())
```

--------------------------------

### Get Annotation by ID (Async)

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/annotations.md

Asynchronously retrieve a specific annotation using its unique identifier. Requires an active AsyncClient session.

```python
async with AsyncClient() as client:
    annotation = await client.annotations.get("ann-id")
```

--------------------------------

### Asynchronously Get Dataset by ID

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Use this method to retrieve a specific dataset using its unique identifier. Ensure you are within an async context.

```python
async with AsyncClient() as client:
    dataset = await client.datasets.get("dataset-id")
```

--------------------------------

### Asynchronous Client Initialization

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Initialize the asynchronous Axiom client. Parameters are the same as the synchronous client.

```python
client = AsyncClient(
    token=None,
    org_id=None,
    url=None,
    edge_url=None,
    edge=None,
)
```

--------------------------------

### MplOptions Dataclass

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Configuration options for MPL (Metrics Processing Language) queries. Requires start and end times, and can include additional parameters.

```python
@dataclass
class MplOptions:
    start_time: datetime                    # Query start time (required)
    end_time: datetime                      # Query end time (required)
    params: Optional[Dict[str, str]] = None # Additional query parameters
    query_options: Optional[Dict[str, str]] = None  # API-specific options
    nocache: bool = False                   # Bypass query cache
```

--------------------------------

### Accessing Async Datasets Client

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/async-client.md

Demonstrates how to initialize the async client and access the datasets sub-client to retrieve a list of datasets. Ensure the client is used within an async context manager.

```python
async with AsyncClient() as client:
    datasets = await client.datasets.get_list()
    for ds in datasets:
        print(ds.name)
```

--------------------------------

### Initialize Synchronous Axiom Client

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Instantiate the synchronous Axiom client. Reads AXIOM_TOKEN and AXIOM_ORG_ID from environment variables by default. Can also be initialized with explicit token, org ID, or edge.

```python
from axiom_py import Client

client = Client()  # Reads AXIOM_TOKEN, AXIOM_ORG_ID from env
client = Client(token="xaat-ப்படாத", org_id="org-id")
client = Client(edge="eu-central-1.aws.edge.axiom.co")  # With edge
```

--------------------------------

### Asynchronous Client Initialization and Usage

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/README.md

Initializes an asynchronous Axiom client within an async context manager and demonstrates ingesting events and performing a query. Requires a valid token.

```python
from axiom_py import AsyncClient

async with AsyncClient(token="xaat-your-token") as client:
    await client.ingest_events("dataset", events)
    result = await client.query("['dataset'] | limit 100")
```

--------------------------------

### Creating Simple and Compound Filters

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/query.md

Demonstrates how to create a basic equality filter and a compound OR filter using Filter and FilterOperation classes. Ensure necessary imports are included.

```python
from axiom_py.query import Filter, FilterOperation

# Create a simple filter: field == "value"
filter1 = Filter(
    op=FilterOperation.EQUAL,
    field="status",
    value="error"
)

# Create a compound filter: (status == "error" OR level == "critical")
compound = Filter(
    op=FilterOperation.OR,
    children=[
        Filter(op=FilterOperation.EQUAL, field="status", value="error"),
        Filter(op=FilterOperation.EQUAL, field="level", value="critical"),
    ]
)
```

--------------------------------

### Synchronous Client Initialization

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/configuration.md

Initialize the synchronous Axiom client. Parameters can be set directly or via environment variables.

```python
client = Client(
    token=None,           # API token
    org_id=None,          # Organization ID
    url=None,             # Base API URL
    edge_url=None,        # Full edge URL
    edge=None,            # Edge domain
)
```

--------------------------------

### Get Current User (Asynchronous)

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/users.md

Asynchronously retrieves the currently authenticated user's details. This method is intended for use with personal access tokens (xapt-).

```python
async with AsyncClient(token="xapt-personal-token") as client:
    user = await client.users.current()
    if user:
        print(f"User: {user.name}")
```

--------------------------------

### Basic Async Client Usage

Source: https://github.com/axiomhq/axiom-py/blob/main/features/ASYNC_IMPLEMENTATION.md

Demonstrates basic asynchronous client operations including ingesting events and querying data. Ensure the AsyncClient is used within an async context manager.

```python
import asyncio
from axiom_py import AsyncClient

async def main():
    async with AsyncClient() as client:
        # Ingest events
        await client.ingest_events(
            "dataset",
            [{"field": "value"}]
        )

        # Query data
        result = await client.query("['dataset'] | limit 100")
        print(f"Found {len(result.matches)} matches")

asyncio.run(main())
```

--------------------------------

### Interval Data Structure

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/types.md

Represents a time bucket within a time-series result. It defines the start and end times of the interval and may contain grouped results.

```python
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional

from .results import EntryGroup

@dataclass
class Interval:
    startTime: datetime                 # Interval start
    endTime: datetime                   # Interval end
    groups: Optional[List[EntryGroup]]  # Grouped results
```

--------------------------------

### Create a New Dataset

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Create a new dataset with a name, optional description, and an optional edge deployment location. Returns the created dataset object.

```python
dataset = client.datasets.create(
    name="application-logs",
    description="Logs from the main application service",
    edgeDeployment="eu-west-1"
)
print(f"Created dataset: {dataset.id}")
```

--------------------------------

### FilterOperation Enum

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Enum representing various filter operations available for queries. Includes equality, inequality, greater than, starts with, and contains operations, among others.

```python
from axiom_py.query import FilterOperation

FilterOperation.EQUAL
FilterOperation.NOT_EQUAL
FilterOperation.GREATER_THAN
FilterOperation.STARTS_WITH
FilterOperation.CONTAINS
# ... and more
```

--------------------------------

### Initialize Axiom Python Client

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/client.md

Instantiate the synchronous Axiom client. You can use environment variables for token and organization ID, or provide them explicitly. For optimal data locality, specify an edge endpoint.

```python
import axiom_py

# Basic client with token from env var
client = axiom_py.Client()

# Explicit credentials
client = axiom_py.Client(
    token="xaat-your-api-token",
    org_id="your-org-id"
)

# With edge endpoint for better data locality
client = axiom_py.Client(
    token="xaat-your-api-token",
    edge="eu-central-1.aws.edge.axiom.co"
)
```

--------------------------------

### Initialize AsyncAxiomProcessor

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Instantiate the asynchronous Axiom processor for use in async structlog configurations. Requires an AsyncClient instance, a dataset name, and optionally a flush interval.

```python
class AsyncAxiomProcessor:
    def __init__(
        self,
        client: AsyncClient,
        dataset: str,
        interval: int = 1,
    ) -> None
```

--------------------------------

### Import Axiom Logging Components

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/logging.md

Import necessary classes for standard logging and structlog integration with Axiom.

```python
from axiom_py import AxiomHandler           # Standard logging
from axiom_py import AsyncAxiomHandler      # Async logging
from axiom_py import AxiomProcessor         # Structlog sync
from axiom_py import AsyncAxiomProcessor    # Structlog async
```

--------------------------------

### Client Constructor

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/client.md

Initializes the synchronous Axiom client. You can configure authentication, organization ID, and API endpoints.

```APIDOC
## Client Constructor

### Description
Initializes the synchronous Axiom client.

### Method
```python
def __init__(
    self,
    token: Optional[str] = None,
    org_id: Optional[str] = None,
    url: Optional[str] = None,
    edge_url: Optional[str] = None,
    edge: Optional[str] = None,
) -> Client
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters Table
| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| `token` | `str` | No | `AXIOM_TOKEN` env var | API authentication token. Edge endpoints require API tokens (xaat-), not personal tokens (xapt-). |
| `org_id` | `str` | No | `AXIOM_ORG_ID` env var | Organization ID (required for organization tokens). |
| `url` | `str` | No | `https://api.axiom.co` | Base URL for Axiom API (or `AXIOM_URL` env var). |
| `edge_url` | `str` | No | `None` | Full edge URL for ingest/query operations (e.g., `https://eu-central-1.aws.edge.axiom.co`). Takes precedence over `edge` parameter. Must be set explicitly. |
| `edge` | `str` | No | `None` | Edge domain for ingest/query operations (e.g., `eu-central-1.aws.edge.axiom.co`). Constructs URL as `https://{edge}/...`. Must be set explicitly. |

### Example
```python
import axiom_py

# Basic client with token from env var
client = axiom_py.Client()

# Explicit credentials
client = axiom_py.Client(
    token="xaat-your-api-token",
    org_id="your-org-id"
)

# With edge endpoint for better data locality
client = axiom_py.Client(
    token="xaat-your-api-token",
    edge="eu-central-1.aws.edge.axiom.co"
)
```
```

--------------------------------

### Configure Axiom Environment Variables

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/QUICK_REFERENCE.md

Set environment variables for token, organization ID, and API URL for authentication and connection.

```bash
export AXIOM_TOKEN="xaat-your-token"
export AXIOM_ORG_ID="org-id"
export AXIOM_URL="https://api.axiom.co"
```

--------------------------------

### Asynchronously Create Dataset

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/datasets.md

Create a new dataset asynchronously. You can provide a name, an optional description, and specify an edge deployment location.

```python
async with AsyncClient() as client:
    ds = await client.datasets.create(
        name="async-logs",
        description="Async dataset"
    )
```

--------------------------------

### Create Annotation

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/annotations.md

Creates a new deployment annotation with specified datasets, time, title, description, URL, and type.

```APIDOC
## POST /v1/annotations

### Description
Creates a new annotation in Axiom.

### Method
POST

### Endpoint
/v1/annotations

### Parameters
#### Request Body
- **datasets** (list[string]) - Required - List of datasets the annotation applies to.
- **time** (datetime) - Required - The timestamp for the annotation.
- **title** (string) - Required - The title of the annotation.
- **description** (string) - Optional - A detailed description of the annotation.
- **url** (string) - Optional - A URL associated with the annotation.
- **type** (string) - Optional - The type of annotation (e.g., 'deployment', 'issue').

### Request Example
```json
{
  "datasets": ["production-logs", "production-metrics"],
  "time": "2023-10-27T10:00:00Z",
  "title": "Production Release v1.5.0",
  "description": "Released new user authentication feature",
  "url": "https://github.com/org/repo/releases/v1.5.0",
  "type": "deployment"
}
```

### Response
#### Success Response (200)
- **id** (string) - The unique identifier of the created annotation.

#### Response Example
```json
{
  "id": "annotation-id-123"
}
```
```

--------------------------------

### Build Package for Distribution

Source: https://github.com/axiomhq/axiom-py/blob/main/CLAUDE.md

Create a distributable package of the axiom-py library. This command is used for preparing the library for release.

```bash
# Build the package for distribution
uv build
```

--------------------------------

### Ingest Events and Query Data with Axiom Python

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/00_START_HERE.md

This snippet demonstrates how to create an Axiom client, ingest events into a dataset, and query that dataset. Ensure the AXIOM_TOKEN environment variable is set or pass your token directly to the Client constructor. The client automatically handles retries for 5xx errors.

```python
import axiom_py

# Create client
client = axiom_py.Client()

# Send events
status = client.ingest_events(
    dataset="my-dataset",
    events=[
        {"message": "Event 1", "level": "info"},
        {"message": "Event 2", "level": "error"},
    ]
)

# Query data
result = client.query("['my-dataset'] | where level == 'error' | limit 100")

print(f"Ingested: {status.ingested}, Found: {len(result.matches)}")
```

--------------------------------

### Get API Token Information

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/tokens.md

Retrieves information about a specific API token using its unique ID. The returned token object contains details like name and capabilities but not the token string itself.

```python
token = client.tokens.get("token-id-123")
print(f"Token: {token.name}")
print(f"Capabilities: {token.orgCapabilities}")
```

--------------------------------

### Get Current User (Synchronous)

Source: https://github.com/axiomhq/axiom-py/blob/main/_autodocs/api-reference/users.md

Retrieves the currently authenticated user's details. This method only works with personal access tokens (xapt-). Returns None if an organization token (xaat-) is used.

```python
user = client.users.current()
if user:
    print(f"Logged in as: {user.name} ({user.email})")
    print(f"Role: {user.role.name}")
else:
    print("Using organization token (not associated with a user)")
```