Important

This SDK is now deprecated and archived.

We now have a new Python SDK with a new home: https://github.com/polarsource/polar/tree/main/sdk/python

polar-sdk

Table of Contents

SDK Installation

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.

uv

uv add polar-sdk

PIP

pip install polar-sdk

Poetry

poetry add polar-sdk

Shell and script usage with uv

uvx --from polar-sdk python

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

from polar_sdk import Polar

sdk = Polar(
  # SDK arguments
)

# Rest of script here...

IDE Support

PyCharm

SDK Example Usage

Example

# Synchronous Example
from polar_sdk import Polar


with Polar(
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

# Asynchronous Example
import asyncio
from polar_sdk import Polar

async def main():

    async with Polar(
        access_token="<YOUR_BEARER_TOKEN_HERE>",
    ) as polar:

        res = await polar.organizations.list_organizations_async(page=1, limit=10)

        while res is not None:
            # Handle items

            res = res.next()

asyncio.run(main())

Webhook support

from flask import Flask, request
from polar_sdk.webhooks import (
    validate_event,
    WebhookUnknownTypeError,
    WebhookVerificationError,
)

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    try:
        event = validate_event(
            payload=request.data,
            headers=request.headers,
            secret='<YOUR_WEBHOOK_SECRET>',
        )

        # Process the event

        return "", 202
    except WebhookVerificationError:
        return "", 403
    except WebhookUnknownTypeError:
        # The signature was valid, but the event `type` is not known to this
        # SDK version (e.g. the API has shipped a new event type and you
        # haven't upgraded yet). Safe to ignore for forward compatibility,
        # or log and acknowledge so the webhook isn't retried.
        return "", 202

Available Resources and Operations

Retries

from polar_sdk import Polar
from polar_sdk.utils import BackoffStrategy, RetryConfig


with Polar(
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10,
        RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False))

    while res is not None:
        # Handle items

        res = res.next()

from polar_sdk import Polar
from polar_sdk.utils import BackoffStrategy, RetryConfig


with Polar(
    retry_config=RetryConfig("backoff", BackoffStrategy(1, 50, 1.1, 100), False),
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Error Handling

Example

import polar_sdk
from polar_sdk import Polar, models


with Polar(
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:
    res = None
    try:

        res = polar.organizations.list_organizations(page=1, limit=10)

        while res is not None:
            # Handle items

            res = res.next()


    except models.PolarError 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, models.HTTPValidationError):
            print(e.data.detail)  # Optional[List[polar_sdk.ValidationError]]

Error Classes

Server Selection

Select Server by Name

Example

from polar_sdk import Polar


with Polar(
    server="production",
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Override Server URL Per-Client

from polar_sdk import Polar


with Polar(
    server_url="https://api.polar.sh",
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Custom HTTP Client

from polar_sdk import Polar
import httpx

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

from polar_sdk import Polar
from polar_sdk.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 = Polar(async_client=CustomClient(httpx.AsyncClient()))

Authentication

Per-Client Security Schemes

from polar_sdk import Polar


with Polar(
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Per-Operation Security Schemes

import polar_sdk
from polar_sdk import Polar


with Polar() as polar:

    res = polar.customer_portal.benefit_grants.list(security=polar_sdk.CustomerPortalBenefitGrantsListSecurity(

    ), page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Resource Management

from polar_sdk import Polar
def main():

    with Polar(
        access_token="<YOUR_BEARER_TOKEN_HERE>",
    ) as polar:
        # Rest of application here...


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

    async with Polar(
        access_token="<YOUR_BEARER_TOKEN_HERE>",
    ) as polar:
        # Rest of application here...

Debugging

from polar_sdk import Polar
import logging

logging.basicConfig(level=logging.DEBUG)
s = Polar(debug_logger=logging.getLogger("polar_sdk"))

Pagination

from polar_sdk import Polar


with Polar(
    access_token="<YOUR_BEARER_TOKEN_HERE>",
) as polar:

    res = polar.organizations.list_organizations(page=1, limit=10)

    while res is not None:
        # Handle items

        res = res.next()

Summary

Development

Maturity

Contributions

SDK Created by Speakeasy