Skip to main content

Python SDK for FlexPrice API

Project description

FlexPrice Python SDK

This is the Python client library for the FlexPrice API.

Installation

pip install flexprice

Usage

"""
FlexPrice Python SDK Example

This example demonstrates how to use the FlexPrice Python SDK
to interact with the FlexPrice API.
"""

import os
import time
import datetime
from pprint import pprint

# Import the FlexPrice SDK
import flexprice
from flexprice.api import customers_api, events_api
from flexprice.models.dto_create_customer_request import DtoCreateCustomerRequest
from flexprice.models.dto_ingest_event_request import DtoIngestEventRequest

# Optional: Load environment variables from .env file
from dotenv import load_dotenv
load_dotenv()


def run_example():
    """Main example function demonstrating FlexPrice SDK usage."""
    print("Starting FlexPrice Python SDK example...")

    try:
        # Configure the API client
        api_key = os.getenv("FLEXPRICE_API_KEY")
        api_host = os.getenv("FLEXPRICE_API_HOST", "api.cloud.flexprice.io")

        if not api_key:
            raise ValueError("FLEXPRICE_API_KEY environment variable is required")
            
        print("Using API Key:", api_key[:4] + "..." + api_key[-4:])  # Show just the start and end for security

        # Configure API key authorization
        configuration = flexprice.Configuration(
            host=f"https://{api_host}/v1"
        )
        configuration.api_key['x-api-key'] = api_key
       
        # Create API client
        with flexprice.ApiClient(configuration) as api_client:
            # Set the API key header
            api_client.default_headers['x-api-key'] = api_key
            # Add User-Agent header
            configuration.user_agent = "FlexPricePythonSDK/1.0.0 Example"
            # Print actual headers for debugging
            
            # Create API instances
            events_api_instance = events_api.EventsApi(api_client)

            # Generate a unique customer ID for this example
            customer_id = f"sample-customer-{int(time.time())}"
            
            print(f"Creating customer with ID: {customer_id}...")

            # Step 1: Create an event
            print("Creating event...")
            
            event_request = DtoIngestEventRequest(
                event_name="Sample Event",
                external_customer_id=customer_id,
                properties={
                    "source": "python_sample_app",
                    "environment": "test",
                    "timestamp": datetime.datetime.now().isoformat()
                },
                source="python_sample_app"
            )
            
            event_result = events_api_instance.events_post(event=event_request)
            print(f"Event created successfully! ID: {event_result.event_id if hasattr(event_result, 'event_id') else 'unknown'}")

            # Step 2: Retrieve events for this customer
            print(f"Retrieving events for customer {customer_id}...")
            
            events_response = events_api_instance.events_get(external_customer_id=customer_id)
            
            # Check if events are available in the response
            if hasattr(events_response, 'events') and events_response.events:
                print(f"Found {len(events_response.events)} events:")
                
                for i, event in enumerate(events_response.events):
                    print(f"Event {i+1}: {event.id if hasattr(event, 'id') else 'unknown'} - {event.event_name if hasattr(event, 'event_name') else 'unknown'}")
                    print(f"Properties: {event.properties if hasattr(event, 'properties') else {}}")
            else:
                print("No events found or events not available in response.")
            
            print("Example completed successfully!")

    except flexprice.ApiException as e:
        print(f"\n=== API Exception ===")
        print(f"Status code: {e.status}")
        print(f"Reason: {e.reason}")
        print(f"HTTP response headers: {e.headers}")
        print(f"HTTP response body: {e.body}")    
    except ValueError as e:
        print(f"Value error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

Asynchronous Event Submission

The FlexPrice SDK provides asynchronous event submission functionality that allows you to:

  • Submit events in a non-blocking manner with "fire-and-forget" capability
  • Include optional callbacks to handle success/failure responses
  • Automatically retry failed event submissions with exponential backoff
  • Process events in background threads

Basic Async Usage

from flexprice import Configuration, ApiClient, EventsApi
from flexprice.models import DtoIngestEventRequest

# Configure the client
configuration = Configuration(api_key={'ApiKeyAuth': 'YOUR_API_KEY'})
configuration.host = "https://api.cloud.flexprice.io/v1"

# Create API client and event API instance
api_client = ApiClient(configuration)
events_api = EventsApi(api_client)

# Create an event
event = DtoIngestEventRequest(
    external_customer_id="customer123",
    event_name="api_call",
    properties={"region": "us-west", "method": "GET"},
    source="my_application"
)

# Submit asynchronously (fire-and-forget)
events_api.events_post_async(event)

Using Callbacks

# Define a callback function
def on_event_processed(result, error, success):
    if success:
        print(f"Event processed successfully: {result}")
    else:
        print(f"Event processing failed: {error}")

# Create and submit event with callback
event = DtoIngestEventRequest(
    external_customer_id="customer123",
    event_name="user_action",
    properties={"action": "login", "device": "mobile"},
    source="user_portal"
)

# Submit with callback
events_api.events_post_async(event, callback=on_event_processed)

Complete Example

For a complete example of asynchronous event submission, see the async_event_example.py file in the examples directory.

Running the Example

To run the provided example:

  1. Clone the repository:

    git clone https://github.com/flexprice/python-sdk.git
    cd python-sdk/examples
    
  2. Create a virtual environment and install dependencies:

    python -m venv venv
    source venv/bin/activate  # On Windows: venv\Scripts\activate
    pip install -r requirements.txt
    
  3. Create a .env file with your API credentials:

    cp .env.sample .env
    # Edit .env with your API key
    
  4. Run the example:

    python example.py
    
  5. Run the async example:

    python async_event_example.py
    

Features

  • Complete API coverage
  • Strong type hints
  • Detailed documentation
  • Error handling
  • Asynchronous support for event submission

Documentation

For detailed API documentation, refer to the code comments and the official FlexPrice API documentation.

Advanced Usage

Handling Errors

The SDK provides detailed error information through exceptions:

try:
    # API call
    result = client.some_api_call()
except flexprice.ApiException as e:
    print(f"API exception: {e}")
    print(f"Status code: {e.status}")
    print(f"Response body: {e.body}")
except Exception as e:
    print(f"General exception: {e}")

Asynchronous API Usage with asyncio

In addition to the built-in asynchronous event submission, the SDK can be used with libraries like asyncio for other operations:

import asyncio
import flexprice
from flexprice.api import customers_api

async def get_customer(customer_id):
    configuration = flexprice.Configuration(
        host="https://api.cloud.flexprice.io/v1"
    )
    configuration.api_key['x-api-key'] = "your-api-key"
    
    async with flexprice.ApiClient(configuration) as api_client:
        api = customers_api.CustomersApi(api_client)
        return await api.customers_id_get(id=customer_id)

# Run with asyncio
customer = asyncio.run(get_customer("customer-123"))
print(customer)

Summary

FlexPrice API: FlexPrice API Service

Table of Contents

SDK Installation

[!TIP] To finish publishing your SDK to PyPI you must run your first generation action.

[!NOTE] Python version upgrade policy

Once a Python version reaches its official end of life date, a 3-month grace period is provided for users to upgrade. Following this grace period, the minimum python version supported in the SDK will be updated.

The SDK can be installed with uv, pip, or poetry package managers.

uv

uv is a fast Python package installer and resolver, designed as a drop-in replacement for pip and pip-tools. It's recommended for its speed and modern Python tooling capabilities.

uv add git+<UNSET>.git

PIP

PIP is the default package installer for Python, enabling easy installation and management of packages from PyPI via the command line.

pip install git+<UNSET>.git

Poetry

Poetry is a modern tool that simplifies dependency management and package publishing by using a single pyproject.toml file to handle project metadata and dependencies.

poetry add git+<UNSET>.git

Shell and script usage with uv

You can use this SDK in a Python shell with uv and the uvx command that comes with it like so:

uvx --from flexprice-sdk-test python

It's also possible to write a standalone Python script without needing to set up a whole project like so:

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = [
#     "flexprice-sdk-test",
# ]
# ///

from flexprice_sdk_test import FlexPrice

sdk = FlexPrice(
  # SDK arguments
)

# Rest of script here...

Once that is saved to a file, you can run it with uv run script.py where script.py can be replaced with the actual file name.

IDE Support

PyCharm

Generally, the SDK will work well with most IDEs out of the box. However, when using PyCharm, you can enjoy much better integration with Pydantic by installing an additional plugin.

SDK Example Usage

Example

# Synchronous Example
from flexprice_sdk_test import FlexPrice


with FlexPrice(
    server_url="https://api.example.com",
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:

    res = flex_price.addons.get_addons()

    # Handle response
    print(res)

The same SDK client can also be used to make asynchronous requests by importing asyncio.

# Asynchronous Example
import asyncio
from flexprice_sdk_test import FlexPrice

async def main():

    async with FlexPrice(
        server_url="https://api.example.com",
        api_key_auth="<YOUR_API_KEY_HERE>",
    ) as flex_price:

        res = await flex_price.addons.get_addons_async()

        # Handle response
        print(res)

asyncio.run(main())

Authentication

Per-Client Security Schemes

This SDK supports the following security scheme globally:

Name Type Scheme
api_key_auth apiKey API key

To authenticate with the API the api_key_auth parameter must be set when initializing the SDK client instance. For example:

from flexprice_sdk_test import FlexPrice


with FlexPrice(
    server_url="https://api.example.com",
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:

    res = flex_price.addons.get_addons()

    # Handle response
    print(res)

Available Resources and Operations

Available methods

Addons

AlertLogs

Auth

Connections

Costs

Coupons

CreditNotes

CreditGrants

Customers

Entitlements

EntityIntegrationMappings

Environments

Events

Features

Groups

Integrations

Invoices

Payments

Plans

PriceUnits

Prices

Rbac

ScheduledTasks

Secrets

Subscriptions

Tasks

TaxAssociations

TaxRates

Tenants

Users

Wallets

Webhooks

File uploads

Certain SDK methods accept file objects as part of a request body or multi-part request. It is possible and typically recommended to upload files as a stream rather than reading the entire contents into memory. This avoids excessive memory consumption and potentially crashing with out-of-memory errors when working with very large files. The following example demonstrates how to attach a file stream to a request.

[!TIP]

For endpoints that handle file uploads bytes arrays can also be used. However, using streams is recommended for large files.

from flexprice_sdk_test import FlexPrice


with FlexPrice(
    server_url="https://api.example.com",
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:

    res = flex_price.events.post_events_analytics(request=open("example.file", "rb"))

    # Handle response
    print(res)

Retries

Some of the endpoints in this SDK support retries. If you use the SDK without any configuration, it will fall back to the default retry strategy provided by the API. However, the default retry strategy can be overridden on a per-operation basis, or across the entire SDK.

To change the default retry strategy for a single API call, simply provide a RetryConfig object to the call:

from flexprice_sdk_test import FlexPrice
from flexprice_sdk_test.utils import BackoffStrategy, RetryConfig


with FlexPrice(
    server_url="https://api.example.com",
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:

    res = flex_price.addons.get_addons(,
        RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False))

    # Handle response
    print(res)

If you'd like to override the default retry strategy for all operations that support retries, you can use the retry_config optional parameter when initializing the SDK:

from flexprice_sdk_test import FlexPrice
from flexprice_sdk_test.utils import BackoffStrategy, RetryConfig


with FlexPrice(
    server_url="https://api.example.com",
    retry_config=RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False),
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:

    res = flex_price.addons.get_addons()

    # Handle response
    print(res)

Error Handling

FlexPriceError is the base class for all HTTP error responses. It has the following properties:

Property Type Description
err.message str Error message
err.status_code int HTTP response status code eg 404
err.headers httpx.Headers HTTP response headers
err.body str HTTP body. Can be empty string if no body is returned.
err.raw_response httpx.Response Raw HTTP response
err.data Optional. Some errors may contain structured data. See Error Classes.

Example

from flexprice_sdk_test import FlexPrice
from flexprice_sdk_test.models import errors


with FlexPrice(
    server_url="https://api.example.com",
    api_key_auth="<YOUR_API_KEY_HERE>",
) as flex_price:
    res = None
    try:

        res = flex_price.addons.get_addons()

        # Handle response
        print(res)


    except errors.FlexPriceError as e:
        # The base class for HTTP error responses
        print(e.message)
        print(e.status_code)
        print(e.body)
        print(e.headers)
        print(e.raw_response)

        # Depending on the method different errors may be thrown
        if isinstance(e, errors.ErrorsErrorResponse):
            print(e.data.error)  # Optional[components.ErrorsErrorDetail]
            print(e.data.success)  # Optional[bool]

Error Classes

Primary errors:

Less common errors (5)

Network errors:

Inherit from FlexPriceError:

  • ResponseValidationError: Type mismatch between the response data and the expected Pydantic model. Provides access to the Pydantic validation error via the cause attribute.

* Check the method documentation to see if the error is applicable.

Custom HTTP Client

The Python SDK makes API calls using the httpx HTTP library. In order to provide a convenient way to configure timeouts, cookies, proxies, custom headers, and other low-level configuration, you can initialize the SDK client with your own HTTP client instance. Depending on whether you are using the sync or async version of the SDK, you can pass an instance of HttpClient or AsyncHttpClient respectively, which are Protocol's ensuring that the client has the necessary methods to make API calls. This allows you to wrap the client with your own custom logic, such as adding custom headers, logging, or error handling, or you can just pass an instance of httpx.Client or httpx.AsyncClient directly.

For example, you could specify a header for every request that this sdk makes as follows:

from flexprice_sdk_test import FlexPrice
import httpx

http_client = httpx.Client(headers={"x-custom-header": "someValue"})
s = FlexPrice(client=http_client)

or you could wrap the client with your own custom logic:

from flexprice_sdk_test import FlexPrice
from flexprice_sdk_test.httpclient import AsyncHttpClient
import httpx

class CustomClient(AsyncHttpClient):
    client: AsyncHttpClient

    def __init__(self, client: AsyncHttpClient):
        self.client = client

    async def send(
        self,
        request: httpx.Request,
        *,
        stream: bool = False,
        auth: Union[
            httpx._types.AuthTypes, httpx._client.UseClientDefault, None
        ] = httpx.USE_CLIENT_DEFAULT,
        follow_redirects: Union[
            bool, httpx._client.UseClientDefault
        ] = httpx.USE_CLIENT_DEFAULT,
    ) -> httpx.Response:
        request.headers["Client-Level-Header"] = "added by client"

        return await self.client.send(
            request, stream=stream, auth=auth, follow_redirects=follow_redirects
        )

    def build_request(
        self,
        method: str,
        url: httpx._types.URLTypes,
        *,
        content: Optional[httpx._types.RequestContent] = None,
        data: Optional[httpx._types.RequestData] = None,
        files: Optional[httpx._types.RequestFiles] = None,
        json: Optional[Any] = None,
        params: Optional[httpx._types.QueryParamTypes] = None,
        headers: Optional[httpx._types.HeaderTypes] = None,
        cookies: Optional[httpx._types.CookieTypes] = None,
        timeout: Union[
            httpx._types.TimeoutTypes, httpx._client.UseClientDefault
        ] = httpx.USE_CLIENT_DEFAULT,
        extensions: Optional[httpx._types.RequestExtensions] = None,
    ) -> httpx.Request:
        return self.client.build_request(
            method,
            url,
            content=content,
            data=data,
            files=files,
            json=json,
            params=params,
            headers=headers,
            cookies=cookies,
            timeout=timeout,
            extensions=extensions,
        )

s = FlexPrice(async_client=CustomClient(httpx.AsyncClient()))

Resource Management

The FlexPrice class implements the context manager protocol and registers a finalizer function to close the underlying sync and async HTTPX clients it uses under the hood. This will close HTTP connections, release memory and free up other resources held by the SDK. In short-lived Python programs and notebooks that make a few SDK method calls, resource management may not be a concern. However, in longer-lived programs, it is beneficial to create a single SDK instance via a context manager and reuse it across the application.

from flexprice_sdk_test import FlexPrice
def main():

    with FlexPrice(
        server_url="https://api.example.com",
        api_key_auth="<YOUR_API_KEY_HERE>",
    ) as flex_price:
        # Rest of application here...


# Or when using async:
async def amain():

    async with FlexPrice(
        server_url="https://api.example.com",
        api_key_auth="<YOUR_API_KEY_HERE>",
    ) as flex_price:
        # Rest of application here...

Debugging

You can setup your SDK to emit debug logs for SDK requests and responses.

You can pass your own logger class directly into your SDK.

from flexprice_sdk_test import FlexPrice
import logging

logging.basicConfig(level=logging.DEBUG)
s = FlexPrice(server_url="https://example.com", debug_logger=logging.getLogger("flexprice_sdk_test"))

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

flexprice_sdk_test-2.0.0.tar.gz (210.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

flexprice_sdk_test-2.0.0-py3-none-any.whl (460.4 kB view details)

Uploaded Python 3

File details

Details for the file flexprice_sdk_test-2.0.0.tar.gz.

File metadata

  • Download URL: flexprice_sdk_test-2.0.0.tar.gz
  • Upload date:
  • Size: 210.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.11.14

File hashes

Hashes for flexprice_sdk_test-2.0.0.tar.gz
Algorithm Hash digest
SHA256 e069d0268f3489af8c679b3e11c6a6f441945310e6a0aec692f53d0fab47feac
MD5 20d881fff5a4284a4ab2de98410d57a0
BLAKE2b-256 991c9d19380524074044fbbd7c92068b771e7ff8748f466be40964ae52e44f0e

See more details on using hashes here.

File details

Details for the file flexprice_sdk_test-2.0.0-py3-none-any.whl.

File metadata

File hashes

Hashes for flexprice_sdk_test-2.0.0-py3-none-any.whl
Algorithm Hash digest
SHA256 cb39f2d08aec0aa9421044386db4bf1b17893e485034fd7ff86b86d15acffc91
MD5 5069472f0052cd91f4dab8983a92c26c
BLAKE2b-256 ae952497afb21a9c75524cb4583afaf0749cbed61abd4daf035e1bc7e4686a43

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page