Перейти к содержимому

Cookbook

Это содержимое пока не доступно на вашем языке.

Complete recipes for the things integrations actually do.

The pattern: create each athlete once, then loop. Quota is per team, so pace yourself — a 40-athlete batch is 40 sessions off the customer’s allowance, and their coaches share that pool.

import time
import requests
BASE = "https://api.ai.aikynetix.app"
HEADERS = {"Authorization": "Bearer aik_yourPartnerTokenHere"}
SQUAD = [("Anna Petrova", "female", 171, 63.5, "anna.mp4"), ...]
def upload(name, gender, height, weight, path):
client = requests.post(f"{BASE}/api/clients/", headers=HEADERS, timeout=30,
json={"display_name": name, "gender": gender,
"height_cm": height, "weight_kg": weight})
client.raise_for_status()
intent = requests.post(f"{BASE}/api/sessions/upload-intent/", headers=HEADERS, timeout=30,
json={"filename": path, "content_type": "video/mp4", "activity": "running"})
intent.raise_for_status()
i = intent.json()
with open(path, "rb") as fh:
put = requests.put(i["url"], data=fh, headers=i["headers"], timeout=600)
put.raise_for_status()
created = requests.post(f"{BASE}/api/sessions/", headers=HEADERS, timeout=30,
json={"id": i["session_id"], "activity": "running", "exercise": "Treadmill",
"video_url": i["public_url"], "client": client.json()["id"],
"camera_view": "side"})
if created.status_code == 402:
raise SystemExit(f"out of quota: {created.json()['detail']}")
created.raise_for_status()
return created.json()["id"]
ids = [upload(*athlete) for athlete in SQUAD]

List with ?from= and ?to= and page through the envelope. Store the session id as your foreign key; it is stable.

def sessions_since(date):
url = f"{BASE}/api/sessions/?from={date}&limit=200"
while url:
r = requests.get(url, headers=HEADERS, timeout=30)
r.raise_for_status()
page = r.json() # { results, count, next, previous }
yield from page["results"]
url = page["next"]

Poll for new ones on a schedule rather than trying to catch them live — there are no webhooks yet. ?from= takes a YYYY-MM-DD local date, combined at the ?tz= you pass, so a daily job in your own timezone stays aligned.

You need three things per session: metrics, metric_targets, and the catalogue for labels and units.

Fetch the catalogue once and cache it — it changes when we add an activity, not per request. Join on the metric key:

catalogue = requests.get(f"{BASE}/api/exercises/?activity=running",
headers=HEADERS, timeout=30).json()
specs = {m["key"]: m for m in catalogue["metrics"]["Treadmill"]}
targets = {t["key"]: t for t in session["metric_targets"]}
for metric in session["metrics"]:
spec = specs.get(metric["key"])
target = targets.get(metric["key"])
if metric["value_num"] is None or spec is None:
continue
in_range = target and target["min_good"] <= metric["value_num"] <= target["max_good"]
print(f"{spec['label']}: {metric['value_num']} {spec['unit']}"
f" {'' if in_range else ''}")

Always render the target next to the value. A number without its range is not actionable.

Three distinct outcomes, three different responses:

if session["status"] == "failed":
# Quota was already refunded. The clip is the problem — surface
# failure_reason to whoever filmed it, and let them re-record.
notify(session["failure_reason"])
elif session["capture_findings"]:
# Completed, but the footage was poor. There IS a report and it DID cost a
# session — but the numbers are not reliable. Do not chart them.
warn_unreliable(session["capture_findings"])
else:
publish(session)

The upload intent’s session_id is your idempotency key. If a create call’s response is lost, retrying with the same id returns {"id": ["session_id_taken"]} rather than creating a duplicate — catch that and fetch the existing session instead of starting over.

503 enqueue_failed is the one error worth retrying blindly: quota was debited, the queue was briefly unreachable, and the same call will work.