From 9992e5616bb3f365cb033e72595d3f05ab52cd6f Mon Sep 17 00:00:00 2001 From: Kenneth DuMez Date: Tue, 17 Feb 2026 11:59:33 -0500 Subject: [PATCH] feat: add cross-cutting code examples --- examples/async_usage.py | 64 +++++++++++++++++++++++++++++++++ examples/error_handling.py | 74 ++++++++++++++++++++++++++++++++++++++ examples/pagination.py | 55 ++++++++++++++++++++++++++++ 3 files changed, 193 insertions(+) create mode 100644 examples/async_usage.py create mode 100644 examples/error_handling.py create mode 100644 examples/pagination.py diff --git a/examples/async_usage.py b/examples/async_usage.py new file mode 100644 index 00000000..eed521ee --- /dev/null +++ b/examples/async_usage.py @@ -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 diff --git a/examples/error_handling.py b/examples/error_handling.py new file mode 100644 index 00000000..353bbdf3 --- /dev/null +++ b/examples/error_handling.py @@ -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 diff --git a/examples/pagination.py b/examples/pagination.py new file mode 100644 index 00000000..9e8b8d60 --- /dev/null +++ b/examples/pagination.py @@ -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