SDK API

Core

Configuration with precedence

explicit kwargs > environment (.env) > defaults

Server-side calls require BOTH jwt and secret.

Source code in univapay/config.py

@dataclass
class UnivapayConfig:
    """
    Configuration with precedence:
      explicit kwargs > environment (.env) > defaults

    Server-side calls require BOTH jwt and secret.
    """

    # Credentials
    jwt: Optional[str] = None
    secret: Optional[str] = None

    # Routing / network
    store_id: Optional[str] = None
    base_url: Optional[str] = None
    timeout: Optional[float] = None

    # Diagnostics
    debug: Optional[bool] = None

    # Optional retry hints (client may use these; also available via env)
    # UNIVAPAY_RETRIES, UNIVAPAY_BACKOFF
    retries: Optional[int] = None
    backoff_factor: Optional[float] = None

    # Internal: where each field was sourced from (arg/env/default) for debugging
    _source: Dict[str, str] = field(default_factory=dict, init=False, repr=False)

    def __post_init__(self) -> None:
        env = os.environ

        # jwt
        if self.jwt is None or self.jwt == "":
            self.jwt = env.get("UNIVAPAY_JWT", "")
            self._source["jwt"] = "env"
        else:
            self._source["jwt"] = "arg"

        # secret
        if self.secret is None or self.secret == "":
            self.secret = env.get("UNIVAPAY_SECRET", "")
            self._source["secret"] = "env"
        else:
            self._source["secret"] = "arg"

        # store_id
        if self.store_id is None or self.store_id == "":
            self.store_id = env.get("UNIVAPAY_STORE_ID") or None
            self._source["store_id"] = "env"
        else:
            self._source["store_id"] = "arg"

        # base_url
        if self.base_url is None or self.base_url == "":
            self.base_url = _normalize_base_url(env.get("UNIVAPAY_BASE_URL"))
            self._source["base_url"] = "env/default"
        else:
            self.base_url = _normalize_base_url(self.base_url)
            self._source["base_url"] = "arg"

        # timeout
        if self.timeout is None:
            self.timeout = _parse_float(env.get("UNIVAPAY_TIMEOUT"), 30.0)
            self._source["timeout"] = "env/default"
        else:
            try:
                self.timeout = float(self.timeout)
            except Exception:
                self.timeout = 30.0
            self._source["timeout"] = "arg"

        # debug
        if self.debug is None:
            self.debug = _parse_bool(env.get("UNIVAPAY_DEBUG", "1"), True)
            self._source["debug"] = "env/default"
        else:
            self.debug = bool(self.debug)
            self._source["debug"] = "arg"

        # optional retry hints
        if self.retries is None:
            self.retries = _parse_int(env.get("UNIVAPAY_RETRIES"), 0)
            self._source["retries"] = "env/default"
        else:
            try:
                self.retries = int(self.retries)
            except Exception:
                self.retries = 0
            self._source["retries"] = "arg"

        if self.backoff_factor is None:
            self.backoff_factor = _parse_float(env.get("UNIVAPAY_BACKOFF"), 0.5)
            self._source["backoff_factor"] = "env/default"
        else:
            try:
                self.backoff_factor = float(self.backoff_factor)
            except Exception:
                self.backoff_factor = 0.5
            self._source["backoff_factor"] = "arg"

        # Initial debug print (sanitized)
        _dprint(
            bool(self.debug),
            "Loaded config:",
            {
                "jwt": _mask(self.jwt, "jwt"),
                "secret": _mask(self.secret, "secret"),
                "store_id": self.store_id or None,
                "base_url": self.base_url,
                "timeout": self.timeout,
                "debug": self.debug,
                "source": self._source,
            },
        )

    # -------- validation & utils --------
    def validate(self) -> "UnivapayConfig":
        """
        Validate presence of credentials for server-side API calls.
        (Widget-only use-cases can bypass by not instantiating this config.)
        """
        if not self.jwt or not self.secret:
            _dprint(
                bool(self.debug),
                "Validation failed: jwt/secret missing",
                {"jwt": _mask(self.jwt, "jwt"), "secret": _mask(self.secret, "secret")},
            )
            raise UnivapayConfigError(
                "UNIVAPAY_JWT and UNIVAPAY_SECRET are required for server-side API calls."
            )
        _dprint(bool(self.debug), "Validation OK")
        return self

    def require_store_id(self) -> str:
        """Ensure a store_id is present for endpoints that need it."""
        if not self.store_id:
            raise UnivapayConfigError("UNIVAPAY_STORE_ID is required for this operation.")
        return self.store_id

    def masked(self) -> dict:
        """Return a sanitized dict for logging/diagnostics."""
        return {
            "jwt": _mask(self.jwt, "jwt"),
            "secret": _mask(self.secret, "secret"),
            "store_id": self.store_id,
            "base_url": self.base_url,
            "timeout": self.timeout,
            "debug": self.debug,
            "retries": self.retries,
            "backoff_factor": self.backoff_factor,
        }

    def copy_with(
        self,
        *,
        jwt: Optional[str] = None,
        secret: Optional[str] = None,
        store_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        debug: Optional[bool] = None,
        retries: Optional[int] = None,
        backoff_factor: Optional[float] = None,
    ) -> "UnivapayConfig":
        """Create a modified copy (handy in tests)."""
        return replace(
            self,
            jwt=self.jwt if jwt is None else jwt,
            secret=self.secret if secret is None else secret,
            store_id=self.store_id if store_id is None else store_id,
            base_url=_normalize_base_url(base_url if base_url is not None else self.base_url),
            timeout=self.timeout if timeout is None else float(timeout),
            debug=self.debug if debug is None else bool(debug),
            retries=self.retries if retries is None else int(retries),
            backoff_factor=self.backoff_factor if backoff_factor is None else float(backoff_factor),
        )

    # -------- alt constructors --------
    @classmethod
    def from_env(cls) -> "UnivapayConfig":
        """Build config strictly from environment (.env considered if loaded)."""
        return cls().validate()

backoff_factor = None class-attribute instance-attribute

base_url = None class-attribute instance-attribute

debug = None class-attribute instance-attribute

jwt = None class-attribute instance-attribute

retries = None class-attribute instance-attribute

secret = None class-attribute instance-attribute

store_id = None class-attribute instance-attribute

timeout = None class-attribute instance-attribute

__init__(jwt=None, secret=None, store_id=None, base_url=None, timeout=None, debug=None, retries=None, backoff_factor=None)

__post_init__()

Source code in univapay/config.py

def __post_init__(self) -> None:
    env = os.environ

    # jwt
    if self.jwt is None or self.jwt == "":
        self.jwt = env.get("UNIVAPAY_JWT", "")
        self._source["jwt"] = "env"
    else:
        self._source["jwt"] = "arg"

    # secret
    if self.secret is None or self.secret == "":
        self.secret = env.get("UNIVAPAY_SECRET", "")
        self._source["secret"] = "env"
    else:
        self._source["secret"] = "arg"

    # store_id
    if self.store_id is None or self.store_id == "":
        self.store_id = env.get("UNIVAPAY_STORE_ID") or None
        self._source["store_id"] = "env"
    else:
        self._source["store_id"] = "arg"

    # base_url
    if self.base_url is None or self.base_url == "":
        self.base_url = _normalize_base_url(env.get("UNIVAPAY_BASE_URL"))
        self._source["base_url"] = "env/default"
    else:
        self.base_url = _normalize_base_url(self.base_url)
        self._source["base_url"] = "arg"

    # timeout
    if self.timeout is None:
        self.timeout = _parse_float(env.get("UNIVAPAY_TIMEOUT"), 30.0)
        self._source["timeout"] = "env/default"
    else:
        try:
            self.timeout = float(self.timeout)
        except Exception:
            self.timeout = 30.0
        self._source["timeout"] = "arg"

    # debug
    if self.debug is None:
        self.debug = _parse_bool(env.get("UNIVAPAY_DEBUG", "1"), True)
        self._source["debug"] = "env/default"
    else:
        self.debug = bool(self.debug)
        self._source["debug"] = "arg"

    # optional retry hints
    if self.retries is None:
        self.retries = _parse_int(env.get("UNIVAPAY_RETRIES"), 0)
        self._source["retries"] = "env/default"
    else:
        try:
            self.retries = int(self.retries)
        except Exception:
            self.retries = 0
        self._source["retries"] = "arg"

    if self.backoff_factor is None:
        self.backoff_factor = _parse_float(env.get("UNIVAPAY_BACKOFF"), 0.5)
        self._source["backoff_factor"] = "env/default"
    else:
        try:
            self.backoff_factor = float(self.backoff_factor)
        except Exception:
            self.backoff_factor = 0.5
        self._source["backoff_factor"] = "arg"

    # Initial debug print (sanitized)
    _dprint(
        bool(self.debug),
        "Loaded config:",
        {
            "jwt": _mask(self.jwt, "jwt"),
            "secret": _mask(self.secret, "secret"),
            "store_id": self.store_id or None,
            "base_url": self.base_url,
            "timeout": self.timeout,
            "debug": self.debug,
            "source": self._source,
        },
    )

copy_with(*, jwt=None, secret=None, store_id=None, base_url=None, timeout=None, debug=None, retries=None, backoff_factor=None)

Create a modified copy (handy in tests).

Source code in univapay/config.py

def copy_with(
    self,
    *,
    jwt: Optional[str] = None,
    secret: Optional[str] = None,
    store_id: Optional[str] = None,
    base_url: Optional[str] = None,
    timeout: Optional[float] = None,
    debug: Optional[bool] = None,
    retries: Optional[int] = None,
    backoff_factor: Optional[float] = None,
) -> "UnivapayConfig":
    """Create a modified copy (handy in tests)."""
    return replace(
        self,
        jwt=self.jwt if jwt is None else jwt,
        secret=self.secret if secret is None else secret,
        store_id=self.store_id if store_id is None else store_id,
        base_url=_normalize_base_url(base_url if base_url is not None else self.base_url),
        timeout=self.timeout if timeout is None else float(timeout),
        debug=self.debug if debug is None else bool(debug),
        retries=self.retries if retries is None else int(retries),
        backoff_factor=self.backoff_factor if backoff_factor is None else float(backoff_factor),
    )

from_env() classmethod

Build config strictly from environment (.env considered if loaded).

Source code in univapay/config.py

@classmethod
def from_env(cls) -> "UnivapayConfig":
    """Build config strictly from environment (.env considered if loaded)."""
    return cls().validate()

masked()

Return a sanitized dict for logging/diagnostics.

Source code in univapay/config.py

def masked(self) -> dict:
    """Return a sanitized dict for logging/diagnostics."""
    return {
        "jwt": _mask(self.jwt, "jwt"),
        "secret": _mask(self.secret, "secret"),
        "store_id": self.store_id,
        "base_url": self.base_url,
        "timeout": self.timeout,
        "debug": self.debug,
        "retries": self.retries,
        "backoff_factor": self.backoff_factor,
    }

require_store_id()

Ensure a store_id is present for endpoints that need it.

Source code in univapay/config.py

def require_store_id(self) -> str:
    """Ensure a store_id is present for endpoints that need it."""
    if not self.store_id:
        raise UnivapayConfigError("UNIVAPAY_STORE_ID is required for this operation.")
    return self.store_id

validate()

Validate presence of credentials for server-side API calls. (Widget-only use-cases can bypass by not instantiating this config.)

Source code in univapay/config.py

def validate(self) -> "UnivapayConfig":
    """
    Validate presence of credentials for server-side API calls.
    (Widget-only use-cases can bypass by not instantiating this config.)
    """
    if not self.jwt or not self.secret:
        _dprint(
            bool(self.debug),
            "Validation failed: jwt/secret missing",
            {"jwt": _mask(self.jwt, "jwt"), "secret": _mask(self.secret, "secret")},
        )
        raise UnivapayConfigError(
            "UNIVAPAY_JWT and UNIVAPAY_SECRET are required for server-side API calls."
        )
    _dprint(bool(self.debug), "Validation OK")
    return self

Lightweight sync client for Univapay REST.

• Adds Authorization header "Bearer {secret}.{jwt}".
• Supports Idempotency-Key on mutating requests.
• Optional simple retries/backoff for transient errors (429/5xx).
• Prints sanitized debug logs (Authorization redacted).

Source code in univapay/client.py

class UnivapayClient:
    """
    Lightweight sync client for Univapay REST.

    - Adds Authorization header "Bearer {secret}.{jwt}".
    - Supports Idempotency-Key on mutating requests.
    - Optional simple retries/backoff for transient errors (429/5xx).
    - Prints sanitized debug logs (Authorization redacted).
    """

    def __init__(
        self,
        config: UnivapayConfig,
        *,
        retries: int = 0,
        backoff_factor: float = 0.5,
    ):
        self.config = config.validate()
        self.retries = max(0, int(retries))
        self.backoff_factor = max(0.0, float(backoff_factor))

        self._client = httpx.Client(
            base_url=self.config.base_url,
            timeout=self.config.timeout,
            headers={
                "User-Agent": f"univapay-python/{SDK_VERSION}",
            },
        )
        dprint(
            "Client init",
            {
                "base_url": self.config.base_url,
                "store_id": self.config.store_id,
                "timeout": self.config.timeout,
                "debug": self.config.debug,
                "retries": self.retries,
                "backoff_factor": self.backoff_factor,
                "sdk_version": SDK_VERSION,
            },
        )

    # ------------ context manager support ------------
    def __enter__(self) -> "UnivapayClient":
        dprint("__enter__()")
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        dprint("__exit__() -> close()")
        self.close()

    # ------------ internal helpers ------------
    def _headers(
        self,
        *,
        idempotency_key: Optional[str] = None,
        extra: Optional[Dict[str, str]] = None,
    ) -> Dict[str, str]:
        h: Dict[str, str] = {
            "Authorization": _auth_header(self.config.secret, self.config.jwt),
            "Accept": "application/json",
            "Content-Type": "application/json",
        }
        if idempotency_key:
            h["Idempotency-Key"] = idempotency_key
        if extra:
            h.update(extra)
        djson("Request headers", scrub_headers(h))
        return h

    def _extract_meta(self, r: httpx.Response) -> Dict[str, Any]:
        meta: Dict[str, Any] = {k: r.headers.get(k) for k in RATE_HEADERS if k in r.headers}
        req_id = _first_header(r.headers, REQUEST_ID_HEADERS)
        if req_id:
            meta["request_id"] = req_id
        retry_after = r.headers.get("Retry-After")
        if retry_after:
            meta["retry_after"] = retry_after
        return meta

    def _handle(self, r: httpx.Response) -> Dict[str, Any]:
        meta = self._extract_meta(r)
        dprint("Response", {"status": r.status_code, "request_id": meta.get("request_id")})
        if meta:
            dprint("Rate limits", {k: v for k, v in meta.items() if k in RATE_HEADERS})

        # Parse body safely
        if r.status_code == 204 or not r.content:
            body: Dict[str, Any] = {}
        else:
            try:
                body = r.json()
                if not isinstance(body, dict):
                    body = {"data": body}
            except Exception:
                body = {"message": r.text}

        djson("Response body", body)

        if 200 <= r.status_code < 300:
            body.setdefault("_meta", {})["rate_limits"] = {k: v for k, v in meta.items() if k in RATE_HEADERS}
            if meta.get("request_id"):
                body["_meta"]["request_id"] = meta["request_id"]
            if meta.get("retry_after"):
                body["_meta"]["retry_after"] = meta["retry_after"]
            return body

        # Error path
        raise UnivapayHTTPError(r.status_code, body, meta.get("request_id"))

    def _path(self, resource: str) -> str:
        p = f"/stores/{self.config.store_id}/{resource}" if self.config.store_id else f"/{resource}"
        dprint("Resolved path", p)
        return p

    def _sleep_for_retry(self, attempt: int, *, retry_after_header: Optional[str]) -> float:
        # Use Retry-After if present (seconds). Fallback to exponential backoff.
        if retry_after_header:
            try:
                # Header may be seconds or HTTP-date; we only support seconds here.
                secs = float(retry_after_header)
                if secs >= 0:
                    return secs
            except Exception:
                pass
        return self.backoff_factor * (2 ** attempt)

    def _send_with_retries(self, method: str, url: str, **kwargs) -> httpx.Response:
        attempt = 0
        while True:
            try:
                dprint("HTTP send", {"method": method.upper(), "url": url})
                r = self._client.request(method, url, **kwargs)
                if r.status_code in TRANSIENT_STATUS and attempt < self.retries:
                    wait = self._sleep_for_retry(attempt, retry_after_header=r.headers.get("Retry-After"))
                    dprint(
                        "Transient response -> retry",
                        {"status": r.status_code, "attempt": attempt + 1, "sleep": wait},
                    )
                    time.sleep(max(0.0, wait))
                    attempt += 1
                    continue
                return r
            except (httpx.ConnectError, httpx.ReadTimeout, httpx.RemoteProtocolError, httpx.HTTPError) as e:
                if attempt < self.retries:
                    wait = self.backoff_factor * (2 ** attempt)
                    dprint(
                        "Network error -> retry",
                        {"error": repr(e), "attempt": attempt + 1, "sleep": wait},
                    )
                    time.sleep(max(0.0, wait))
                    attempt += 1
                    continue
                dprint("Network error -> giving up", {"error": repr(e)})
                raise UnivapayHTTPError(-1, {"message": str(e)}, None) from e

    # ------------ public request helpers ------------
    def get(
        self,
        resource_path: str,
        *,
        polling: bool = False,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        params = dict(params or {})
        if polling:
            params["polling"] = "true"
        dprint("GET", {"path": resource_path, "params": params})
        r = self._send_with_retries(
            "GET",
            resource_path,
            params=params,
            headers=self._headers(extra=extra_headers),
        )
        return self._handle(r)

    def post(
        self,
        resource_path: str,
        *,
        json: Dict[str, Any],
        idempotency_key: Optional[str] = None,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        dprint("POST", {"path": resource_path})
        djson("Request JSON", json)
        r = self._send_with_retries(
            "POST",
            resource_path,
            json=json,
            params=params,
            headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
        )
        return self._handle(r)

    def patch(
        self,
        resource_path: str,
        *,
        json: Dict[str, Any],
        idempotency_key: Optional[str] = None,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        dprint("PATCH", {"path": resource_path})
        djson("Request JSON", json)
        r = self._send_with_retries(
            "PATCH",
            resource_path,
            json=json,
            params=params,
            headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
        )
        return self._handle(r)

    def put(
        self,
        resource_path: str,
        *,
        json: Dict[str, Any],
        idempotency_key: Optional[str] = None,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        dprint("PUT", {"path": resource_path})
        djson("Request JSON", json)
        r = self._send_with_retries(
            "PUT",
            resource_path,
            json=json,
            params=params,
            headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
        )
        return self._handle(r)

    def delete(
        self,
        resource_path: str,
        *,
        json: Optional[Dict[str, Any]] = None,
        idempotency_key: Optional[str] = None,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        dprint("DELETE", {"path": resource_path})
        if json is not None:
            djson("Request JSON", json)
        r = self._send_with_retries(
            "DELETE",
            resource_path,
            json=json,
            params=params,
            headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
        )
        return self._handle(r)

    def head(
        self,
        resource_path: str,
        *,
        params: Dict[str, Any] | None = None,
        extra_headers: Optional[Dict[str, str]] = None,
    ) -> Dict[str, Any]:
        dprint("HEAD", {"path": resource_path, "params": params or {}})
        r = self._send_with_retries(
            "HEAD",
            resource_path,
            params=params or {},
            headers=self._headers(extra=extra_headers),
        )
        # HEAD returns no body; expose meta only
        meta = self._extract_meta(r)
        return {"_meta": {"rate_limits": {k: v for k, v in meta.items() if k in RATE_HEADERS},
                          "request_id": meta.get("request_id"),
                          "retry_after": meta.get("retry_after")}}

    def close(self) -> None:
        dprint("Client close()")
        self._client.close()

backoff_factor = max(0.0, float(backoff_factor)) instance-attribute

config = config.validate() instance-attribute

retries = max(0, int(retries)) instance-attribute

__enter__()

Source code in univapay/client.py

def __enter__(self) -> "UnivapayClient":
    dprint("__enter__()")
    return self

__exit__(exc_type, exc, tb)

Source code in univapay/client.py

def __exit__(self, exc_type, exc, tb) -> None:
    dprint("__exit__() -> close()")
    self.close()

__init__(config, *, retries=0, backoff_factor=0.5)

Source code in univapay/client.py

def __init__(
    self,
    config: UnivapayConfig,
    *,
    retries: int = 0,
    backoff_factor: float = 0.5,
):
    self.config = config.validate()
    self.retries = max(0, int(retries))
    self.backoff_factor = max(0.0, float(backoff_factor))

    self._client = httpx.Client(
        base_url=self.config.base_url,
        timeout=self.config.timeout,
        headers={
            "User-Agent": f"univapay-python/{SDK_VERSION}",
        },
    )
    dprint(
        "Client init",
        {
            "base_url": self.config.base_url,
            "store_id": self.config.store_id,
            "timeout": self.config.timeout,
            "debug": self.config.debug,
            "retries": self.retries,
            "backoff_factor": self.backoff_factor,
            "sdk_version": SDK_VERSION,
        },
    )

close()

Source code in univapay/client.py

def close(self) -> None:
    dprint("Client close()")
    self._client.close()

delete(resource_path, *, json=None, idempotency_key=None, params=None, extra_headers=None)

Source code in univapay/client.py

def delete(
    self,
    resource_path: str,
    *,
    json: Optional[Dict[str, Any]] = None,
    idempotency_key: Optional[str] = None,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    dprint("DELETE", {"path": resource_path})
    if json is not None:
        djson("Request JSON", json)
    r = self._send_with_retries(
        "DELETE",
        resource_path,
        json=json,
        params=params,
        headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
    )
    return self._handle(r)

get(resource_path, *, polling=False, params=None, extra_headers=None)

Source code in univapay/client.py

def get(
    self,
    resource_path: str,
    *,
    polling: bool = False,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    params = dict(params or {})
    if polling:
        params["polling"] = "true"
    dprint("GET", {"path": resource_path, "params": params})
    r = self._send_with_retries(
        "GET",
        resource_path,
        params=params,
        headers=self._headers(extra=extra_headers),
    )
    return self._handle(r)

head(resource_path, *, params=None, extra_headers=None)

Source code in univapay/client.py

def head(
    self,
    resource_path: str,
    *,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    dprint("HEAD", {"path": resource_path, "params": params or {}})
    r = self._send_with_retries(
        "HEAD",
        resource_path,
        params=params or {},
        headers=self._headers(extra=extra_headers),
    )
    # HEAD returns no body; expose meta only
    meta = self._extract_meta(r)
    return {"_meta": {"rate_limits": {k: v for k, v in meta.items() if k in RATE_HEADERS},
                      "request_id": meta.get("request_id"),
                      "retry_after": meta.get("retry_after")}}

patch(resource_path, *, json, idempotency_key=None, params=None, extra_headers=None)

Source code in univapay/client.py

def patch(
    self,
    resource_path: str,
    *,
    json: Dict[str, Any],
    idempotency_key: Optional[str] = None,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    dprint("PATCH", {"path": resource_path})
    djson("Request JSON", json)
    r = self._send_with_retries(
        "PATCH",
        resource_path,
        json=json,
        params=params,
        headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
    )
    return self._handle(r)

post(resource_path, *, json, idempotency_key=None, params=None, extra_headers=None)

Source code in univapay/client.py

def post(
    self,
    resource_path: str,
    *,
    json: Dict[str, Any],
    idempotency_key: Optional[str] = None,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    dprint("POST", {"path": resource_path})
    djson("Request JSON", json)
    r = self._send_with_retries(
        "POST",
        resource_path,
        json=json,
        params=params,
        headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
    )
    return self._handle(r)

put(resource_path, *, json, idempotency_key=None, params=None, extra_headers=None)

Source code in univapay/client.py

def put(
    self,
    resource_path: str,
    *,
    json: Dict[str, Any],
    idempotency_key: Optional[str] = None,
    params: Dict[str, Any] | None = None,
    extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    dprint("PUT", {"path": resource_path})
    djson("Request JSON", json)
    r = self._send_with_retries(
        "PUT",
        resource_path,
        json=json,
        params=params,
        headers=self._headers(idempotency_key=idempotency_key, extra=extra_headers),
    )
    return self._handle(r)

Resources

One-time & recurring charges API.

Notes

• For one-time: pass a transaction token produced by a one-time widget.
• For recurring: pass a transaction token produced by a 'recurring' widget.
• Use idempotency_key on POSTs to avoid duplicate charges on retries.
• Use get(..., polling=True) or wait_until_terminal(...) to block until a terminal status.

Source code in univapay/resources/charges.py

class ChargesAPI:
    """
    One-time & recurring charges API.

    Notes:
      - For one-time: pass a transaction token produced by a one-time widget.
      - For recurring: pass a transaction token produced by a 'recurring' widget.
      - Use `idempotency_key` on POSTs to avoid duplicate charges on retries.
      - Use `get(..., polling=True)` or `wait_until_terminal(...)` to block until a terminal status.
    """

    def __init__(self, client: UnivapayClient):
        self.client = client
        dprint("charges.__init__()", {"base_path": _charges_base(self.client)})

    # ----------------------- create: one-time -----------------------

    def create_one_time(
        self,
        *,
        token_id: str,
        amount: int,
        currency: str = "jpy",
        capture: bool = True,
        metadata: Optional[Dict[str, Any]] = None,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Charge:
        """
        Create a one-time charge using a one-time transaction token.
        """
        _validate_token(token_id)
        _validate_amount(amount)
        currency = _validate_currency(currency)

        dprint("charges.create_one_time()", {
            "token_id": token_id,
            "amount": amount,
            "currency": currency,
            "capture": capture,
            "idempotency_key_present": bool(idempotency_key),
        })

        body = ChargeCreate(
            transaction_token_id=token_id,
            amount=amount,
            currency=currency,
            capture=capture,
        ).model_dump(by_alias=True)

        if metadata:
            body["metadata"] = metadata
        body.update(extra or {})

        djson("charges.create_one_time body", body)
        resp = self.client.post(_charges_base(self.client), json=body, idempotency_key=idempotency_key)
        return Charge.model_validate(resp)

    # ----------------------- create: recurring -----------------------

    def create_recurring(
        self,
        *,
        token_id: str,
        amount: int,
        currency: str = "jpy",
        capture: bool = True,
        metadata: Optional[Dict[str, Any]] = None,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Charge:
        """
        Create a charge using a **recurring** transaction token.
        Endpoint is the same as one-time; the server enforces token type.
        """
        _validate_token(token_id)
        _validate_amount(amount)
        currency = _validate_currency(currency)

        dprint("charges.create_recurring()", {
            "token_id": token_id,
            "amount": amount,
            "currency": currency,
            "capture": capture,
            "idempotency_key_present": bool(idempotency_key),
        })

        # Delegate to the same creation logic to keep behavior identical.
        return self.create_one_time(
            token_id=token_id,
            amount=amount,
            currency=currency,
            capture=capture,
            metadata=metadata,
            idempotency_key=idempotency_key,
            **(extra or {}),
        )

    # ----------------------- read -----------------------

    def get(self, charge_id: str, *, polling: bool = False) -> Charge:
        """
        Retrieve a charge. If polling=True, server blocks until a terminal state when supported.
        """
        _validate_id("charge_id", charge_id)
        dprint("charges.get()", {"charge_id": charge_id, "polling": polling})
        resp = self.client.get(f"{_charges_base(self.client)}/{charge_id}", polling=polling)
        return Charge.model_validate(resp)

    # ----------------------- waiter -----------------------

    def wait_until_terminal(
        self,
        charge_id: str,
        *,
        server_polling: bool = True,
        timeout_s: int = 90,
        interval_s: float = 2.0,
    ) -> Charge:
        """
        Block until the charge reaches a terminal state.

        If server_polling=True, perform a single GET with polling=true.
        Otherwise, poll client-side every interval_s until timeout_s is reached.
        """
        _validate_id("charge_id", charge_id)
        if timeout_s <= 0 or interval_s <= 0:
            raise ValueError("timeout_s and interval_s must be positive.")

        dprint("charges.wait_until_terminal()", {
            "charge_id": charge_id,
            "server_polling": server_polling,
            "timeout_s": timeout_s,
            "interval_s": interval_s,
        })

        if server_polling:
            ch = self.get(charge_id, polling=True)
            dprint("charges.wait_until_terminal: server-polling returned", {"status": ch.status})
            return ch

        deadline = time.time() + timeout_s
        while True:
            ch = self.get(charge_id)
            status_norm = (ch.status or "").lower()
            if status_norm in _TERMINAL_STATES:
                dprint("charges.wait_until_terminal -> terminal", {"status": ch.status})
                return ch
            if time.time() >= deadline:
                dprint("charges.wait_until_terminal -> timeout", {"last_status": ch.status})
                return ch
            time.sleep(interval_s)

    # ----------------------- actions -----------------------

    def refund(
        self,
        charge_id: str,
        *,
        amount: Optional[int] = None,
        idempotency_key: Optional[str] = None,
    ) -> Refund:
        """
        Create a refund for a charge. If `amount` is None, a full refund may be performed (API-dependent).
        """
        _validate_id("charge_id", charge_id)
        if amount is not None:
            _validate_amount(amount)

        dprint("charges.refund()", {
            "charge_id": charge_id,
            "amount": amount,
            "idempotency_key_present": bool(idempotency_key),
        })

        path = f"{_charges_base(self.client)}/{charge_id}/refunds"
        body: Dict[str, Any] = {"amount": amount} if amount is not None else {}
        djson("charges.refund body", body)
        resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
        return Refund.model_validate(resp)

    def capture(self, charge_id: str, *, idempotency_key: Optional[str] = None, **extra: Any) -> Charge:
        """
        Capture a previously authorized charge (if your account flow supports auth/capture).
        """
        _validate_id("charge_id", charge_id)
        dprint("charges.capture()", {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)})
        path = f"{_charges_base(self.client)}/{charge_id}/capture"
        djson("charges.capture body", {**extra} if extra else {})
        resp = self.client.post(path, json={**(extra or {})}, idempotency_key=idempotency_key)
        return Charge.model_validate(resp)

    def cancel(self, charge_id: str, *, idempotency_key: Optional[str] = None, **extra: Any) -> Charge:
        """
        Cancel (void) a charge. Route name may vary by capture flow; adjust if needed.
        """
        _validate_id("charge_id", charge_id)
        dprint("charges.cancel()", {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)})
        path = f"{_charges_base(self.client)}/{charge_id}/cancel"
        djson("charges.cancel body", {**extra} if extra else {})
        resp = self.client.post(path, json={**(extra or {})}, idempotency_key=idempotency_key)
        return Charge.model_validate(resp)

client = client instance-attribute

__init__(client)

Source code in univapay/resources/charges.py

def __init__(self, client: UnivapayClient):
    self.client = client
    dprint("charges.__init__()", {"base_path": _charges_base(self.client)})

cancel(charge_id, *, idempotency_key=None, **extra)

Cancel (void) a charge. Route name may vary by capture flow; adjust if needed.

Source code in univapay/resources/charges.py

def cancel(self, charge_id: str, *, idempotency_key: Optional[str] = None, **extra: Any) -> Charge:
    """
    Cancel (void) a charge. Route name may vary by capture flow; adjust if needed.
    """
    _validate_id("charge_id", charge_id)
    dprint("charges.cancel()", {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)})
    path = f"{_charges_base(self.client)}/{charge_id}/cancel"
    djson("charges.cancel body", {**extra} if extra else {})
    resp = self.client.post(path, json={**(extra or {})}, idempotency_key=idempotency_key)
    return Charge.model_validate(resp)

capture(charge_id, *, idempotency_key=None, **extra)

Capture a previously authorized charge (if your account flow supports auth/capture).

Source code in univapay/resources/charges.py

def capture(self, charge_id: str, *, idempotency_key: Optional[str] = None, **extra: Any) -> Charge:
    """
    Capture a previously authorized charge (if your account flow supports auth/capture).
    """
    _validate_id("charge_id", charge_id)
    dprint("charges.capture()", {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)})
    path = f"{_charges_base(self.client)}/{charge_id}/capture"
    djson("charges.capture body", {**extra} if extra else {})
    resp = self.client.post(path, json={**(extra or {})}, idempotency_key=idempotency_key)
    return Charge.model_validate(resp)

create_one_time(*, token_id, amount, currency='jpy', capture=True, metadata=None, idempotency_key=None, **extra)

Create a one-time charge using a one-time transaction token.

Source code in univapay/resources/charges.py

def create_one_time(
    self,
    *,
    token_id: str,
    amount: int,
    currency: str = "jpy",
    capture: bool = True,
    metadata: Optional[Dict[str, Any]] = None,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Charge:
    """
    Create a one-time charge using a one-time transaction token.
    """
    _validate_token(token_id)
    _validate_amount(amount)
    currency = _validate_currency(currency)

    dprint("charges.create_one_time()", {
        "token_id": token_id,
        "amount": amount,
        "currency": currency,
        "capture": capture,
        "idempotency_key_present": bool(idempotency_key),
    })

    body = ChargeCreate(
        transaction_token_id=token_id,
        amount=amount,
        currency=currency,
        capture=capture,
    ).model_dump(by_alias=True)

    if metadata:
        body["metadata"] = metadata
    body.update(extra or {})

    djson("charges.create_one_time body", body)
    resp = self.client.post(_charges_base(self.client), json=body, idempotency_key=idempotency_key)
    return Charge.model_validate(resp)

create_recurring(*, token_id, amount, currency='jpy', capture=True, metadata=None, idempotency_key=None, **extra)

Create a charge using a recurring transaction token. Endpoint is the same as one-time; the server enforces token type.

Source code in univapay/resources/charges.py

def create_recurring(
    self,
    *,
    token_id: str,
    amount: int,
    currency: str = "jpy",
    capture: bool = True,
    metadata: Optional[Dict[str, Any]] = None,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Charge:
    """
    Create a charge using a **recurring** transaction token.
    Endpoint is the same as one-time; the server enforces token type.
    """
    _validate_token(token_id)
    _validate_amount(amount)
    currency = _validate_currency(currency)

    dprint("charges.create_recurring()", {
        "token_id": token_id,
        "amount": amount,
        "currency": currency,
        "capture": capture,
        "idempotency_key_present": bool(idempotency_key),
    })

    # Delegate to the same creation logic to keep behavior identical.
    return self.create_one_time(
        token_id=token_id,
        amount=amount,
        currency=currency,
        capture=capture,
        metadata=metadata,
        idempotency_key=idempotency_key,
        **(extra or {}),
    )

get(charge_id, *, polling=False)

Retrieve a charge. If polling=True, server blocks until a terminal state when supported.

Source code in univapay/resources/charges.py

def get(self, charge_id: str, *, polling: bool = False) -> Charge:
    """
    Retrieve a charge. If polling=True, server blocks until a terminal state when supported.
    """
    _validate_id("charge_id", charge_id)
    dprint("charges.get()", {"charge_id": charge_id, "polling": polling})
    resp = self.client.get(f"{_charges_base(self.client)}/{charge_id}", polling=polling)
    return Charge.model_validate(resp)

refund(charge_id, *, amount=None, idempotency_key=None)

Create a refund for a charge. If amount is None, a full refund may be performed (API-dependent).

Source code in univapay/resources/charges.py

def refund(
    self,
    charge_id: str,
    *,
    amount: Optional[int] = None,
    idempotency_key: Optional[str] = None,
) -> Refund:
    """
    Create a refund for a charge. If `amount` is None, a full refund may be performed (API-dependent).
    """
    _validate_id("charge_id", charge_id)
    if amount is not None:
        _validate_amount(amount)

    dprint("charges.refund()", {
        "charge_id": charge_id,
        "amount": amount,
        "idempotency_key_present": bool(idempotency_key),
    })

    path = f"{_charges_base(self.client)}/{charge_id}/refunds"
    body: Dict[str, Any] = {"amount": amount} if amount is not None else {}
    djson("charges.refund body", body)
    resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
    return Refund.model_validate(resp)

wait_until_terminal(charge_id, *, server_polling=True, timeout_s=90, interval_s=2.0)

Block until the charge reaches a terminal state.

If server_polling=True, perform a single GET with polling=true. Otherwise, poll client-side every interval_s until timeout_s is reached.

Source code in univapay/resources/charges.py

def wait_until_terminal(
    self,
    charge_id: str,
    *,
    server_polling: bool = True,
    timeout_s: int = 90,
    interval_s: float = 2.0,
) -> Charge:
    """
    Block until the charge reaches a terminal state.

    If server_polling=True, perform a single GET with polling=true.
    Otherwise, poll client-side every interval_s until timeout_s is reached.
    """
    _validate_id("charge_id", charge_id)
    if timeout_s <= 0 or interval_s <= 0:
        raise ValueError("timeout_s and interval_s must be positive.")

    dprint("charges.wait_until_terminal()", {
        "charge_id": charge_id,
        "server_polling": server_polling,
        "timeout_s": timeout_s,
        "interval_s": interval_s,
    })

    if server_polling:
        ch = self.get(charge_id, polling=True)
        dprint("charges.wait_until_terminal: server-polling returned", {"status": ch.status})
        return ch

    deadline = time.time() + timeout_s
    while True:
        ch = self.get(charge_id)
        status_norm = (ch.status or "").lower()
        if status_norm in _TERMINAL_STATES:
            dprint("charges.wait_until_terminal -> terminal", {"status": ch.status})
            return ch
        if time.time() >= deadline:
            dprint("charges.wait_until_terminal -> timeout", {"last_status": ch.status})
            return ch
        time.sleep(interval_s)

Subscriptions API.

• Create a subscription from a transaction token produced by a subscription widget.
• Use idempotency_key on POSTs to avoid dupes on retry.
• Use get(..., polling=True) or wait_until_terminal(...) to block until a terminal-ish state.
• Cancel with cancel(subscription_id, ...).

Source code in univapay/resources/subscriptions.py

class SubscriptionsAPI:
    """
    Subscriptions API.

    - Create a subscription from a transaction token produced by a **subscription** widget.
    - Use `idempotency_key` on POSTs to avoid dupes on retry.
    - Use `get(..., polling=True)` or `wait_until_terminal(...)` to block until a terminal-ish state.
    - Cancel with `cancel(subscription_id, ...)`.
    """

    def __init__(self, client: UnivapayClient):
        self.client = client
        dprint("subscriptions.__init__()", {"base_path": _subs_base(self.client)})

    # ------------------------ create ------------------------

    def create(
        self,
        *,
        token_id: str,
        amount: int,
        period: str,
        currency: str = "jpy",
        metadata: Optional[Dict[str, Any]] = None,
        start_on: Optional[str] = None,  # ISO date (YYYY-MM-DD)
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Subscription:
        """Create a subscription using a subscription-capable transaction token."""
        _validate_token(token_id)
        _validate_amount(amount)
        currency = _validate_currency(currency)
        period = _validate_period(period)

        dprint(
            "subscriptions.create()",
            {
                "token_id": token_id,
                "amount": amount,
                "period": period,
                "currency": currency,
                "start_on": start_on,
                "idempotency_key_present": bool(idempotency_key),
            },
        )

        body = SubscriptionCreate(
            transaction_token_id=token_id,
            amount=amount,
            currency=currency,
            period=period,
            start_on=start_on,
        ).model_dump(by_alias=True, exclude_none=True)

        if metadata:
            body["metadata"] = metadata
        body.update(extra or {})

        djson("subscriptions.create body", body)
        resp = self.client.post(_subs_base(self.client), json=body, idempotency_key=idempotency_key)
        return Subscription.model_validate(resp)

    # ------------------------ read ------------------------

    def get(self, subscription_id: str, *, polling: bool = False) -> Subscription:
        """Retrieve a subscription. If polling=True, server may block until steady/terminal state."""
        _validate_id("subscription_id", subscription_id)
        dprint("subscriptions.get()", {"subscription_id": subscription_id, "polling": polling})
        resp = self.client.get(f"{_subs_base(self.client)}/{subscription_id}", polling=polling)
        return Subscription.model_validate(resp)

    # ------------------------ waiter ------------------------

    def wait_until_terminal(
        self,
        subscription_id: str,
        *,
        server_polling: bool = False,  # default False to avoid long blocks by default
        timeout_s: int = 60,
        interval_s: float = 2.0,
    ) -> Subscription:
        """
        Return once the subscription is in a terminal-ish state.

        Flow:
          1) Quick GET without polling; if already terminal-ish (e.g., 'current'), return immediately.
          2) If server_polling=True, do a single GET with polling=true (server may block).
          3) Else, client-side poll until terminal or timeout.
        """
        _validate_id("subscription_id", subscription_id)
        if timeout_s <= 0 or interval_s <= 0:
            raise ValueError("timeout_s and interval_s must be positive.")

        dprint(
            "subscriptions.wait_until_terminal()",
            {
                "subscription_id": subscription_id,
                "server_polling": server_polling,
                "timeout_s": timeout_s,
                "interval_s": interval_s,
            },
        )

        # Step 1: quick check
        sub = self.get(subscription_id, polling=False)
        if _is_terminal(sub.status):
            dprint("subscriptions.wait_until_terminal: already terminal-ish", {"status": sub.status})
            return sub

        # Step 2: server polling if requested
        if server_polling:
            sub_p = self.get(subscription_id, polling=True)
            dprint("subscriptions.wait_until_terminal: server-polling returned", {"status": sub_p.status})
            return sub_p

        # Step 3: client-side loop
        deadline = time.time() + timeout_s
        last = sub
        while time.time() < deadline:
            time.sleep(interval_s)
            last = self.get(subscription_id, polling=False)
            if _is_terminal(last.status):
                dprint("subscriptions.wait_until_terminal: reached terminal-ish", {"status": last.status})
                return last

        dprint(
            "subscriptions.wait_until_terminal timeout",
            {"subscription_id": subscription_id, "last_status": getattr(last, "status", None)},
        )
        return last

    # ------------------------ actions ------------------------

    def cancel(
        self,
        subscription_id: str,
        *,
        idempotency_key: Optional[str] = None,
        termination_mode: Optional[str] = None,
        **extra: Any,
    ) -> Subscription:
        """
        Cancel a subscription and return the updated Subscription resource.

        Primary attempt: POST /subscriptions/{id}/cancel.
        Fallback: PATCH /subscriptions/{id} with {'termination_mode': 'immediate'|'on_next_payment'}
        """
        _validate_id("subscription_id", subscription_id)
        dprint(
            "subscriptions.cancel()",
            {"subscription_id": subscription_id, "idempotency_key_present": bool(idempotency_key)},
        )
        base = _subs_base(self.client)
        path = f"{base}/{subscription_id}/cancel"
        body = {**(extra or {})}
        if termination_mode:
            body.setdefault("termination_mode", termination_mode)
        djson("subscriptions.cancel body", body if body else {})
        try:
            resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
            return Subscription.model_validate(resp)
        except Exception as e:
            # Fallback for accounts without /cancel endpoint
            try:
                from ..errors import UnivapayHTTPError  # local import to avoid cycle
                not_found = isinstance(e, UnivapayHTTPError) and e.status in (404, 405)
            except Exception:
                not_found = False
            if not_found:
                dprint("subscriptions.cancel fallback -> PATCH", {"subscription_id": subscription_id})
                patch_body: Dict[str, Any] = {**(extra or {})}
                patch_body.setdefault("termination_mode", termination_mode or "immediate")
                djson("subscriptions.cancel PATCH body", patch_body)
                resp2 = self.client.patch(f"{base}/{subscription_id}", json=patch_body, idempotency_key=idempotency_key)
                return Subscription.model_validate(resp2)
            raise

client = client instance-attribute

__init__(client)

Source code in univapay/resources/subscriptions.py

def __init__(self, client: UnivapayClient):
    self.client = client
    dprint("subscriptions.__init__()", {"base_path": _subs_base(self.client)})

cancel(subscription_id, *, idempotency_key=None, termination_mode=None, **extra)

Cancel a subscription and return the updated Subscription resource.

Primary attempt: POST /subscriptions/{id}/cancel. Fallback: PATCH /subscriptions/{id} with {'termination_mode': 'immediate'|'on_next_payment'}

Source code in univapay/resources/subscriptions.py

def cancel(
    self,
    subscription_id: str,
    *,
    idempotency_key: Optional[str] = None,
    termination_mode: Optional[str] = None,
    **extra: Any,
) -> Subscription:
    """
    Cancel a subscription and return the updated Subscription resource.

    Primary attempt: POST /subscriptions/{id}/cancel.
    Fallback: PATCH /subscriptions/{id} with {'termination_mode': 'immediate'|'on_next_payment'}
    """
    _validate_id("subscription_id", subscription_id)
    dprint(
        "subscriptions.cancel()",
        {"subscription_id": subscription_id, "idempotency_key_present": bool(idempotency_key)},
    )
    base = _subs_base(self.client)
    path = f"{base}/{subscription_id}/cancel"
    body = {**(extra or {})}
    if termination_mode:
        body.setdefault("termination_mode", termination_mode)
    djson("subscriptions.cancel body", body if body else {})
    try:
        resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
        return Subscription.model_validate(resp)
    except Exception as e:
        # Fallback for accounts without /cancel endpoint
        try:
            from ..errors import UnivapayHTTPError  # local import to avoid cycle
            not_found = isinstance(e, UnivapayHTTPError) and e.status in (404, 405)
        except Exception:
            not_found = False
        if not_found:
            dprint("subscriptions.cancel fallback -> PATCH", {"subscription_id": subscription_id})
            patch_body: Dict[str, Any] = {**(extra or {})}
            patch_body.setdefault("termination_mode", termination_mode or "immediate")
            djson("subscriptions.cancel PATCH body", patch_body)
            resp2 = self.client.patch(f"{base}/{subscription_id}", json=patch_body, idempotency_key=idempotency_key)
            return Subscription.model_validate(resp2)
        raise

create(*, token_id, amount, period, currency='jpy', metadata=None, start_on=None, idempotency_key=None, **extra)

Create a subscription using a subscription-capable transaction token.

Source code in univapay/resources/subscriptions.py

def create(
    self,
    *,
    token_id: str,
    amount: int,
    period: str,
    currency: str = "jpy",
    metadata: Optional[Dict[str, Any]] = None,
    start_on: Optional[str] = None,  # ISO date (YYYY-MM-DD)
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Subscription:
    """Create a subscription using a subscription-capable transaction token."""
    _validate_token(token_id)
    _validate_amount(amount)
    currency = _validate_currency(currency)
    period = _validate_period(period)

    dprint(
        "subscriptions.create()",
        {
            "token_id": token_id,
            "amount": amount,
            "period": period,
            "currency": currency,
            "start_on": start_on,
            "idempotency_key_present": bool(idempotency_key),
        },
    )

    body = SubscriptionCreate(
        transaction_token_id=token_id,
        amount=amount,
        currency=currency,
        period=period,
        start_on=start_on,
    ).model_dump(by_alias=True, exclude_none=True)

    if metadata:
        body["metadata"] = metadata
    body.update(extra or {})

    djson("subscriptions.create body", body)
    resp = self.client.post(_subs_base(self.client), json=body, idempotency_key=idempotency_key)
    return Subscription.model_validate(resp)

get(subscription_id, *, polling=False)

Retrieve a subscription. If polling=True, server may block until steady/terminal state.

Source code in univapay/resources/subscriptions.py

def get(self, subscription_id: str, *, polling: bool = False) -> Subscription:
    """Retrieve a subscription. If polling=True, server may block until steady/terminal state."""
    _validate_id("subscription_id", subscription_id)
    dprint("subscriptions.get()", {"subscription_id": subscription_id, "polling": polling})
    resp = self.client.get(f"{_subs_base(self.client)}/{subscription_id}", polling=polling)
    return Subscription.model_validate(resp)

wait_until_terminal(subscription_id, *, server_polling=False, timeout_s=60, interval_s=2.0)

Return once the subscription is in a terminal-ish state.

Flow

1) Quick GET without polling; if already terminal-ish (e.g., 'current'), return immediately. 2) If server_polling=True, do a single GET with polling=true (server may block). 3) Else, client-side poll until terminal or timeout.

Source code in univapay/resources/subscriptions.py

def wait_until_terminal(
    self,
    subscription_id: str,
    *,
    server_polling: bool = False,  # default False to avoid long blocks by default
    timeout_s: int = 60,
    interval_s: float = 2.0,
) -> Subscription:
    """
    Return once the subscription is in a terminal-ish state.

    Flow:
      1) Quick GET without polling; if already terminal-ish (e.g., 'current'), return immediately.
      2) If server_polling=True, do a single GET with polling=true (server may block).
      3) Else, client-side poll until terminal or timeout.
    """
    _validate_id("subscription_id", subscription_id)
    if timeout_s <= 0 or interval_s <= 0:
        raise ValueError("timeout_s and interval_s must be positive.")

    dprint(
        "subscriptions.wait_until_terminal()",
        {
            "subscription_id": subscription_id,
            "server_polling": server_polling,
            "timeout_s": timeout_s,
            "interval_s": interval_s,
        },
    )

    # Step 1: quick check
    sub = self.get(subscription_id, polling=False)
    if _is_terminal(sub.status):
        dprint("subscriptions.wait_until_terminal: already terminal-ish", {"status": sub.status})
        return sub

    # Step 2: server polling if requested
    if server_polling:
        sub_p = self.get(subscription_id, polling=True)
        dprint("subscriptions.wait_until_terminal: server-polling returned", {"status": sub_p.status})
        return sub_p

    # Step 3: client-side loop
    deadline = time.time() + timeout_s
    last = sub
    while time.time() < deadline:
        time.sleep(interval_s)
        last = self.get(subscription_id, polling=False)
        if _is_terminal(last.status):
            dprint("subscriptions.wait_until_terminal: reached terminal-ish", {"status": last.status})
            return last

    dprint(
        "subscriptions.wait_until_terminal timeout",
        {"subscription_id": subscription_id, "last_status": getattr(last, "status", None)},
    )
    return last

Refunds API (per-charge).

Common flows

• Create a refund for a given charge (full refund if amount omitted and your account allows it).
• Get a specific refund (optionally with server-side polling).
• List refunds for a charge (basic pagination passthrough).
• Wait until a refund reaches a terminal status.

Source code in univapay/resources/refunds.py

class RefundsAPI:
    """
    Refunds API (per-charge).

    Common flows:
      - Create a refund for a given charge (full refund if `amount` omitted and your account allows it).
      - Get a specific refund (optionally with server-side polling).
      - List refunds for a charge (basic pagination passthrough).
      - Wait until a refund reaches a terminal status.
    """

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

    # ---- Create ----
    def create(
        self,
        charge_id: str,
        *,
        amount: Optional[int] = None,
        reason: Optional[str] = None,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Refund:
        """
        Create a refund.

        Parameters
        ----------
        charge_id : str
            The charge to refund.
        amount : Optional[int]
            Amount in minor units. If None, a full refund may be performed (API/account dependent).
        reason : Optional[str]
            Optional reason string for audit trails.
        idempotency_key : Optional[str]
            Recommended: pass a stable id for safe retries.

        Returns
        -------
        Refund
        """
        _validate_id("charge_id", charge_id)
        if amount is not None:
            _validate_amount(amount)

        payload: Dict[str, Any] = {}
        if amount is not None:
            payload["amount"] = amount
        if reason:
            payload["reason"] = reason
        payload.update(extra)

        dprint(
            "refunds.create()",
            {
                "charge_id": charge_id,
                "amount": amount,
                "reason": reason,
                "idempotency_key_present": bool(idempotency_key),
            },
        )
        djson("refunds.create body", payload)

        resp = self.client.post(_base(self.client, charge_id), json=payload, idempotency_key=idempotency_key)
        return Refund.model_validate(resp)

    def create_full_refund(
        self,
        charge_id: str,
        *,
        reason: Optional[str] = None,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Refund:
        """
        Convenience: request a full refund (omit `amount`).
        """
        dprint("refunds.create_full_refund()", {"charge_id": charge_id})
        return self.create(
            charge_id,
            amount=None,
            reason=reason,
            idempotency_key=idempotency_key,
            **extra,
        )

    def create_partial_refund(
        self,
        charge_id: str,
        *,
        amount: int,
        reason: Optional[str] = None,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Refund:
        """
        Convenience: request a partial refund (requires `amount`).
        """
        dprint("refunds.create_partial_refund()", {"charge_id": charge_id, "amount": amount})
        _validate_amount(amount)
        return self.create(
            charge_id,
            amount=amount,
            reason=reason,
            idempotency_key=idempotency_key,
            **extra,
        )

    # ---- Read ----
    def get(self, charge_id: str, refund_id: str, *, polling: bool = False) -> Refund:
        """
        Fetch a specific refund. If `polling=True`, server may block until a terminal state (when supported).
        """
        _validate_id("charge_id", charge_id)
        _validate_id("refund_id", refund_id)

        path = f"{_base(self.client, charge_id)}/{refund_id}"
        dprint("refunds.get()", {"charge_id": charge_id, "refund_id": refund_id, "polling": polling})
        resp = self.client.get(path, polling=polling)
        return Refund.model_validate(resp)

    # ---- List ----
    def list(
        self,
        charge_id: str,
        *,
        limit: Optional[int] = None,
        cursor: Optional[str] = None,
        extra_params: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """
        List refunds for a charge (passthrough dict to preserve API fields/pagination).
        """
        _validate_id("charge_id", charge_id)

        params: Dict[str, Any] = {}
        if limit is not None:
            if not isinstance(limit, int) or limit <= 0:
                raise ValueError("limit must be a positive integer.")
            params["limit"] = limit
        if cursor:
            params["cursor"] = cursor
        if extra_params:
            params.update(extra_params)

        dprint("refunds.list()", {"charge_id": charge_id, "params": params})
        resp = self.client.get(_base(self.client, charge_id), params=params)
        djson("refunds.list response", resp)
        return resp

    # ---- Wait until terminal ----
    def wait_until_terminal(
        self,
        charge_id: str,
        refund_id: str,
        *,
        server_polling: bool = True,
        timeout_s: int = 60,
        interval_s: float = 2.0,
    ) -> Refund:
        """
        Block until the refund reaches a terminal-ish status.

        If server_polling=True, perform a single GET with polling=true.
        Otherwise, poll client-side every `interval_s` until `timeout_s` is reached.
        """
        _validate_id("charge_id", charge_id)
        _validate_id("refund_id", refund_id)
        if timeout_s <= 0 or interval_s <= 0:
            raise ValueError("timeout_s and interval_s must be positive.")

        dprint(
            "refunds.wait_until_terminal()",
            {
                "charge_id": charge_id,
                "refund_id": refund_id,
                "server_polling": server_polling,
                "timeout_s": timeout_s,
                "interval_s": interval_s,
            },
        )

        if server_polling:
            # Single blocking call if the API supports server-side polling
            return self.get(charge_id, refund_id, polling=True)

        # Client-side polling loop
        deadline = time.time() + timeout_s
        last = self.get(charge_id, refund_id, polling=False)
        if _is_terminal(last.status):
            dprint("refunds.wait_until_terminal: already terminal-ish", {"status": last.status})
            return last

        while time.time() < deadline:
            time.sleep(interval_s)
            last = self.get(charge_id, refund_id, polling=False)
            if _is_terminal(last.status):
                dprint("refunds.wait_until_terminal -> terminal-ish", {"status": last.status})
                return last

        dprint("refunds.wait_until_terminal -> timeout", {"last_status": getattr(last, "status", None)})
        return last

client = client instance-attribute

__init__(client)

Source code in univapay/resources/refunds.py

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

create(charge_id, *, amount=None, reason=None, idempotency_key=None, **extra)

Create a refund.

Parameters

charge_id : str The charge to refund. amount : Optional[int] Amount in minor units. If None, a full refund may be performed (API/account dependent). reason : Optional[str] Optional reason string for audit trails. idempotency_key : Optional[str] Recommended: pass a stable id for safe retries.

Returns

Refund

Source code in univapay/resources/refunds.py

def create(
    self,
    charge_id: str,
    *,
    amount: Optional[int] = None,
    reason: Optional[str] = None,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Refund:
    """
    Create a refund.

    Parameters
    ----------
    charge_id : str
        The charge to refund.
    amount : Optional[int]
        Amount in minor units. If None, a full refund may be performed (API/account dependent).
    reason : Optional[str]
        Optional reason string for audit trails.
    idempotency_key : Optional[str]
        Recommended: pass a stable id for safe retries.

    Returns
    -------
    Refund
    """
    _validate_id("charge_id", charge_id)
    if amount is not None:
        _validate_amount(amount)

    payload: Dict[str, Any] = {}
    if amount is not None:
        payload["amount"] = amount
    if reason:
        payload["reason"] = reason
    payload.update(extra)

    dprint(
        "refunds.create()",
        {
            "charge_id": charge_id,
            "amount": amount,
            "reason": reason,
            "idempotency_key_present": bool(idempotency_key),
        },
    )
    djson("refunds.create body", payload)

    resp = self.client.post(_base(self.client, charge_id), json=payload, idempotency_key=idempotency_key)
    return Refund.model_validate(resp)

create_full_refund(charge_id, *, reason=None, idempotency_key=None, **extra)

Convenience: request a full refund (omit amount).

Source code in univapay/resources/refunds.py

def create_full_refund(
    self,
    charge_id: str,
    *,
    reason: Optional[str] = None,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Refund:
    """
    Convenience: request a full refund (omit `amount`).
    """
    dprint("refunds.create_full_refund()", {"charge_id": charge_id})
    return self.create(
        charge_id,
        amount=None,
        reason=reason,
        idempotency_key=idempotency_key,
        **extra,
    )

create_partial_refund(charge_id, *, amount, reason=None, idempotency_key=None, **extra)

Convenience: request a partial refund (requires amount).

Source code in univapay/resources/refunds.py

def create_partial_refund(
    self,
    charge_id: str,
    *,
    amount: int,
    reason: Optional[str] = None,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Refund:
    """
    Convenience: request a partial refund (requires `amount`).
    """
    dprint("refunds.create_partial_refund()", {"charge_id": charge_id, "amount": amount})
    _validate_amount(amount)
    return self.create(
        charge_id,
        amount=amount,
        reason=reason,
        idempotency_key=idempotency_key,
        **extra,
    )

get(charge_id, refund_id, *, polling=False)

Fetch a specific refund. If polling=True, server may block until a terminal state (when supported).

Source code in univapay/resources/refunds.py

def get(self, charge_id: str, refund_id: str, *, polling: bool = False) -> Refund:
    """
    Fetch a specific refund. If `polling=True`, server may block until a terminal state (when supported).
    """
    _validate_id("charge_id", charge_id)
    _validate_id("refund_id", refund_id)

    path = f"{_base(self.client, charge_id)}/{refund_id}"
    dprint("refunds.get()", {"charge_id": charge_id, "refund_id": refund_id, "polling": polling})
    resp = self.client.get(path, polling=polling)
    return Refund.model_validate(resp)

list(charge_id, *, limit=None, cursor=None, extra_params=None)

List refunds for a charge (passthrough dict to preserve API fields/pagination).

Source code in univapay/resources/refunds.py

def list(
    self,
    charge_id: str,
    *,
    limit: Optional[int] = None,
    cursor: Optional[str] = None,
    extra_params: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """
    List refunds for a charge (passthrough dict to preserve API fields/pagination).
    """
    _validate_id("charge_id", charge_id)

    params: Dict[str, Any] = {}
    if limit is not None:
        if not isinstance(limit, int) or limit <= 0:
            raise ValueError("limit must be a positive integer.")
        params["limit"] = limit
    if cursor:
        params["cursor"] = cursor
    if extra_params:
        params.update(extra_params)

    dprint("refunds.list()", {"charge_id": charge_id, "params": params})
    resp = self.client.get(_base(self.client, charge_id), params=params)
    djson("refunds.list response", resp)
    return resp

wait_until_terminal(charge_id, refund_id, *, server_polling=True, timeout_s=60, interval_s=2.0)

Block until the refund reaches a terminal-ish status.

If server_polling=True, perform a single GET with polling=true. Otherwise, poll client-side every interval_s until timeout_s is reached.

Source code in univapay/resources/refunds.py

def wait_until_terminal(
    self,
    charge_id: str,
    refund_id: str,
    *,
    server_polling: bool = True,
    timeout_s: int = 60,
    interval_s: float = 2.0,
) -> Refund:
    """
    Block until the refund reaches a terminal-ish status.

    If server_polling=True, perform a single GET with polling=true.
    Otherwise, poll client-side every `interval_s` until `timeout_s` is reached.
    """
    _validate_id("charge_id", charge_id)
    _validate_id("refund_id", refund_id)
    if timeout_s <= 0 or interval_s <= 0:
        raise ValueError("timeout_s and interval_s must be positive.")

    dprint(
        "refunds.wait_until_terminal()",
        {
            "charge_id": charge_id,
            "refund_id": refund_id,
            "server_polling": server_polling,
            "timeout_s": timeout_s,
            "interval_s": interval_s,
        },
    )

    if server_polling:
        # Single blocking call if the API supports server-side polling
        return self.get(charge_id, refund_id, polling=True)

    # Client-side polling loop
    deadline = time.time() + timeout_s
    last = self.get(charge_id, refund_id, polling=False)
    if _is_terminal(last.status):
        dprint("refunds.wait_until_terminal: already terminal-ish", {"status": last.status})
        return last

    while time.time() < deadline:
        time.sleep(interval_s)
        last = self.get(charge_id, refund_id, polling=False)
        if _is_terminal(last.status):
            dprint("refunds.wait_until_terminal -> terminal-ish", {"status": last.status})
            return last

    dprint("refunds.wait_until_terminal -> timeout", {"last_status": getattr(last, "status", None)})
    return last

Cancel helpers for charges (authorization/charge cancel).

Most accounts use

POST /charges/{charge_id}/cancel

Depending on your capture flow, this may be equivalent to voiding an authorization. This SDK provides a void_authorization alias for clarity.

Source code in univapay/resources/cancels.py

class CancelsAPI:
    """
    Cancel helpers for charges (authorization/charge cancel).

    Most accounts use:
        POST /charges/{charge_id}/cancel

    Depending on your capture flow, this may be equivalent to voiding an
    authorization. This SDK provides a `void_authorization` alias for clarity.
    """

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

    def cancel_charge(
        self,
        charge_id: str,
        *,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Charge:
        """
        Cancel a charge (or an authorization prior to capture).

        Parameters
        ----------
        charge_id : str
            The charge identifier to cancel.
        idempotency_key : Optional[str]
            Recommended for safe retries.
        **extra :
            Additional fields supported by your Univapay account.

        Returns
        -------
        Charge
            The canceled/voided charge resource (typed).
        """
        _validate_id("charge_id", charge_id)

        path = f"{self.client._path('charges')}/{charge_id}/cancel"
        dprint(
            "cancels.cancel_charge()",
            {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)},
        )
        body = {**extra} if extra else {}
        djson("cancels.cancel_charge body", body)

        resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
        return Charge.model_validate(resp)

    # ---------------------------------------------------------------------
    # Alias for accounts that conceptually distinguish "void" vs "cancel".
    # This calls the same endpoint as `cancel_charge` by default.
    # ---------------------------------------------------------------------
    def void_authorization(
        self,
        charge_id: str,
        *,
        idempotency_key: Optional[str] = None,
        **extra: Any,
    ) -> Charge:
        """
        Void a previously authorized (not yet captured) charge.

        Notes
        -----
        Many Univapay setups map this to the same route as cancel:
            POST /charges/{charge_id}/cancel

        Returns
        -------
        Charge
        """
        dprint("cancels.void_authorization() -> cancel_charge()", {"charge_id": charge_id})
        return self.cancel_charge(
            charge_id,
            idempotency_key=idempotency_key,
            **extra,
        )

client = client instance-attribute

__init__(client)

Source code in univapay/resources/cancels.py

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

cancel_charge(charge_id, *, idempotency_key=None, **extra)

Cancel a charge (or an authorization prior to capture).

Parameters

charge_id : str The charge identifier to cancel. idempotency_key : Optional[str] Recommended for safe retries. **extra : Additional fields supported by your Univapay account.

Returns

Charge The canceled/voided charge resource (typed).

Source code in univapay/resources/cancels.py

def cancel_charge(
    self,
    charge_id: str,
    *,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Charge:
    """
    Cancel a charge (or an authorization prior to capture).

    Parameters
    ----------
    charge_id : str
        The charge identifier to cancel.
    idempotency_key : Optional[str]
        Recommended for safe retries.
    **extra :
        Additional fields supported by your Univapay account.

    Returns
    -------
    Charge
        The canceled/voided charge resource (typed).
    """
    _validate_id("charge_id", charge_id)

    path = f"{self.client._path('charges')}/{charge_id}/cancel"
    dprint(
        "cancels.cancel_charge()",
        {"charge_id": charge_id, "idempotency_key_present": bool(idempotency_key)},
    )
    body = {**extra} if extra else {}
    djson("cancels.cancel_charge body", body)

    resp = self.client.post(path, json=body, idempotency_key=idempotency_key)
    return Charge.model_validate(resp)

void_authorization(charge_id, *, idempotency_key=None, **extra)

Void a previously authorized (not yet captured) charge.

Notes

Many Univapay setups map this to the same route as cancel: POST /charges/{charge_id}/cancel

Returns

Charge

Source code in univapay/resources/cancels.py

def void_authorization(
    self,
    charge_id: str,
    *,
    idempotency_key: Optional[str] = None,
    **extra: Any,
) -> Charge:
    """
    Void a previously authorized (not yet captured) charge.

    Notes
    -----
    Many Univapay setups map this to the same route as cancel:
        POST /charges/{charge_id}/cancel

    Returns
    -------
    Charge
    """
    dprint("cancels.void_authorization() -> cancel_charge()", {"charge_id": charge_id})
    return self.cancel_charge(
        charge_id,
        idempotency_key=idempotency_key,
        **extra,
    )

Transaction Tokens API (read-only on the server).

Notes: - Transaction tokens are typically created client-side by the Univapay widget. - Use this API to fetch token details server-side when needed (e.g., auditing).

Source code in univapay/resources/tokens.py

class TokensAPI:
    """
    Transaction Tokens API (read-only on the server).

    Notes:
    - Transaction tokens are typically created client-side by the Univapay widget.
    - Use this API to fetch token details server-side when needed (e.g., auditing).
    """

    def __init__(self, client: UnivapayClient):
        self.client = client
        dprint("tokens.__init__()", {"base_path": _tokens_base(self.client)})

    def get(self, token_id: str) -> TransactionToken:
        """
        Fetch a transaction token by ID.

        Parameters
        ----------
        token_id : str
            The transaction token id (from the FE widget callback).

        Returns
        -------
        TransactionToken
            A typed model of the token response (extra fields are preserved).

        Raises
        ------
        ValueError
            If token_id is empty/invalid.
        UnivapayHTTPError
            If the HTTP call fails.
        """
        _validate_id("token_id", token_id)
        dprint("tokens.get()", {"token_id": token_id})
        resp = self.client.get(f"{_tokens_base(self.client)}/{token_id}")
        djson("tokens.get response", resp)
        return TransactionToken.model_validate(resp)

    def try_get(self, token_id: str) -> Optional[TransactionToken]:
        """
        Like `get()` but returns None if the token does not exist (HTTP 404).
        Propagates other HTTP errors.
        """
        _validate_id("token_id", token_id)
        dprint("tokens.try_get()", {"token_id": token_id})
        try:
            return self.get(token_id)
        except UnivapayHTTPError as e:
            if e.status == 404:
                dprint("tokens.try_get: not found", {"token_id": token_id})
                return None
            dprint("tokens.try_get: error", {"status": e.status})
            raise

client = client instance-attribute

__init__(client)

Source code in univapay/resources/tokens.py

def __init__(self, client: UnivapayClient):
    self.client = client
    dprint("tokens.__init__()", {"base_path": _tokens_base(self.client)})

get(token_id)

Fetch a transaction token by ID.

Parameters

token_id : str The transaction token id (from the FE widget callback).

Returns

TransactionToken A typed model of the token response (extra fields are preserved).

Raises

ValueError If token_id is empty/invalid. UnivapayHTTPError If the HTTP call fails.

Source code in univapay/resources/tokens.py

def get(self, token_id: str) -> TransactionToken:
    """
    Fetch a transaction token by ID.

    Parameters
    ----------
    token_id : str
        The transaction token id (from the FE widget callback).

    Returns
    -------
    TransactionToken
        A typed model of the token response (extra fields are preserved).

    Raises
    ------
    ValueError
        If token_id is empty/invalid.
    UnivapayHTTPError
        If the HTTP call fails.
    """
    _validate_id("token_id", token_id)
    dprint("tokens.get()", {"token_id": token_id})
    resp = self.client.get(f"{_tokens_base(self.client)}/{token_id}")
    djson("tokens.get response", resp)
    return TransactionToken.model_validate(resp)

try_get(token_id)

Like get() but returns None if the token does not exist (HTTP 404). Propagates other HTTP errors.

Source code in univapay/resources/tokens.py

def try_get(self, token_id: str) -> Optional[TransactionToken]:
    """
    Like `get()` but returns None if the token does not exist (HTTP 404).
    Propagates other HTTP errors.
    """
    _validate_id("token_id", token_id)
    dprint("tokens.try_get()", {"token_id": token_id})
    try:
        return self.get(token_id)
    except UnivapayHTTPError as e:
        if e.status == 404:
            dprint("tokens.try_get: not found", {"token_id": token_id})
            return None
        dprint("tokens.try_get: error", {"status": e.status})
        raise

Webhooks

Bases: BaseModel

Generic Univapay webhook envelope (best-effort typed). We keep it permissive so unknown fields won't break you.

Source code in univapay/resources/webhooks.py

class WebhookEvent(BaseModel):
    """
    Generic Univapay webhook envelope (best-effort typed).
    We keep it permissive so unknown fields won't break you.
    """
    id: Optional[str] = None
    type: Optional[str] = None
    resource_type: Optional[str] = Field(None, alias="resourceType")
    created_on: Optional[str] = Field(None, alias="createdOn")
    created: Optional[str] = None  # some payloads use `created`
    mode: Optional[str] = None     # "test" / "live", etc.

    # Full raw payload for convenience
    data: Dict[str, Any] = Field(default_factory=dict)

    model_config = ConfigDict(extra="allow", populate_by_name=True)

created = None class-attribute instance-attribute

created_on = Field(None, alias='createdOn') class-attribute instance-attribute

data = Field(default_factory=dict) class-attribute instance-attribute

id = None class-attribute instance-attribute

mode = None class-attribute instance-attribute

model_config = ConfigDict(extra='allow', populate_by_name=True) class-attribute instance-attribute

resource_type = Field(None, alias='resourceType') class-attribute instance-attribute

type = None class-attribute instance-attribute

Minimal event router

router = WebhookRouter() @router.on("charge.successful") def _h(e): ...

wildcard handler:

@router.on("*") def _all(e): ...

info, event = verify_and_parse(body=..., headers=..., secret=...) results = router.dispatch(event)

Source code in univapay/resources/webhooks.py

class WebhookRouter:
    """
    Minimal event router:
        router = WebhookRouter()
        @router.on("charge.successful")
        def _h(e): ...
        # wildcard handler:
        @router.on("*")
        def _all(e): ...

        info, event = verify_and_parse(body=..., headers=..., secret=...)
        results = router.dispatch(event)
    """
    def __init__(self) -> None:
        self._map: Dict[str, List[Handler]] = {}

    def on(self, event_type: str) -> Callable[[Handler], Handler]:
        if not event_type or not isinstance(event_type, str):
            raise ValueError("event_type must be a non-empty string (or '*').")

        def _decorator(func: Handler) -> Handler:
            self._map.setdefault(event_type, []).append(func)
            dprint("webhooks.router.on()", {"event_type": event_type, "handler": getattr(func, "__name__", "handler")})
            return func

        return _decorator

    def add(self, event_type: str, func: Handler) -> None:
        self._map.setdefault(event_type, []).append(func)
        dprint("webhooks.router.add()", {"event_type": event_type, "handler": getattr(func, "__name__", "handler")})

    def handlers_for(self, event_type: Optional[str]) -> Iterable[Handler]:
        if not event_type:
            # no type => only wildcard
            return self._map.get("*", [])
        return [*self._map.get(event_type, []), *self._map.get("*", [])]

    def dispatch(self, event: WebhookEvent) -> List[Any]:
        dprint("webhooks.router.dispatch()", {"type": event.type})
        out: List[Any] = []
        for fn in self.handlers_for(event.type):
            try:
                res = fn(event)
                out.append(res)
            except Exception as e:
                dprint("webhooks.router handler error", {"handler": getattr(fn, "__name__", "handler"), "error": repr(e)})
                out.append(e)
        return out

__init__()

Source code in univapay/resources/webhooks.py

def __init__(self) -> None:
    self._map: Dict[str, List[Handler]] = {}

add(event_type, func)

Source code in univapay/resources/webhooks.py

def add(self, event_type: str, func: Handler) -> None:
    self._map.setdefault(event_type, []).append(func)
    dprint("webhooks.router.add()", {"event_type": event_type, "handler": getattr(func, "__name__", "handler")})

dispatch(event)

Source code in univapay/resources/webhooks.py

def dispatch(self, event: WebhookEvent) -> List[Any]:
    dprint("webhooks.router.dispatch()", {"type": event.type})
    out: List[Any] = []
    for fn in self.handlers_for(event.type):
        try:
            res = fn(event)
            out.append(res)
        except Exception as e:
            dprint("webhooks.router handler error", {"handler": getattr(fn, "__name__", "handler"), "error": repr(e)})
            out.append(e)
    return out

handlers_for(event_type)

Source code in univapay/resources/webhooks.py

def handlers_for(self, event_type: Optional[str]) -> Iterable[Handler]:
    if not event_type:
        # no type => only wildcard
        return self._map.get("*", [])
    return [*self._map.get(event_type, []), *self._map.get("*", [])]

on(event_type)

Source code in univapay/resources/webhooks.py

def on(self, event_type: str) -> Callable[[Handler], Handler]:
    if not event_type or not isinstance(event_type, str):
        raise ValueError("event_type must be a non-empty string (or '*').")

    def _decorator(func: Handler) -> Handler:
        self._map.setdefault(event_type, []).append(func)
        dprint("webhooks.router.on()", {"event_type": event_type, "handler": getattr(func, "__name__", "handler")})
        return func

    return _decorator

Parse and (optionally) verify a Univapay webhook.

Parameters:

Name	Type	Description	Default
body	Union[str, bytes, bytearray]	raw request body (bytes/str).	required
headers	Mapping[str, str]	incoming HTTP headers (case-insensitive handling).	required
secret	Optional[Union[str, bytes]]	your webhook signing secret (None + skip_verification=True for dev).	None
header_name	Optional[str]	force a specific signature header name (optional).	None
tolerance_s	int	timestamp tolerance when signature contains a timestamp.	5 * 60
skip_verification	bool	set True for local/dev only.	False

Returns:

Type	Description
WebhookEvent	WebhookEvent

Source code in univapay/resources/webhooks.py

def parse_event(
    *,
    body: Union[str, bytes, bytearray],
    headers: Mapping[str, str],
    secret: Optional[Union[str, bytes]] = None,
    header_name: Optional[str] = None,
    tolerance_s: int = 5 * 60,
    skip_verification: bool = False,
) -> WebhookEvent:
    """
    Parse and (optionally) verify a Univapay webhook.

    Args:
      body: raw request body (bytes/str).
      headers: incoming HTTP headers (case-insensitive handling).
      secret: your webhook signing secret (None + skip_verification=True for dev).
      header_name: force a specific signature header name (optional).
      tolerance_s: timestamp tolerance when signature contains a timestamp.
      skip_verification: set True for local/dev only.

    Returns:
      WebhookEvent
    """
    dprint("webhooks.parse_event() begin", {
        "skip_verification": skip_verification,
        "header_override": header_name,
    })

    # Verify signature first (unless skipped)
    verify_signature(
        payload=body,
        headers=headers,
        secret=secret,
        header_name=header_name,
        tolerance_s=tolerance_s,
        skip_verification=skip_verification,
    )

    # Decode JSON
    if isinstance(body, (bytes, bytearray)):
        raw_text = body.decode("utf-8", errors="replace")
    else:
        raw_text = body

    try:
        payload = json.loads(raw_text) if raw_text else {}
    except Exception as e:
        raise WebhookVerificationError(f"invalid JSON body: {e}") from e

    djson("webhooks.parse_event() payload", payload)

    # Lift common fields if present
    ev = WebhookEvent(
        id=payload.get("id"),
        type=payload.get("type") or payload.get("event") or payload.get("event_type"),
        resource_type=payload.get("resourceType") or payload.get("resource_type"),
        created_on=payload.get("createdOn") or payload.get("created_on"),
        created=payload.get("created"),
        mode=payload.get("mode"),
        data=payload,
    )
    djson("webhooks.parse_event() event", ev.model_dump(mode="json"))
    return ev

Verify HMAC signatures using common header conventions. Returns details dict (including which header matched). Raises WebhookVerificationError on failure (unless skip_verification=True).

DEV NOTE

• If secret is None/empty and skip_verification=False -> raise.
• If skip_verification=True, we log and return without checking.

Source code in univapay/resources/webhooks.py

def verify_signature(
    *,
    payload: Union[str, bytes, bytearray],
    headers: Mapping[str, str],
    secret: Optional[Union[str, bytes]],
    header_name: Optional[str] = None,
    tolerance_s: int = 5 * 60,   # used when header has a timestamp like t=...
    skip_verification: bool = False,
) -> Dict[str, Any]:
    """
    Verify HMAC signatures using common header conventions.
    Returns details dict (including which header matched).
    Raises WebhookVerificationError on failure (unless skip_verification=True).

    DEV NOTE:
      - If `secret` is None/empty and skip_verification=False -> raise.
      - If `skip_verification=True`, we log and return without checking.
    """
    dprint("webhooks.verify_signature() start", {
        "skip_verification": skip_verification,
        "header_name_override": bool(header_name),
        "tolerance_s": tolerance_s,
    })

    if skip_verification:
        dprint("webhooks.verify_signature() skipped (dev mode)")
        return {"skipped": True}

    if not secret:
        raise WebhookVerificationError("webhook secret missing")

    # Find signature header
    sig_value, found_name = _find_sig_header(headers, header_name)
    dprint("webhooks.verify_signature() header found", {"header": found_name, "value_len": len(sig_value)})

    # Normalize payload for hashing
    if isinstance(payload, (bytes, bytearray)):
        body_bytes = bytes(payload)
    else:
        body_bytes = payload.encode("utf-8")

    # Parse header
    parsed = _parse_sig_header(sig_value)
    djson("webhooks.verify_signature() parsed header", parsed)

    ok = False
    reason = "no_match"

    # Case 1: timestamped (t=..., v1=...) - assume v1 is SHA-256 HMAC of "t.<body>"
    if "t" in parsed and "v1" in parsed:
        t_val = parsed["t"]
        sig_hex = parsed["v1"]
        try:
            ts = int(t_val)
            now = int(time.time())
        except Exception as e:
            raise WebhookVerificationError(f"signature verification failed: bad_timestamp:{e}")

        # Enforce tolerance first
        if abs(now - ts) > int(tolerance_s):
            dprint("webhooks.verify_signature() timestamp out of tolerance", {"now": now, "ts": ts})
            raise WebhookVerificationError("signature verification failed: timestamp_out_of_tolerance")

        # canonical message is "<timestamp>.<body>"
        signed_payload = f"{t_val}.".encode("utf-8") + body_bytes
        comp = _hmac_hex(secret, signed_payload, "sha256")
        ok = _cmp(comp, sig_hex)
        if not ok:
            reason = "mismatch"

    # Case 2: sha256=... (GitHub-like)
    elif "sha256" in parsed:
        comp = _hmac_hex(secret, body_bytes, "sha256")
        ok = _cmp(comp, parsed["sha256"])
        if not ok:
            reason = "mismatch"

    # Case 3: sha1=... (rare but supported for flexibility)
    elif "sha1" in parsed:
        comp = _hmac_hex(secret, body_bytes, "sha1")
        ok = _cmp(comp, parsed["sha1"])
        if not ok:
            reason = "mismatch"

    # Case 4: raw hex in header
    elif "raw" in parsed:
        comp = _hmac_hex(secret, body_bytes, "sha256")
        ok = _cmp(comp, parsed["raw"])
        if not ok:
            reason = "mismatch"

    else:
        reason = "unsupported_header_format"

    dprint("webhooks.verify_signature() result", {"ok": ok, "reason": reason, "header": found_name})
    if not ok:
        raise WebhookVerificationError(f"signature verification failed: {reason}")

    return {"ok": True, "header": found_name, "reason": "verified"}

Widgets

__all__ = ['build_one_time_widget_config', 'build_subscription_widget_config', 'build_recurring_widget_config', 'build_widget_bundle_envelope', 'widget_loader_src', 'to_json'] module-attribute

build_one_time_widget_config(*, amount, form_id, button_id, description, widget_key='oneTime', app_jwt=None, env=None, base_config=None, callbacks=None, api=None, payment_methods=None, **extra)

Build FE config for a ONE-TIME payment widget. - Supports enabling/disabling card/paidy/online brands/konbini/bank_transfer via payment_methods.

Source code in univapay/widgets.py

def build_one_time_widget_config(
    *,
    amount: int,
    form_id: str,
    button_id: str,
    description: str,
    widget_key: str = "oneTime",
    app_jwt: Optional[str] = None,
    env: Mapping[str, str] | None = None,
    base_config: Optional[Mapping[str, Any]] = None,
    callbacks: Optional[Mapping[str, Any]] = None,
    api: Optional[Mapping[str, Any]] = None,
    payment_methods: Optional[Mapping[str, Any]] = None,
    **extra: Any,
) -> Dict[str, Any]:
    """
    Build FE config for a ONE-TIME payment widget.
    - Supports enabling/disabling card/paidy/online brands/konbini/bank_transfer via `payment_methods`.
    """
    _validate_amount(amount)
    _validate_id("form_id", form_id)
    _validate_id("button_id", button_id)
    _validate_id("description", description)

    app_id = _require_env_app_id(app_jwt, env)
    methods = _normalize_payment_methods(payment_methods)
    widget = {
        "checkout": "payment",
        "amount": amount,
        "formId": form_id.strip(),
        "buttonId": button_id.strip(),
        "description": description.strip(),
        # New: pass method toggles for FE to filter options
        "paymentMethods": methods,
        **extra,
    }
    dprint("widgets: build_one_time", {"widget_key": widget_key, "amount": amount})
    djson("widgets.one_time.widget", widget)
    return _envelope_single(
        app_id=app_id,
        widget_key=widget_key,
        widget=widget,
        base_config=base_config,
        callbacks=callbacks,
        api=api,
    )

build_recurring_widget_config(*, amount, form_id, button_id, description, widget_key='recurring', app_jwt=None, env=None, base_config=None, callbacks=None, api=None, payment_methods=None, **extra)

Build FE config for a RECURRING payment widget (tokenize card for merchant-initiated charges). - Recurring is effectively a card-token flow; other methods will be debug-warned if supplied.

Source code in univapay/widgets.py

def build_recurring_widget_config(
    *,
    amount: int,
    form_id: str,
    button_id: str,
    description: str,
    widget_key: str = "recurring",
    app_jwt: Optional[str] = None,
    env: Mapping[str, str] | None = None,
    base_config: Optional[Mapping[str, Any]] = None,
    callbacks: Optional[Mapping[str, Any]] = None,
    api: Optional[Mapping[str, Any]] = None,
    payment_methods: Optional[Mapping[str, Any]] = None,
    **extra: Any,
) -> Dict[str, Any]:
    """
    Build FE config for a RECURRING payment widget (tokenize card for merchant-initiated charges).
    - Recurring is effectively a card-token flow; other methods will be debug-warned if supplied.
    """
    _validate_amount(amount)
    _validate_id("form_id", form_id)
    _validate_id("button_id", button_id)
    _validate_id("description", description)

    app_id = _require_env_app_id(app_jwt, env)
    methods = _normalize_payment_methods(payment_methods)
    _warn_incompatible("recurring", methods)

    widget = {
        "checkout": "payment",
        "tokenType": "recurring",
        "amount": amount,
        "formId": form_id.strip(),
        "buttonId": button_id.strip(),
        "description": description.strip(),
        "paymentMethods": methods,
        **extra,
    }
    dprint("widgets: build_recurring", {"widget_key": widget_key, "amount": amount})
    djson("widgets.recurring.widget", widget)
    return _envelope_single(
        app_id=app_id,
        widget_key=widget_key,
        widget=widget,
        base_config=base_config,
        callbacks=callbacks,
        api=api,
    )

build_subscription_widget_config(*, amount, period, form_id, button_id, description, widget_key='subscription', app_jwt=None, env=None, base_config=None, callbacks=None, api=None, payment_methods=None, **extra)

Build FE config for a SUBSCRIPTION payment widget. - Typically card; other methods will be debug-warned if supplied.

Source code in univapay/widgets.py

def build_subscription_widget_config(
    *,
    amount: int,
    period: str,                 # e.g., "monthly", "semiannually", "yearly"
    form_id: str,
    button_id: str,
    description: str,
    widget_key: str = "subscription",
    app_jwt: Optional[str] = None,
    env: Mapping[str, str] | None = None,
    base_config: Optional[Mapping[str, Any]] = None,
    callbacks: Optional[Mapping[str, Any]] = None,
    api: Optional[Mapping[str, Any]] = None,
    payment_methods: Optional[Mapping[str, Any]] = None,
    **extra: Any,
) -> Dict[str, Any]:
    """
    Build FE config for a SUBSCRIPTION payment widget.
    - Typically card; other methods will be debug-warned if supplied.
    """
    _validate_amount(amount)
    p = _validate_period(period)
    _validate_id("form_id", form_id)
    _validate_id("button_id", button_id)
    _validate_id("description", description)

    app_id = _require_env_app_id(app_jwt, env)
    methods = _normalize_payment_methods(payment_methods)
    _warn_incompatible("subscription", methods)

    widget = {
        "checkout": "payment",
        "tokenType": "subscription",
        "subscriptionPeriod": p,
        "amount": amount,
        "formId": form_id.strip(),
        "buttonId": button_id.strip(),
        "description": description.strip(),
        "paymentMethods": methods,
        **extra,
    }
    dprint("widgets: build_subscription", {"widget_key": widget_key, "amount": amount, "period": p})
    djson("widgets.subscription.widget", widget)
    return _envelope_single(
        app_id=app_id,
        widget_key=widget_key,
        widget=widget,
        base_config=base_config,
        callbacks=callbacks,
        api=api,
    )

build_widget_bundle_envelope(*, widgets, app_jwt=None, env=None, base_config=None, callbacks=None, api=None)

Build a single payload with multiple widgets

widgets = { "oneTimeAlpha": {...}, # result of builders below (already normalized) "subscriptionSemiannual": {...}, "recurringVault": {...} }

Source code in univapay/widgets.py

def build_widget_bundle_envelope(
    *,
    widgets: Mapping[str, Mapping[str, Any]],
    app_jwt: Optional[str] = None,
    env: Mapping[str, str] | None = None,
    base_config: Optional[Mapping[str, Any]] = None,
    callbacks: Optional[Mapping[str, Any]] = None,
    api: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
    """
    Build a single payload with multiple widgets:
      widgets = {
        "oneTimeAlpha": {...},         # result of builders below (already normalized)
        "subscriptionSemiannual": {...},
        "recurringVault": {...}
      }
    """
    app_id = _require_env_app_id(app_jwt, env)
    payload = {
        "appId": app_id,
        "baseConfig": _normalize_base_config(base_config),
        "widgets": {k: dict(v) for k, v in widgets.items()},
        "callbacks": _normalize_callbacks(callbacks),
        "api": _normalize_api(api),
    }
    dprint("widgets: bundle envelope built", {"count": len(widgets)})
    djson("widgets.envelope", {**payload, "appId": _mask_token(app_id)})
    return payload

to_json(payload, *, pretty=False)

Serialize any widget envelope to JSON (useful in tests or manual output).

Source code in univapay/widgets.py

def to_json(payload: Dict[str, Any], *, pretty: bool = False) -> str:
    """
    Serialize any widget envelope to JSON (useful in tests or manual output).
    """
    s = json.dumps(
        payload,
        ensure_ascii=False,
        indent=2 if pretty else None,
        separators=None if pretty else (",", ":"),
    )
    dprint("widgets: to_json length", {"chars": len(s)})
    return s

widget_loader_src(env=None)

Return the official Univapay widget loader URL, optionally overridden by env.

Env override key: "UNIVAPAY_WIDGET_URL" Default: "https://widget.univapay.com/client/checkout.js"

Source code in univapay/widgets.py

def widget_loader_src(env: Mapping[str, str] | None = None) -> str:
    """
    Return the official Univapay widget loader URL, optionally overridden by env.

    Env override key: "UNIVAPAY_WIDGET_URL"
    Default: "https://widget.univapay.com/client/checkout.js"
    """
    env = env or os.environ
    return (env.get("UNIVAPAY_WIDGET_URL") or "https://widget.univapay.com/client/checkout.js").strip()

Errors

Bases: Exception

Base exception for all Univapay SDK errors.

Source code in univapay/errors.py

class UnivapaySDKError(Exception):
    """Base exception for all Univapay SDK errors."""
    pass

Bases: UnivapaySDKError

Raised when configuration/credentials are invalid or missing.

Source code in univapay/errors.py

class UnivapayConfigError(UnivapaySDKError):
    """Raised when configuration/credentials are invalid or missing."""
    pass

Bases: UnivapaySDKError

Unified HTTP error for API requests.

Attributes

status : int HTTP status code (or -1 for network errors). payload : Any Parsed JSON or fallback body describing the error (kept verbatim). request_id : Optional[str] Server-provided request correlation id, if available. method : Optional[str] Best-effort HTTP method that triggered the error (if provided). url : Optional[str] Best-effort URL that triggered the error (if provided).

Convenience

.code -> extracted error code (if present in payload) .message_text -> human-friendly error message .retryable -> bool, True if typical transient status (429, 500-504) .to_dict() -> sanitized summary dict for logging

Source code in univapay/errors.py

class UnivapayHTTPError(UnivapaySDKError):
    """
    Unified HTTP error for API requests.

    Attributes
    ----------
    status : int
        HTTP status code (or -1 for network errors).
    payload : Any
        Parsed JSON or fallback body describing the error (kept verbatim).
    request_id : Optional[str]
        Server-provided request correlation id, if available.
    method : Optional[str]
        Best-effort HTTP method that triggered the error (if provided).
    url : Optional[str]
        Best-effort URL that triggered the error (if provided).

    Convenience
    -----------
    .code            -> extracted error code (if present in payload)
    .message_text    -> human-friendly error message
    .retryable       -> bool, True if typical transient status (429, 500-504)
    .to_dict()       -> sanitized summary dict for logging
    """

    def __init__(
        self,
        status: int,
        payload: Any,
        request_id: Optional[str] = None,
        *,
        method: Optional[str] = None,
        url: Optional[str] = None,
    ):
        self.status = int(status)
        self.payload = payload
        self.request_id = request_id
        self.method = method
        self.url = url

        # Debug output (payload is printed via djson; avoid leaking secrets)
        dprint("UnivapayHTTPError", {
            "status": self.status,
            "request_id": self.request_id,
            "method": self.method,
            "url": self.url,
        })
        djson("UnivapayHTTPError payload", self.payload)

        super().__init__(self._message())

    # ---------------- convenience properties ----------------

    @property
    def retryable(self) -> bool:
        """Return True for common transient HTTP statuses."""
        return self.status in (429, 500, 502, 503, 504)

    @property
    def code(self) -> Optional[str]:
        """Try to extract a structured error code from the payload if present."""
        p = self.payload
        try:
            if isinstance(p, dict):
                # common shapes: {"code": "..."} or {"error": {"code": "..."}} or {"error_code": "..."}
                if isinstance(p.get("error"), dict) and "code" in p["error"]:
                    return str(p["error"]["code"])
                if "code" in p:
                    return str(p["code"])
                if "error_code" in p:
                    return str(p["error_code"])
            return None
        except Exception:
            return None

    @property
    def message_text(self) -> str:
        """
        Human-friendly message guessed from payload.
        Keeps it short and safe for logs.
        """
        p = self.payload
        # strings straight through
        if isinstance(p, str):
            return p.strip() or "error"
        # dict heuristics
        if isinstance(p, dict):
            for key in ("message", "error", "detail", "description"):
                val = p.get(key)
                if isinstance(val, str) and val.strip():
                    return val.strip()
                # nested {"error": {"message": "..."}}
                if isinstance(val, dict):
                    nested = val.get("message") or val.get("detail") or val.get("description")
                    if isinstance(nested, str) and nested.strip():
                        return nested.strip()
            # fallback: very short JSON preview
            try:
                import json
                s = json.dumps(p, ensure_ascii=False)
                return s if len(s) <= 240 else s[:237] + "..."
            except Exception:
                return "error"
        # other types
        try:
            return str(p)
        except Exception:
            return "error"

    # ---------------- rendering & serialization ----------------

    def _message(self) -> str:
        rid = f" req_id={self.request_id}" if self.request_id else ""
        meth = f" {self.method}" if self.method else ""
        url = f" {self.url}" if self.url else ""
        code = f" code={self.code}" if self.code else ""
        msg = self.message_text
        # Keep single-line & compact; payload already printed by djson()
        return f"HTTP {self.status}{meth}{url}{rid}{code}: {msg}"

    def __str__(self) -> str:
        return self._message()

    def __repr__(self) -> str:
        return f"UnivapayHTTPError(status={self.status}, request_id={self.request_id!r}, code={self.code!r})"

    def to_dict(self) -> Dict[str, Any]:
        """Sanitized summary for logs/telemetry; includes only non-sensitive fields."""
        return {
            "status": self.status,
            "request_id": self.request_id,
            "method": self.method,
            "url": self.url,
            "code": self.code,
            "message": self.message_text,
            "retryable": self.retryable,
        }

code property

Try to extract a structured error code from the payload if present.

message_text property

Human-friendly message guessed from payload. Keeps it short and safe for logs.

method = method instance-attribute

payload = payload instance-attribute

request_id = request_id instance-attribute

retryable property

Return True for common transient HTTP statuses.

status = int(status) instance-attribute

url = url instance-attribute

__init__(status, payload, request_id=None, *, method=None, url=None)

Source code in univapay/errors.py

def __init__(
    self,
    status: int,
    payload: Any,
    request_id: Optional[str] = None,
    *,
    method: Optional[str] = None,
    url: Optional[str] = None,
):
    self.status = int(status)
    self.payload = payload
    self.request_id = request_id
    self.method = method
    self.url = url

    # Debug output (payload is printed via djson; avoid leaking secrets)
    dprint("UnivapayHTTPError", {
        "status": self.status,
        "request_id": self.request_id,
        "method": self.method,
        "url": self.url,
    })
    djson("UnivapayHTTPError payload", self.payload)

    super().__init__(self._message())

__repr__()

Source code in univapay/errors.py

def __repr__(self) -> str:
    return f"UnivapayHTTPError(status={self.status}, request_id={self.request_id!r}, code={self.code!r})"

__str__()

Source code in univapay/errors.py

def __str__(self) -> str:
    return self._message()

to_dict()

Sanitized summary for logs/telemetry; includes only non-sensitive fields.

Source code in univapay/errors.py

def to_dict(self) -> Dict[str, Any]:
    """Sanitized summary for logs/telemetry; includes only non-sensitive fields."""
    return {
        "status": self.status,
        "request_id": self.request_id,
        "method": self.method,
        "url": self.url,
        "code": self.code,
        "message": self.message_text,
        "retryable": self.retryable,
    }

Bases: UnivapaySDKError

Raised for webhook signature/format errors.

Source code in univapay/errors.py

class UnivapayWebhookError(UnivapaySDKError):
    """Raised for webhook signature/format errors."""
    pass