---
title: "Step 3: Connect to Qikink"
space: "Commera"
url: "https://docs.bwh.tech/commera/build-a-plugin/connect-to-qikink"
updated: "2026-10-09"
---

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](https://github.com/Rl0007/commera_qikink/commit/4ed55d6). Later tidied in [refactor: call frappe's request helpers directly and parse replies with parse_json](https://github.com/Rl0007/commera_qikink/commit/4900622). 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:

```python
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.

:::caution[Name the file client.py]

Do not name this file `qikink.py`. The module folder `commera_qikink/qikink/` from step 1 has the same name, and `from commera_qikink.qikink import Qikink` fails with an `ImportError`.

:::

## 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.

```python
	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
```

## 3. Share one token between workers

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.

```python
	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.

:::danger[A new token logs out the old one]

Qikink keeps one active token for each Client ID. When you get a token in Postman or on a second site with the same keys, your plugin's token stops working. The plugin recovers with one retry, but two systems that share keys take the token from each other on every request. Give each site its own Qikink keys.

:::

## 4. Read Qikink's errors

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

```python
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.

:::note[Every failed request adds an Error Log]

Frappe's `make_get_request` and `make_post_request` write an Error Log before they raise an HTTP error. A refused token on the retry path, or a Qikink 400, adds a row to **Error Log** although the client handles it. These two functions also take no `timeout` argument.

:::

:::caution[The sandbox has no order list]

`GET /api/order/list` returns 404 on the sandbox. `get_order` uses the older `GET /api/order?id=`, which works on the sandbox and on live. It returns a list with one order.

:::

## Check it works

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

```python
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](/commera/build-a-plugin/map-products)
- [Gotchas](/commera/plugin-reference/gotchas): more traps when you call an outside service