File size: 4,027 Bytes
048b1e8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | from os import environ
from typing import Optional
import asyncio
from openmeter.aio import Client
from corehttp.exceptions import HttpResponseError
ENDPOINT: str = environ.get("OPENMETER_ENDPOINT") or "https://openmeter.cloud"
token: Optional[str] = environ.get("OPENMETER_TOKEN")
customer_key: str = environ.get("OPENMETER_CUSTOMER_KEY") or "acme-corp-1"
feature_key: str = environ.get("OPENMETER_FEATURE_KEY") or "api_access"
async def main() -> None:
async with Client(
endpoint=ENDPOINT,
token=token,
) as client:
try:
# Check customer access to a specific feature
print(f"Checking access for customer '{customer_key}' to feature '{feature_key}'...")
entitlement_value = await client.customer_entitlement.get_customer_entitlement_value(
customer_key, feature_key
)
print(f"\nEntitlement Value:")
print(f"Has Access: {entitlement_value.has_access}")
# For metered entitlements, additional properties are available
if entitlement_value.balance is not None:
print(f"Balance: {entitlement_value.balance}")
if entitlement_value.usage is not None:
print(f"Usage: {entitlement_value.usage}")
if entitlement_value.overage is not None:
print(f"Overage: {entitlement_value.overage}")
# For static entitlements, config is available
if entitlement_value.config is not None:
print(f"Config: {entitlement_value.config}")
# List customer entitlements and demonstrate type-specific handling
print(f"\nListing all entitlements for customer '{customer_key}'...")
entitlements_response = await client.customer_entitlements_v2.list(customer_key)
print(f"\nEntitlements by Type:")
for entitlement in entitlements_response.items_property:
# Note: Due to a deserialization issue in the SDK, items come back as dicts
# Access fields using dict syntax or .get()
print(f"\n Feature: {entitlement.get('featureKey')}")
print(f" ID: {entitlement.get('id')}")
# Handle different entitlement types using discriminator
entitlement_type = entitlement.get("type")
if entitlement_type == "metered":
# Metered entitlement
print(f" Type: Metered")
print(f" Soft Limit: {entitlement.get('isSoftLimit')}")
if entitlement.get("issueAfterReset") is not None:
print(f" Issue After Reset: {entitlement.get('issueAfterReset')}")
elif entitlement_type == "static":
# Static entitlement
print(f" Type: Static")
if entitlement.get("config") is not None:
print(f" Config: {entitlement.get('config')}")
elif entitlement_type == "boolean":
# Boolean entitlement
print(f" Type: Boolean")
# Get overall customer access to all features
print(f"\nGetting overall access for customer '{customer_key}'...")
customer_access = await client.customer.get_customer_access(customer_key)
print(f"\nCustomer Access Summary:")
print(f"Total entitlements: {len(customer_access.entitlements)}")
for feature, value in customer_access.entitlements.items():
access_status = "✓" if value.has_access else "✗"
print(f" {access_status} {feature}: has_access={value.has_access}")
if value.balance is not None:
print(f" Balance: {value.balance}")
if value.usage is not None:
print(f" Usage: {value.usage}")
except HttpResponseError as e:
print(f"Error: {e}")
asyncio.run(main())
|