Recipes
Python SDK

5min
Install the SDK
Set up virtual environment and install SDK.
Get your API key
You can sign up for free at https://ui.amberflo.io/ to get an API key and start testing.
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]
﻿
3. Set up the Customer API client
- You can use it to manage your customers. Meter records are associated with customers.
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))
﻿
4. Create a customer
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)
﻿
5. Set up Ingest API client
By default, it will collect the meter records and send them in batches asynchronously.
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)
﻿
6. Send a meter record
Meter records will be collected in batches and sent asynchronously.
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)
﻿
7. Shut down Ingest API client
This is a blocking call that ensures all batches have been sent. Use it in your server shutdown procedure.
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()
﻿
8. Set up Usage API client
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

usage_api = UsageApiClient(os.environ.get("API_KEY"))
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

usage_api = UsageApiClient(os.environ.get("API_KEY"))
﻿
9. Get a usage report
Python
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="[email protected]",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

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

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

query = create_usage_query(
    meter_api_name="api-calls",
    aggregation=AggregationType.SUM,
    time_grouping_interval=TimeGroupingInterval.DAY,
    time_range=since_two_days_ago,
    group_by=["customerId"],
    usage_filter={"customerId": ["customer-123"]},
    take=Take(limit=10, is_ascending=False),
)
report = usage_api.get(query)
# Set up virtual environment and install SDK
#
#   $ python3 -m virtualenv venv
#   $ . venv/bin/activate
#   $ pip install amberflo-metering-python

import os

# Sign up for free at https://ui.amberflo.io/ to get an API key
API_KEY = os.environ["AMBERFLO_API_KEY"]

# Set up customer API client and create a customer
from metering.customer import CustomerApiClient, create_customer_payload

customer_api = CustomerApiClient(os.environ.get("API_KEY"))

message = create_customer_payload(
    customer_id="customer-123",
    customer_name="Customer 123",
    # email is optional
    customer_email="customer-123@example.com",
    # traits are optional. They can be used as filters or aggregation buckets
    traits={
        "region": "us-east-1",
    },
)
customer = customer_api.add_or_update(message)

# Set up ingest API client and ingest a meter
from time import time
from metering.ingest import create_ingest_client

ingest_api = create_ingest_client(api_key=API_KEY)

ingest_api.meter(
    meter_api_name="api-calls",
    meter_value=1,
    meter_time_in_millis=int(time() * 1000),
    customer_id=customer["customerId"],
)

# Shut down ingest API client
ingest_api.shutdown()

# Setup usage API client and get a usage report
from metering.usage import (AggregationType, Take, TimeGroupingInterval,
                            TimeRange, UsageApiClient, create_usage_query)

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

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

query = create_usage_query(
    meter_api_name="api-calls",
    aggregation=AggregationType.SUM,
    time_grouping_interval=TimeGroupingInterval.DAY,
    time_range=since_two_days_ago,
    group_by=["customerId"],
    usage_filter={"customerId": ["customer-123"]},
    take=Take(limit=10, is_ascending=False),
)
report = usage_api.get(query)
﻿
﻿
Updated 11 Sep 2024
Did this page help you?
Recipes
Java SDK- simple