axiom-py

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())