opa-sh

Official Python SDK for the Opa link shortener API (https://api.opa.sh/v1).

Resource-oriented, fully typed (pydantic v2 models, py.typed), sync and async clients side by side, idiomatic Python error handling — failures raise typed exceptions, not a Result/tuple you have to unwrap.

Install

pip install opa-sh
# or: uv add opa-sh
# or: poetry add opa-sh

Quick start

from opa_sh import OpaClient

opa = OpaClient(api_key="opa_...")

link = opa.links.create(
    destination_url="https://example.com/black-friday",
    domain="opa.sh",
    tag_ids=["tag_campaign2026"],
)

print(link.short_link)
print(link.destination_url)

opa.close()

Or as a context manager, which closes the underlying connection pool for you:

with OpaClient(api_key="opa_...") as opa:
    link = opa.links.create(destination_url="https://example.com")
    print(link.short_link)

Async

Every resource method has an async counterpart on AsyncOpaClient — same names, same arguments, awaited:

import asyncio
from opa_sh import AsyncOpaClient


async def main() -> None:
    async with AsyncOpaClient(api_key="opa_...") as opa:
        link = await opa.links.create(destination_url="https://example.com")
        print(link.short_link)


asyncio.run(main())

Get your API key at opa.sh/settings/api-keys.

Authentication

Server-side only — never expose a key from a browser bundle or a mobile app:

opa = OpaClient(api_key="opa_...")

Base URL defaults to https://api.opa.sh/v1. Override it for staging or a self-hosted instance:

opa = OpaClient(api_key="...", base_url="https://api.staging.opa.sh/v1")

Error handling

Every failed call raises a subclass of OpaError — a Python-idiomatic raise-based SDK (mirrors Stripe, OpenAI, Anthropic, boto3), not a Result/Either pattern. Catch OpaError to handle any SDK failure, or a specific subclass to handle one case:

from opa_sh import OpaClient
from opa_sh.errors import NotFoundError, RateLimitError, ValidationError, OpaError

opa = OpaClient(api_key="...")

try:
    link = opa.links.get("lnk_xxx")
except NotFoundError:
    ...  # 404
except ValidationError as e:
    print(e.details)  # e.g. {"issues": [{"path": "destinationUrl", "message": "Required"}]}
except RateLimitError as e:
    print("retry after", e.retry_after, "seconds")
except OpaError as e:
    # catch-all: e.code, e.status, e.message, e.details
    print(e.code, e.status, e.message)

All of the above subclass OpaError, which subclasses Exception.

Resources

Links

# Create — only destination_url is required
link = opa.links.create(
    destination_url="https://example.com",
    domain="opa.sh",
    key="custom-slug",  # optional — omit for a random key
    tag_ids=["tag_xxx"],
    password="s3cret",
    expires_at="2027-01-01T00:00:00Z",
)

# Read
link = opa.links.get("lnk_xxx")

# Update — destination_url and domain are required (full replace, not a merge patch)
link = opa.links.update("lnk_xxx", destination_url="https://example.com/new", domain="opa.sh")

# Lifecycle
opa.links.archive("lnk_xxx")  # reversible — the link keeps resolving
opa.links.restore("lnk_xxx")
opa.links.duplicate("lnk_xxx")

# List — single page
page = opa.links.list(search="marketing", limit=50)
print(page.items, page.has_more, page.next_cursor)

# List — every page, auto-paginating (memory-safe generator)
for link in opa.links.list_all(search="marketing"):
    print(link.short_link, link.tags)

# Bulk operations
opa.links.bulk_archive(["lnk_a", "lnk_b", "lnk_c"])
opa.links.bulk_restore(["lnk_a", "lnk_b"])
opa.links.bulk_move(["lnk_a", "lnk_b"], "folder_xxx")
opa.links.bulk_tag(["lnk_a", "lnk_b"], ["tag_q1"])

Analytics

summary = opa.analytics.summary(from_="2026-01-01", to="2026-01-31", link_id="lnk_xxx")
print(summary.clicks, summary.unique_clicks)

timeseries = opa.analytics.timeseries(from_="2026-01-01", to="2026-01-31", link_id="lnk_xxx")
for point in timeseries.points:
    print(point.date, point.clicks)

# Raw click events — single page or auto-paginating
page = opa.analytics.events(link_id="lnk_xxx", limit=100)
for event in opa.analytics.events_all(link_id="lnk_xxx"):
    print(event.timestamp, event.country, event.device)

from_/to are ISO dates (max 366-day range). The trailing underscore on from_ avoids colliding with the from keyword — it's sent as from on the wire.

Domains

domains = opa.domains.list()
for domain in domains:
    print(domain.domain, domain.primary, domain.verified)

Not paginated — this list is small by nature (verified custom domains plus the shared app domain).

Track

opa.track is stateless and safe to share across concurrent server requests. Every call carries its own identity; identify() never sets a current user on the client. AsyncOpaClient.track exposes the same methods with await.

# Bind the anonymous id generated by @opa.sh/analytics to your own user id.
opa.track.identify(
    anonymous_id="anon_123",
    external_id="user_123",
    traits={"email": "person@example.com", "plan": "pro"},
)

# Events may reference either an anonymous identity or an external identity.
opa.track.event(
    event_id="evt_checkout_123",  # stable idempotency key
    event_name="checkout_started",
    anonymous_id="anon_123",
    properties={"plan": "pro"},
)

# Legacy conversion contracts remain available server-side.
opa.track.lead(
    click_id="clk_123",
    event_name="Signup",
    customer_external_id="user_123",
)

opa.track.sale(
    customer_external_id="user_123",
    amount=14990,  # minor units (cents for BRL/USD)
    currency="BRL",
    invoice_id="inv_123",  # stable idempotency key
)

The SDK automatically retries transient failures. Reuse the same event_id or invoice_id across retries and webhook redeliveries so the API can deduplicate the operation. This server SDK does not generate browser identities, cookies, sessions, consent state, or pageviews; forward an anonymous_id captured by the browser integration when you need to reconcile anonymous history.

Retries and rate limiting

Automatic exponential backoff (with jitter) on 5xx and 429, respecting the Retry-After header. Never retries other 4xx statuses, so idempotent semantics are preserved for POST. Transport-level failures (DNS, connection reset) are retried the same way, then surfaced as NetworkError if every attempt fails.

opa = OpaClient(
    api_key="...",
    retries=3,  # default: 3
    retry_delay=1.0,  # default: 1.0s base, doubles each attempt (capped at 20s)
)

Pagination

Every list-style method comes in two shapes:

• list(...) (or events(...)) — returns a single Page[T], with .items, .has_more, and .next_cursor.
• list_all(...) (or events_all(...)) — a generator that walks every page automatically, yielding items one at a time. Memory-safe: it never holds more than one page in memory.

# Sync generator
for link in opa.links.list_all():
    ...

# Async generator — same method name, `async for`
async for link in opa.links.list_all():
    ...

An error partway through iteration raises the same OpaError subclass you'd get from the eager list() call — you already handle that exception type, so this doesn't introduce a new failure mode.

Custom httpx client

For advanced configuration — custom transports, proxies, mTLS, connection pooling tuned for your workload — pass your own httpx.Client / httpx.AsyncClient. The SDK merges the auth header into it and does not override anything else you've configured:

import httpx
from opa_sh import OpaClient

custom_client = httpx.Client(
    base_url="https://api.opa.sh/v1",
    limits=httpx.Limits(max_connections=50),
    proxy="http://localhost:8080",
)
opa = OpaClient(api_key="...", http_client=custom_client)

Note: the SDK's own retry transport (opa_sh.retry.RetryTransport) only applies when it builds the httpx.Client itself. A custom http_client is responsible for its own retry/backoff behavior.

Types

Every response is a pydantic v2 model (Link, LinkSummary, Domain, AnalyticsSummary, AnalyticsTimeseries, AnalyticsEvent, ...) with snake_case attributes and full type hints — the package ships a py.typed marker, so mypy/pyright pick up types automatically in consuming projects.

from opa_sh import Link, LinkSummary, Domain, Page


def handle(link: Link) -> None:
    print(link.short_link)

Runtime support

• Python 3.9, 3.10, 3.11, 3.12, 3.13
• CPython and PyPy (no C-extension dependencies beyond what httpx/pydantic already require)

Contributing

The SDK's request/response shapes are hand-written pydantic models in src/opa_sh/models.py, kept in sync by hand against openapi/v1.json (the raw OpenAPI spec has no named component schemas, so a mechanical generator produces synthetic, unusable names — see that file's module docstring for detail). src/opa_sh/_generated/ is a raw client generated by openapi-python-client from the same spec; it's committed as a diff target for the sync-openapi workflow but is not imported at runtime.

openapi/v1.json is refreshed daily from https://api.opa.sh/v1/openapi via the sync-openapi workflow, which opens a PR when the spec changes — extend it with matching updates to models.py and the relevant resources/*.py methods.

The reserved tracking endpoints temporarily use the hand-written contract in resources/track.py until the canonical OpenAPI document publishes them.

git clone https://github.com/espocalabs/opa-sdk-python
cd opa-sdk-python
uv sync --all-extras
uv run pytest
uv run ruff check .
uv run mypy src
uv build

License

MIT © Espoca Labs