Python

15 min

https://github.com/amberflo/metering-python

Amberflo Python SDK

Features

Add and update Customers
Assign or update product plans for customers
Send meter events
- In asynchronous batches (high throughput)
- Synchronously
- Via Amberflo’s API or through an AWS S3 bucket
Query usage data
Fine-grained logging and error handling

Quick Start

Go to Settings and copy your API key.
Install the SDK pip install amberflo-metering-python
Create a customer

import os
from time import time
from metering.ingest import create_ingest_client

client = create_ingest_client(api_key=os.environ["API_KEY"])

dimensions = {"region": "us-east-1"}
customer_id = "sample-customer-123"

client.meter(
    meter_api_name="sample-meter",
    meter_value=5,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer_id,
    dimensions=dimensions,
)

4. Ingest meter events

import os
from time import time
from metering.ingest import create_ingest_client

client = create_ingest_client(api_key=os.environ["API_KEY"])

dimensions = {"region": "us-east-1"}
customer_id = "sample-customer-123"

client.meter(
    meter_api_name="sample-meter",
    meter_value=5,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer_id,
    dimensions=dimensions,
)

5. Query usage

import os
from time import time
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

client = UsageApiClient(os.environ.get("API_KEY"))

since_two_days_ago = TimeRange(int(time()) - 60 * 60 * 24 * 2)

query = create_usage_query(
    meter_api_name="my_meter",
    aggregation=AggregationType.SUM,
    time_grouping_interval=TimeGroupingInterval.DAY,
    time_range=since_two_days_ago,
    group_by=["customerId"],
    usage_filter={"customerId": ["some-customer-321", "sample-customer-123"]},
    take=Take(limit=10, is_ascending=False),
)
report = client.get(query)

High Throughput Ingestion

The Amberflo Python SDK is designed for high throughput environments. You can safely send hundreds of meter records per second. For example, this works well in a web server that handles hundreds of requests each second.

Each meter event is queued in memory rather than sent immediately. Events are batched and flushed in the background to improve performance. Both the batch size and the flush rate can be customized.

Flush on Demand

At the end of your program, you may want to flush the queue to ensure that all events are sent. This is a blocking call that waits until the queue is empty. It is best used in cleanup scripts and should be avoided during regular request handling.

Error handling: You can define an on_error callback to handle errors that occur during batch sends.

Here is a complete example, showing the default values of all options:

def on_error_callback(error, batch):
    ...

client = create_ingest_client(
    api_key=API_KEY,
    max_queue_size=100000,  # max number of items in the queue before rejecting new items
    threads=2,  # number of worker threads doing the sending
    retries=2,  # max number of retries after failures
    batch_size=100,  # max number of meter records in a batch
    send_interval_in_secs=0.5,  # wait time before sending an incomplete batch
    sleep_interval_in_secs=0.1,  # wait time after failure to send or queue empty
    on_error=on_error_callback,  # handle failures to send a batch
)

...

client.meter(...)

client.flush()  # block and make sure all messages are sent

Handling Message Overload

If the SDK detects that it cannot flush messages as quickly as they are being added, it will stop accepting new messages. This prevents your program from crashing due to a backed-up metering queue and allows it to continue running smoothly.

Ingesting Through the S3 Bucket

The SDK includes a metering.ingest.IngestS3Client that allows you to send meter records via an AWS S3 bucket.

To enable this feature, install the SDK with the s3 option:

pip install amberflo-metering-python[s3]

Then, pass your S3 bucket credentials to the factory function to send records:

client = create_ingest_client(
    bucket_name=os.environ.get("BUCKET_NAME"),
    access_key=os.environ.get("ACCESS_KEY"),
    secret_key=os.environ.get("SECRET_KEY"),
)

Documentation

General documentation on how to use Amberflo is available at Product Walktrough.

The full REST API documentation is available at API Reference.

Samples

Code samples covering different scenarios are available here.

Reference

API Clients

Ingest meter records

from metering.ingest import (
    create_ingest_payload,
    create_ingest_client,
)

Create a customer

from metering.customer import (
    CustomerApiClient,
    create_customer_payload,
)

Usage

from metering.usage import (
    AggregationType,
    Take,
    TimeGroupingInterval,
    TimeRange,
    UsageApiClient,
    create_usage_query,
    create_all_usage_query,
)

Customer Portal Session

from metering.customer_portal_session import (
    CustomerPortalSessionApiClient,
    create_customer_portal_session_payload,
)

Customer Prepaid Order

from metering.customer_prepaid_order import (
    BillingPeriod,
    BillingPeriodUnit,
    CustomerPrepaidOrderApiClient,
    create_customer_prepaid_order_payload,
)

Customer Product Invoice

from metering.customer_product_invoice import (
    CustomerProductInvoiceApiClient,
    create_all_invoices_query,
    create_latest_invoice_query,
    create_invoice_query,
)

Customer Product Plan

from metering.customer_product_plan import (
    CustomerProductPlanApiClient,
    create_customer_product_plan_payload,
)

Exceptions

from metering.exceptions import ApiError

Logging

amberflo-metering-python uses the standard Python logging framework. By default, logging is and set at the WARNING level.

The following loggers are used:

metering.ingest.producer
metering.ingest.s3_client
metering.ingest.consumer
metering.session.ingest_session
metering.session.api_session