Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions examples/async_usage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
"""Async client examples for the Metronome Python SDK.

Demonstrates how to use the async client with both httpx and aiohttp backends.

Usage:
export METRONOME_BEARER_TOKEN="your-api-token"
python examples/async_usage.py
"""

import asyncio

from metronome import AsyncMetronome


async def basic_async():
"""Basic async client usage with httpx (default backend)."""
client = AsyncMetronome()

# All methods are the same as the sync client, but awaited
customers = []
async for customer in client.v1.customers.list():
customers.append(customer)
print(f"Fetched {len(customers)} customers")

await client.close()


async def async_with_context_manager():
"""Use the async client as a context manager for automatic cleanup."""
async with AsyncMetronome() as client:
client.v1.usage.ingest(
usage=[
{
"transaction_id": "90e9401f-0f8c-4cd3-9a9f-d6beb56d8d72",
"customer_id": "team@example.com",
"event_type": "heartbeat",
"timestamp": "2025-01-01T00:00:00Z",
"properties": {"cpu_seconds": 60},
}
],
)
print("Usage event ingested")


async def async_with_aiohttp():
"""Use the aiohttp backend for improved concurrency performance.

Requires: pip install metronome-sdk[aiohttp]
"""
from metronome import DefaultAioHttpClient

async with AsyncMetronome(http_client=DefaultAioHttpClient()) as client:
customers = []
async for customer in client.v1.customers.list():
customers.append(customer)
print(f"Fetched {len(customers)} customers")


if __name__ == "__main__":
# Uncomment the example you want to run:
# asyncio.run(basic_async())
# asyncio.run(async_with_context_manager())
# asyncio.run(async_with_aiohttp())
pass
74 changes: 74 additions & 0 deletions examples/error_handling.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
"""Error handling examples for the Metronome Python SDK.

Demonstrates how to handle API errors, configure retries, and set timeouts.

Usage:
export METRONOME_BEARER_TOKEN="your-api-token"
python examples/error_handling.py
"""

from datetime import datetime

import metronome
from metronome import Metronome


def basic_error_handling():
"""Handle common API errors with specific exception types."""
client = Metronome()

try:
client.v1.contracts.create(
customer_id="13117714-3f05-48e5-a6e9-a66093f13b4d",
starting_at=datetime.fromisoformat("2025-01-01T00:00:00"),
)
except metronome.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx.
except metronome.RateLimitError as e:
print("A 429 status code was received; we should back off a bit.")
except metronome.BadRequestError as e:
print(f"Bad request (400): {e.message}")
except metronome.AuthenticationError as e:
print(f"Authentication failed (401): {e.message}")
except metronome.NotFoundError as e:
print(f"Resource not found (404): {e.message}")
except metronome.APIStatusError as e:
print(f"API error (status {e.status_code}): {e.message}")
print(f"Response body: {e.response}")


def configure_retries():
"""Configure automatic retry behavior."""
# Disable retries entirely
client = Metronome(max_retries=0)

# Or increase retries (default is 2)
client = Metronome(max_retries=5)

# Override retries for a specific request
client.with_options(max_retries=5).v1.customers.list()


def configure_timeouts():
"""Configure request timeouts."""
import httpx

# Set a global timeout (in seconds, default is 60)
client = Metronome(timeout=20.0)

# Fine-grained timeout control
client = Metronome(
timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override timeout for a specific request
client.with_options(timeout=5.0).v1.customers.list()


if __name__ == "__main__":
# Uncomment the example you want to run:
# basic_error_handling()
# configure_retries()
# configure_timeouts()
pass
55 changes: 55 additions & 0 deletions examples/pagination.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
"""Pagination examples for the Metronome Python SDK.

Demonstrates auto-pagination and manual page control for list endpoints.

Usage:
export METRONOME_BEARER_TOKEN="your-api-token"
python examples/pagination.py
"""

from metronome import Metronome

client = Metronome()


def auto_paginate():
"""Automatically iterate through all pages of results."""
all_customers = []
for customer in client.v1.customers.list():
all_customers.append(customer)
print(f"Fetched {len(all_customers)} customers")


def auto_paginate_with_limit():
"""Iterate through results with a maximum limit per page."""
for plan in client.v1.plans.list(limit=10):
print(f"Plan: {plan.id}")


def manual_page_control():
"""Manually control pagination using page objects."""
first_page = client.v1.contracts.products.list()

print(f"Items on first page: {len(first_page.data)}")

if first_page.has_next_page():
print(f"Next page info: {first_page.next_page_info()}")
next_page = first_page.get_next_page()
print(f"Items on second page: {len(next_page.data)}")


def collect_all_items():
"""Collect all items from a paginated endpoint into a list."""
all_products = []
for product in client.v1.contracts.products.list():
all_products.append(product)
print(f"Total products: {len(all_products)}")


if __name__ == "__main__":
# Uncomment the example you want to run:
# auto_paginate()
# auto_paginate_with_limit()
# manual_page_control()
# collect_all_items()
pass
Loading