Step 3: Connect to Qikink

In this step, you write an API client for Qikink. It gets an access token, keeps one token for all workers, and turns Qikink's errors into messages that a user can read.

Commit: feat: add a qikink api client with a shared cached token. Later tidied in refactor: call frappe's request helpers directly and parse replies with parse_json. The code on this page is the tidied version.

What you'll build

commera_qikink/client.py, a Qikink class that gets a token, sends requests and retries once.
One cached access token that all workers share.
Error messages that a user can read.

The client follows four rules from the Qikink API:

POST /api/token returns an access token for your Client ID. The token is valid for one hour.
Every other request sends the ClientId and Accesstoken headers.
Each Client ID has one active token. When you get a new token, the old token stops working.
Qikink can answer with HTTP 200 and an error in the body.

1. Create the client

Create commera_qikink/client.py. Start with the settings and the class:

from collections.abc import Callable

import frappe
import requests
from frappe import _
from frappe.integrations.utils import make_get_request, make_post_request
from frappe.utils.data import cint, cstr

SANDBOX_URL = "https://sandbox.qikink.com"
LIVE_URL = "https://api.qikink.com"
# Tokens live 3600 s; renewing early keeps a request from carrying one that expires on the way.
TOKEN_TTL_SECONDS = 3500
TOKEN_LOCK_SECONDS = 30


class QikinkError(frappe.ValidationError):
	pass


class InvalidTokenError(Exception):
	pass


class Qikink:
	def __init__(self):
		settings = frappe.get_cached_doc("Qikink Settings")
		self.client_id = cstr(settings.client_id).strip()
		self.client_secret = settings.get_password("client_secret", raise_exception=False)
		if not self.client_id or not self.client_secret:
			frappe.throw(_("Set the Qikink Client ID and Client Secret in Settings > Qikink."), QikinkError)
		self.base_url = SANDBOX_URL if settings.sandbox else LIVE_URL
		self.token_key = frappe.cache.make_key(f"commera_qikink:token:{self.base_url}:{self.client_id}")

	def get_order(self, order_id: str | int) -> dict:
		return self.request(make_get_request, "/api/order", params={"id": order_id})

get_password with raise_exception=False returns None for an empty secret. Without it, Frappe raises its own error before your message shows.

2. Send a request and retry once

Add these methods to the Qikink class. The caller passes the Frappe request helper, make_get_request or make_post_request, as make_request. request sends with the cached token. When Qikink refuses the token, it clears the token, gets a new one and sends again, one time only.

	def request(self, make_request: Callable, path: str, **kwargs) -> dict:
		token = self.get_token()
		try:
			return self.send(make_request, path, token, **kwargs)
		except InvalidTokenError:
			self.clear_token(token)

		try:
			return self.send(make_request, path, self.get_token(), **kwargs)
		except InvalidTokenError as error:
			frappe.throw(_("Qikink refused the access token: {0}").format(error), QikinkError)

	def send(self, make_request: Callable, path: str, token: str, **kwargs) -> dict:
		headers = {"ClientId": self.client_id, "Accesstoken": token}
		status, body = get_reply(make_request, self.base_url + path, headers=headers, **kwargs)
		if error := get_error(status, body):
			if status == 401 or "token" in error.lower():
				raise InvalidTokenError(error)
			frappe.throw(_("Qikink: {0}").format(error), QikinkError)
		return body

Add the token methods to the class. The token lives in Redis, so every worker and every job uses the same token. Only one process gets a new token at a time, under a lock.

	def get_token(self) -> str:
		if token := self.read_token():
			return token
		# One token per ClientId, and minting a new one kills the old: only one process may mint.
		with frappe.cache.lock(
			f"{self.token_key}:lock", timeout=TOKEN_LOCK_SECONDS, blocking_timeout=TOKEN_LOCK_SECONDS
		):
			return self.read_token() or self.mint_token()

	def mint_token(self) -> str:
		status, body = get_reply(
			make_post_request,
			f"{self.base_url}/api/token",
			data={"ClientId": self.client_id, "client_secret": self.client_secret},
		)
		error = get_error(status, body)
		if error or not body.get("Accesstoken"):
			frappe.throw(_("Qikink did not issue an access token: {0}").format(error or body), QikinkError)

		expires_in = cint(body.get("expires_in"))
		ttl = min(TOKEN_TTL_SECONDS, expires_in - 100) if expires_in > 100 else TOKEN_TTL_SECONDS
		frappe.cache.set(self.token_key, body["Accesstoken"], ex=ttl)
		return body["Accesstoken"]

	def read_token(self) -> str | None:
		# Raw Redis rather than get_value: the request-local copy would hide a token another worker replaced.
		token = frappe.cache.get(self.token_key)
		return token.decode() if token else None

	def clear_token(self, token: str):
		if self.read_token() == token:
			frappe.cache.delete(self.token_key)

clear_token deletes the token only when it is still the refused one. When another worker has already put a new token in Redis, that token stays.

4. Read Qikink's errors

Add these functions at the end of the file, outside the class:

def get_reply(make_request: Callable, url: str, **kwargs) -> tuple[int, object]:
	try:
		return 200, make_request(url, **kwargs)
	except requests.HTTPError as error:
		# The 4xx body carries Qikink's message, and "invalid token" there is what triggers the retry.
		try:
			return error.response.status_code, frappe.parse_json(error.response.text)
		except ValueError:
			return error.response.status_code, error.response.text
	except requests.RequestException as error:
		frappe.throw(_("Could not reach Qikink: {0}").format(error), QikinkError)


def get_error(status: int, body: object) -> str | None:
	# Qikink can answer HTTP 200 with an error body, so the body decides as much as the status.
	if isinstance(body, dict):
		error = body.get("error")
		message = body.get("message") if error is True or body.get("success") is False else error
		if error or message or cint(body.get("status_code")) >= 400 or status >= 400:
			return cstr(message or body)
		return None
	if isinstance(body, list) and status < 400:
		return None
	if status >= 400:
		return cstr(body)[:500] or f"HTTP {status}"
	return _("Unexpected response from Qikink: {0}").format(cstr(body)[:500])

get_reply returns the HTTP status and the reply body. On a 4xx, Frappe's helpers raise requests.HTTPError, so get_reply reads the body with frappe.parse_json. When the body is not JSON, it keeps the raw text.

Qikink puts the error text in different keys. A bad value returns {"error": true, "message": "Invalid value: ..."}, with the text in message. A bad token returns {"error": "Invalid AccessToken or Client Id"}, and an unknown SKU returns {"error": "Invalid SKU", "status_code": "400"}, with the text in error. get_error reads both keys.

Check it works

Open the console with bench --site your.site console, then run:

from commera_qikink.client import Qikink
qikink = Qikink()
len(qikink.get_token())
qikink.get_token() == qikink.get_token()
qikink.get_order("not-a-real-order")

The len line shows the token length, and the comparison shows True: the second call reads the same token from Redis. The last line raises QikinkError: Qikink: Invalid value: not-a-real-order passed for id. You placed no order yet, so this error proves that the client reached Qikink with a valid token.

Next steps

Step 4: Map products
Gotchas: more traps when you call an outside service

Commera

What you'll build

1. Create the client

2. Send a request and retry once

3. Share one token between workers

4. Read Qikink's errors

Check it works

Next steps