Create an identity

import json
import requests

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
job_id = "job_abc123..."

headers = {"Authorization": "Bearer " + access_token}

with requests.get(
    api_url + "/notifications/events",
    headers=headers,
    stream=True,
    timeout=600,
) as response:
    response.raise_for_status()

    for raw_line in response.iter_lines(decode_unicode=True):
        if not raw_line or raw_line.startswith(":"):
            continue
        if not raw_line.startswith("data: "):
            continue

        notification = json.loads(raw_line[6:])
        data = notification.get("data", {})
        task_id = data.get("id_task") or data.get("job_id")
        if task_id != job_id:
            continue

        if notification["name"] == "completed":
            break
        if notification["name"] == "error":
            raise RuntimeError("Identity creation failed")

# Fetch draft results
results = requests.get(
    api_url + f"/jobs/{job_id}/results",
    headers={"Authorization": "Bearer " + access_token},
).json()

for draft in results["results"]:
    print(f"Draft #{draft['image_result_id']}: {draft['url']}")

import hashlib
import hmac

import requests
from flask import Flask, abort, request

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
job_id = "job_abc123..."
public_webhook_url = "https://example.com/webhooks/piktid"

setup = requests.put(
    api_url + "/webhooks",
    headers={"Authorization": "Bearer " + access_token},
    json={"url": public_webhook_url},
).json()
webhook_secret = setup["secret"]

app = Flask(__name__)


def verify_signature(secret: str, body: bytes, signature_header: str) -> bool:
    expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)


@app.post("/webhooks/piktid")
def handle_piktid_webhook():
    body = request.get_data()
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_signature(webhook_secret, body, signature):
        abort(401)

    notification = request.get_json()
    data = notification.get("data", {})
    task_id = data.get("id_task") or data.get("job_id")
    if task_id != job_id:
        return "", 204

    if notification["name"] == "completed":
        print("Creation job completed")
    elif notification["name"] == "error":
        print("Identity creation failed")

    return "", 204


app.run(port=8000)

import json
import requests

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
preprocessing_job_id = "job_pre456..."

headers = {"Authorization": "Bearer " + access_token}

with requests.get(
    api_url + "/notifications/events",
    headers=headers,
    stream=True,
    timeout=600,
) as response:
    response.raise_for_status()

    for raw_line in response.iter_lines(decode_unicode=True):
        if not raw_line or raw_line.startswith(":"):
            continue
        if not raw_line.startswith("data: "):
            continue

        notification = json.loads(raw_line[6:])
        data = notification.get("data", {})
        if notification["name"] != "identity_preprocessing":
            continue
        task_id = data.get("id_task") or data.get("job_id")
        if task_id != preprocessing_job_id:
            continue

        status = data.get("status")
        if status == "completed":
            print("Identity preprocessing completed")
            break
        if status == "failed":
            raise RuntimeError(data.get("error_message", "Preprocessing failed"))

import hashlib
import hmac

import requests
from flask import Flask, abort, request

api_url = "https://v2.api.piktid.com"
access_token = "your_access_token"
preprocessing_job_id = "job_pre456..."
public_webhook_url = "https://example.com/webhooks/piktid"

setup = requests.put(
    api_url + "/webhooks",
    headers={"Authorization": "Bearer " + access_token},
    json={"url": public_webhook_url},
).json()
webhook_secret = setup["secret"]

app = Flask(__name__)


def verify_signature(secret: str, body: bytes, signature_header: str) -> bool:
    expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)


@app.post("/webhooks/piktid")
def handle_piktid_webhook():
    body = request.get_data()
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_signature(webhook_secret, body, signature):
        abort(401)

    notification = request.get_json()
    data = notification.get("data", {})
    if notification.get("name") != "identity_preprocessing":
        return "", 204
    task_id = data.get("id_task") or data.get("job_id")
    if task_id != preprocessing_job_id:
        return "", 204

    status = data.get("status")
    if status == "completed":
        print("Identity preprocessing completed")
    elif status == "failed":
        print(f"Preprocessing failed: {data.get('error_message')}")

    return "", 204


app.run(port=8000)