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
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_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
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
4. Create a customer
Python
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = 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
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = customer_api.add_or_update(message)
28
29# Set up ingest API client and ingest a meter
30from time import time
31from metering.ingest import create_ingest_client
32
33ingest_api = create_ingest_client(api_key=API_KEY)
6. Send a meter record Meter records will be collected in batches and sent asynchronously.
Python
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = customer_api.add_or_update(message)
28
29# Set up ingest API client and ingest a meter
30from time import time
31from metering.ingest import create_ingest_client
32
33ingest_api = create_ingest_client(api_key=API_KEY)
34
35ingest_api.meter(
36 meter_api_name="api-calls",
37 meter_value=1,
38 meter_time_in_millis=int(time() * 1000),
39 customer_id=customer["customerId"],
40)
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
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = customer_api.add_or_update(message)
28
29# Set up ingest API client and ingest a meter
30from time import time
31from metering.ingest import create_ingest_client
32
33ingest_api = create_ingest_client(api_key=API_KEY)
34
35ingest_api.meter(
36 meter_api_name="api-calls",
37 meter_value=1,
38 meter_time_in_millis=int(time() * 1000),
39 customer_id=customer["customerId"],
40)
41
42# Shut down ingest API client
43ingest_api.shutdown()
8. Set up Usage API client
Python
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = customer_api.add_or_update(message)
28
29# Set up ingest API client and ingest a meter
30from time import time
31from metering.ingest import create_ingest_client
32
33ingest_api = create_ingest_client(api_key=API_KEY)
34
35ingest_api.meter(
36 meter_api_name="api-calls",
37 meter_value=1,
38 meter_time_in_millis=int(time() * 1000),
39 customer_id=customer["customerId"],
40)
41
42# Shut down ingest API client
43ingest_api.shutdown()
44
45# Setup usage API client and get a usage report
46from metering.usage import (AggregationType, Take, TimeGroupingInterval,
47 TimeRange, UsageApiClient, create_usage_query)
48
49usage_api = UsageApiClient(os.environ.get("API_KEY"))
9. Get a usage report
Python
1# Set up virtual environment and install SDK
2#
3# $ python3 -m virtualenv venv
4# $ . venv/bin/activate
5# $ pip install amberflo-metering-python
6
7import os
8
9# Sign up for free at https://ui.amberflo.io/ to get an API key
10API_KEY = os.environ["AMBERFLO_API_KEY"]
11
12# Set up customer API client and create a customer
13from metering.customer import CustomerApiClient, create_customer_payload
14
15customer_api = CustomerApiClient(os.environ.get("API_KEY"))
16
17message = create_customer_payload(
18 customer_id="customer-123",
19 customer_name="Customer 123",
20 # email is optional
21 customer_email="[email protected]",
22 # traits are optional. They can be used as filters or aggregation buckets
23 traits={
24 "region": "us-east-1",
25 },
26)
27customer = customer_api.add_or_update(message)
28
29# Set up ingest API client and ingest a meter
30from time import time
31from metering.ingest import create_ingest_client
32
33ingest_api = create_ingest_client(api_key=API_KEY)
34
35ingest_api.meter(
36 meter_api_name="api-calls",
37 meter_value=1,
38 meter_time_in_millis=int(time() * 1000),
39 customer_id=customer["customerId"],
40)
41
42# Shut down ingest API client
43ingest_api.shutdown()
44
45# Setup usage API client and get a usage report
46from metering.usage import (AggregationType, Take, TimeGroupingInterval,
47 TimeRange, UsageApiClient, create_usage_query)
48
49usage_api = UsageApiClient(os.environ.get("API_KEY"))
50
51since_two_days_ago = TimeRange(int(time()) - 60 * 60 * 24 * 2)
52
53query = create_usage_query(
54 meter_api_name="api-calls",
55 aggregation=AggregationType.SUM,
56 time_grouping_interval=TimeGroupingInterval.DAY,
57 time_range=since_two_days_ago,
58 group_by=["customerId"],
59 usage_filter={"customerId": ["customer-123"]},
60 take=Take(limit=10, is_ascending=False),
61)
62report = usage_api.get(query)
Updated 11 Sep 2024
Did this page help you?