tierforge 0.1.0

TierForge Python SDK

tierforge

Official Python client for the TierForge REST API. Sync and async clients, Pydantic v2 models, and webhook verification helpers.

Installation

pip install tierforge
uv add tierforge
poetry add tierforge

Runtime requirements

• Python 3.9+
• Pydantic v2 (installed automatically)

Quickstart (sync)

import os

from tierforge import TierForge

with TierForge(api_key=os.environ["TIERFORGE_API_KEY"]) as tf:
    for product in tf.products.list():
        print(product.name)

Optional DX SLO telemetry

Opt-in timing logs for the DX SLO (tierforge.prd11.sdk.time_to_first_call_ms). Never logs API keys — only duration_ms, SDK name, version, and language.

import json
import os
import time
from importlib.metadata import version

from tierforge import TierForge

# Optional DX SLO telemetry (opt-in — remove if you do not want timing logs)
dx_start = time.perf_counter()
with TierForge(api_key=os.environ["TIERFORGE_API_KEY"]) as tf:
    for product in tf.products.list():
        duration_ms = int((time.perf_counter() - dx_start) * 1000)
        print(
            json.dumps(
                {
                    "tag": "[tierforge.metric]",
                    "metric": "tierforge.prd11.sdk.time_to_first_call_ms",
                    "sdk": "tierforge",
                    "version": version("tierforge"),
                    "language": "python",
                    "duration_ms": duration_ms,
                }
            )
        )
        print(product.name)
        break

Quickstart (async)

import asyncio
import os

from tierforge import AsyncTierForge

async def main() -> None:
    async with AsyncTierForge(api_key=os.environ["TIERFORGE_API_KEY"]) as tf:
        async for product in tf.products.list():
            print(product.name)

asyncio.run(main())

base_url defaults to https://www.tierforge.dev/api/v1. Override for local dev, e.g. http://127.0.0.1:3040/api/v1.

Resources

Attribute	Methods	Python naming notes
products	create, list, get, update, archive
dimensions	create, list, get, update, archive
meters	create, list, get, update, archive
plans	create, list, get, update, archive
plan_versions	create, list, get, update, publish, deprecate, delete	snake_case vs TS planVersions
customer_accounts	create, list, get, update, archive	cursor pagination
customer_users	create, list, get, update, archive
billing_accounts	create, list, get, update, archive
subscriptions	create, list, get, update, cancel, change_plan, extend_trial, end_trial_now
trial_consumptions	list, get, void	PRD 16 trial surface
usage_events	create, create_batch	max 100 events per batch
usage_aggregates	get_trial_window	trial window rollup
entitlements	for_subscription, for_customer_account, for_customer_user, usage_summary
webhook_endpoints	create, list, get, update, rotate_secret, archive

API reference: mkdocstrings output (generated on release).

Pagination

import os

from tierforge import TierForge

with TierForge(api_key=os.environ["TIERFORGE_API_KEY"]) as tf:
    for account in tf.customer_accounts.list(limit=50):
        print(account.id)

    for page in tf.customer_accounts.list().per_page():
        print(len(page.items), page.next_cursor)

Async:

async for account in tf.customer_accounts.list():
    ...
async for page in tf.usage_events.list().per_page():
    ...

Idempotency

Write methods attach Idempotency-Key automatically and reuse it across retries:

tf.products.create(body, idempotency_key="my-stable-key")

Error handling

import os

from tierforge import (
    AsyncTierForge,
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ServerError,
    TierForgeError,
    TrialEligibilityBlockedError,
    TransientNetworkError,
    ValidationError,
)

async def demo() -> None:
    async with AsyncTierForge(api_key=os.environ["TIERFORGE_API_KEY"]) as tf:
        try:
            await tf.products.get("00000000-0000-4000-8000-000000000000")
        except ValidationError as err:
            print("validation", err.issues)
        except AuthenticationError:
            print("check TIERFORGE_API_KEY")
        except NotFoundError as err:
            print("not found", err.status_code)
        except RateLimitError as err:
            print("rate limited", err.retry_after)
        except TrialEligibilityBlockedError as err:
            print("trial blocked", err.code)
        except ServerError as err:
            print("server error", err.status_code)
        except TransientNetworkError as err:
            print("network", err)
        except TierForgeError as err:
            print("api", err.status_code, err.message)

Webhook verification

Import verify_webhook_signature from tierforge (re-exported). Read the raw body before JSON parsing. Secret from os.environ["TIERFORGE_WEBHOOK_SECRET"].

Canonical framework examples live in tierforge/utils/webhook_helpers.py (FastAPI, Flask, Django, aiohttp). Copy those patterns into your handler.

import json
import os

from fastapi import FastAPI, Request, Response
from tierforge import verify_webhook_signature

app = FastAPI()

@app.post("/webhooks/tierforge")
async def tierforge_webhook(request: Request) -> Response:
    raw_body = await request.body()
    signature = request.headers.get("x-tierforge-signature", "")
    secret = os.environ["TIERFORGE_WEBHOOK_SECRET"]
    if not verify_webhook_signature(raw_body, signature, secret):
        return Response(status_code=401, content="invalid signature")
    event = json.loads(raw_body)
    return Response(
        status_code=200,
        content=json.dumps({"received": True, "id": event["id"]}),
    )

Debug mode

Pass debug=True to TierForge / AsyncTierForge. Logs use Python logging with credential redaction. Disable in production.

Type hints

Response models are Pydantic v2 classes generated from OpenAPI (tierforge.types.models). Use .model_validate for custom parsing and model_dump() for serialization.

Migration

See MIGRATING.md.

Contributing

See CONTRIBUTING.md.

License

MIT — see LICENSE.