antonlytics

Installation

pip install antonlytics

Quick Start

from antonlytics import Antonlytics, Triplet, EntityRef

anto = Antonlytics(api_key="anto_live_xxx")

# Ingest a relationship
anto.ingest.track(
    "proj_abc",
    triplets=Triplet(
        subject=EntityRef("Customer", id="c1", properties={"name": "Alice"}),
        predicate="PURCHASED",
        object=EntityRef("Product", id="p1", properties={"title": "Laptop", "price": 999}),
    ),
)

# Query the graph
result = (
    anto.query.build("proj_abc")
    .select("Customer", alias="c1")
        .properties("name", "email", "country")
        .eq("country", "USA")
        .gte("age", 18)
    .done()
    .order_by("age", direction="desc")
    .limit(50)
    .run()
)
for row in result:
    print(row["name"], row["country"])

Sync Client

from antonlytics import Antonlytics

# Context manager — closes connection pool on exit
with Antonlytics(
    api_key="anto_live_xxx",
    base_url="https://api.antonlytics.com",  # optional
    timeout=30.0,
    max_retries=2,
    debug=False,
) as anto:
    projects = anto.projects.list()
    metrics  = anto.dashboard.metrics("proj_abc")

Async Client

import asyncio
from antonlytics import AsyncAntonlytics

async def main():
    async with AsyncAntonlytics(api_key="anto_live_xxx") as anto:
        # Parallel calls
        ontology, metrics = await asyncio.gather(
            anto.query.ontology("proj_abc"),
            anto.dashboard.metrics("proj_abc"),
        )
        # Fluent query
        result = await (
            anto.query.build("proj_abc")
            .select("Customer").eq("country", "USA").done()
            .limit(20)
            .run()
        )
        for row in result:
            print(row)

asyncio.run(main())

For FastAPI, use AsyncAntonlytics as a module-level singleton and call await anto.aclose() in the shutdown event.

Query Builder

result = (
    anto.query.build("proj_abc")
    .select("Customer", alias="c1")
        .properties("name", "email", "country", "age")
        .eq("country", "USA")
        .gte("age", 21)
        .relates_to("PURCHASED", "p1")   # join via relationship
    .done()
    .select("Product", alias="p1")
        .properties("title", "price")
        .lte("price", 500)
    .done()
    .order_by("age", direction="desc")
    .limit(100)
    .name("US adults buying affordable products")
    .run()
)

print(f"{result.total} rows in {result.execution_ms}ms")
for row in result:             # QueryResult is directly iterable
    print(row)

Error Handling

from antonlytics import (
    AntoError, AuthenticationError, PlanLimitError,
    NotFoundError, RateLimitError, NetworkError,
)

try:
    anto.ingest.track(...)
except PlanLimitError as e:
    print(f"Limit hit: {e.message}")
    # Redirect to /billing
except AuthenticationError:
    print("API key invalid or revoked")
except AntoError as e:
    print(f"[{e.code}] HTTP {e.status}: {e.message}")
    print(e.details)

CLI Reference

# Install
pip install antonlytics

# Set API key
export ANTO_API_KEY=anto_live_xxx

# Commands
anto projects
anto stats      <project-id>
anto ontology   <project-id>
anto ingest     <project-id> ./triplets.json
anto query      <project-id> ./query.json
anto dashboard  <project-id>
anto poll       <event-id>

# Options
ANTO_BASE_URL=http://localhost:8000   # self-hosted
ANTO_DEBUG=1                          # log HTTP calls