itaja

Sell subscriptions, credits and licence keys from your Python server. Declare what you sell in code, push it to Itaja, mount one webhook route (Django, FastAPI, Starlette, Flask), react to typed events.

pip install itaja          # uv add itaja · poetry add itaja

Requires Python 3.10+.

Server only. This package holds your secret API key. Never ship it to a browser or a mobile app.

Quick start

0. Configure

Put your keys in .env (never commit it) and pass them to create_itaja:

# .env
ITAJA_API_KEY=itjs_…              # sandbox key; an itjp_… key works on the live catalogue
ITAJA_WEBHOOK_SECRET=whsec_…      # signing secret of your webhook endpoint (dashboard)
ITAJA_API_URL=https://api.getitaja.com/api/v1   # optional

import os
from itaja import create_itaja

itaja = create_itaja(
    api_key=os.environ["ITAJA_API_KEY"],
    webhook_secret=os.environ["ITAJA_WEBHOOK_SECRET"],
    base_url=os.environ.get("ITAJA_API_URL"),
    catalog=[...],
)

Each argument falls back to its variable when you leave it out. Pass other values to use another account, a tenant's key or a staging API. The itaja command loads .env and .env.local itself: nothing to export in your shell. In your app, load .env the way you already do (django-environ, pydantic-settings, python-dotenv…).

1. Declare what you sell

A free plan and a paid plan, with a quota that depends on the plan.

# itaja_config.py
import os
from itaja import create_itaja, feature, group, plan

# What the customer can do
messages = feature("Messages", "metered")
export = feature("Export", "boolean")

# One customer holds one plan of a group
app = group("My App")

free = plan(
    "Free", app,
    default=True,  # everyone starts here, no payment
    includes=[messages(limit=100, reset="month")],
)

pro = plan(
    "Pro", app,
    price={"amount": 5000, "currency": "XOF", "interval": "month"},
    trial_days=7,
    includes=[messages(limit=2000, reset="month"), export()],
)

itaja = create_itaja(
    api_key=os.environ.get("ITAJA_API_KEY"),
    webhook_secret=os.environ.get("ITAJA_WEBHOOK_SECRET"),
    catalog=[messages, export, free, pro],
)

Everything is identified by its name — there is no id to choose. Names are compared trimmed, inner spaces collapsed and case-insensitively ("Pro" and " pro " are the same plan). Product names are unique across all your products, feature names across your features, plan names within their group, pack names within their credits product. Rename an item and you declare a new one; the old one is archived. Amounts are in major units (5000 XOF, 9.99 EUR), as in the API. Mistakes the API would refuse — a blank or over-long name (200 characters at most), two items of one name, an undeclared feature, two default plans, a paid plan without a price — raise ItajaConfigError as soon as the module is imported.

2. Push it

itaja push                  # shows the diff, asks, applies

push uses the api_key of the client in itaja_config.py. Run it again: nothing changes. Change a price and a new version of the plan is created; running subscriptions keep theirs. Whatever you remove is archived, never deleted, and products made in the dashboard are never touched.

The key's prefix picks the mode: itjs_ is sandbox, itjp_ is live.

3. Mount the webhook

One route receives Itaja's events. It checks the signature on the raw body, ignores a delivery it already handled, ignores events of the other mode, and answers 500 when your handler raises so Itaja delivers again. It checks deliveries with webhook_secret.

Django (pip install "itaja[django]")

# urls.py
from django.urls import path
from itaja.integrations.django import webhook_view
from itaja_config import itaja

urlpatterns = [path("itaja/webhook", webhook_view(itaja))]  # CSRF-exempt, POST only

FastAPI (pip install "itaja[fastapi]")

from fastapi import FastAPI
from itaja.integrations.starlette import webhook_router
from itaja_config import itaja

app = FastAPI()
app.include_router(webhook_router(itaja))  # POST /itaja/webhook

Starlette: Route("/itaja/webhook", webhook_endpoint(itaja), methods=["POST"]) from itaja.integrations.starlette.

Flask (pip install "itaja[flask]")

from flask import Flask
from itaja.integrations.flask import webhook_blueprint
from itaja_config import itaja

app = Flask(__name__)
app.register_blueprint(webhook_blueprint(itaja))  # POST /itaja/webhook

Anything else: itaja.webhooks.process(raw_body, headers) (or await itaja.webhooks.process_async(...)) returns the status and body to answer.

4. React to events

@itaja.on("checkout.paid")
def fulfil(event):
    Order.fulfil(event.data["checkoutId"])

@itaja.on("subscription.canceled")
async def downgrade(event):        # async handlers work too
    await accounts.downgrade(event.data["customerId"])

Or pass them at creation: create_itaja(..., on={"checkout.paid": fulfil}). event.data holds the fields Itaja documents for that type. The "*" handler receives every delivery, including types this version does not know yet.

Test a handler locally with a signed sample:

itaja trigger checkout.paid --url http://localhost:8000/itaja/webhook   # signs with ITAJA_WEBHOOK_SECRET from .env

What you can sell

Each one can live in the same catalog list.

Subscriptions

The example above is the whole declaration. Around it:

team = plan(
    "Team", app,
    price={"amount": 15000, "currency": "XOF", "interval": "month"},
    includes=[messages(limit=20000, reset="month"), export()],
)

• interval: week · month · quarter · year. reset on a quota: day · week · month · year.
• A plan without a price must be default=True; a group has at most one default.
• trial_days gives a free period before the first charge.

itaja.subscriptions.list(customerId=customer_id)
itaja.subscriptions.cancel(subscription_id)  # at period end

Events: invoice.paid · invoice.payment_failed · subscription.past_due · subscription.canceled.

Coming with the API: itaja.subscribe(customer_id=..., plan="Pro", product="My App") (new, upgrade, downgrade, cancel-to-free in one call; the plan named as in your catalogue, product optional, to pick between products that sell a plan of that name), check() and report() for entitlements exist and raise NOT_YET_AVAILABLE until the API serves them (itaja-services #465, #466).

Credits

A currency your app spends as the customer uses it, sold in packs.

from itaja import credit_pack, credits

tokens = credits("AI tokens", coin_name="Tokens")

small = credit_pack("100 tokens", tokens, 100, {"amount": 1000, "currency": "XOF"})
large = credit_pack("1000 tokens", tokens, 1000, {"amount": 8000, "currency": "XOF"})

checkout = itaja.credits.buy(
    pack="100 tokens",  # the pack's name, as the buyer reads it
    customer_id=customer_id,
    home_url="https://shop.example/thanks",
    product="AI tokens",
    success_url="https://shop.example/credits/thanks",  # optional, as on checkout.create
)
# redirect the buyer to checkout["checkoutUrl"]

itaja.credits.consume(customer_id=customer_id, amount=3, product="AI tokens")
itaja.credits.balance(customer_id=customer_id, product="AI tokens")

Events: credits.granted · credits.consumed · credits.expired · credits.low · credits.clawed_back.

Licence keys

Software that unlocks on a number of devices.

from itaja import license

desktop = license(
    "Desktop app",
    price={"amount": 25000, "currency": "XOF"},
    devices=3,             # None = unlimited
    duration="days-365",   # "lifetime" · "days-N" · "from-first-activation-days-N"
)

itaja.licenses.issue(product="Desktop app", customer_id=customer_id)
itaja.licenses.revoke(key_id, product="Desktop app", reason="chargeback")

Events: license.issued · activated · deactivated · revoked · expired · rotated · suspended · resumed · updated.

One-off payments

Nothing to declare: create a checkout for an amount, send the customer to checkoutUrl, and fulfil on checkout.paid.

checkout = itaja.checkout.create(
    homeUrl="https://shop.example/thanks",
    successUrl="https://shop.example/order/paid",  # optional
    cancelUrl="https://shop.example/cart",         # optional
    currency="XOF",
    lineItems=[{"name": "Sandals", "quantity": 1, "unitPrice": 5000}],
)

The payer comes back to homeUrl by default. Pass successUrl, failureUrl and/or cancelUrl (success_url… on credits.buy) (optional, absolute https, on a live key a domain verified for your account) to send each outcome to its own page; one left out keeps Itaja's return page. Itaja adds itaja_checkout_id and itaja_status (success | failure | cancel) to the address. The redirect is payer-visible: confirm payment from the webhook, never from it.

Payment links are a dashboard feature: no builder or endpoint here creates one.

Also in the SDK

itaja.customers.get("user_42")   # your own customer id works too
itaja.request("GET", "/me")      # any endpoint the helpers do not cover

Products and packs are referenced by their name at runtime too: the SDK resolves product="AI tokens" to Itaja's product once, by comparing the normalised name with what your last push created.

Verify a delivery yourself, without the route:

from itaja import construct_event

event = construct_event(raw_body, headers, secret=os.environ["ITAJA_WEBHOOK_SECRET"])

Behind several processes or servers, share the processed-delivery store (it is in memory by default) with any object that has has(event_id) and add(event_id):

class RedisDedupe:
    def has(self, event_id): return bool(redis.exists(f"itaja:{event_id}"))
    def add(self, event_id): redis.set(f"itaja:{event_id}", 1, ex=86400)

itaja = create_itaja(..., dedupe=RedisDedupe())

Errors and retries

Every failure is an ItajaError with a stable code, the HTTP status and the API's details: ItajaValidationError (400/422), ItajaAuthenticationError (401), ItajaPermissionError (403), ItajaNotFoundError (404), ItajaConflictError (409), ItajaRateLimitError (429), ItajaServerError (5xx), ItajaConnectionError, ItajaSignatureError, ItajaConfigError. No error ever contains your key.

Every write carries an Idempotency-Key, one per call and reused on its retries, so a retried write never runs twice. Network errors, 5xx, 429 and 409 IDEMPOTENCY_KEY_IN_PROGRESS are retried twice, honouring Retry-After. Pass idempotency_key= (an order id, say) to make your own retries safe too.

Configuration

Every argument of create_itaja, and the variable it falls back to:

There is no environment argument: the key's prefix decides sandbox or live.

Development

uv sync
uv run pytest
uv run mypy
uv run ruff check

spec/catalog.schema.json and spec/signature-vectors.json (repository root) are shared with every Itaja SDK: this package builds, field for field, the catalogue the TypeScript SDK builds (spec/conformance/reference-catalog.json), and accepts every webhook the backend's own signer produced.

License

MIT