Python desktop: your first licensed app

python -m venv .venv

python -m pip install "kdpika-licensing==2.0.1"
python -m tkinter

export KDP_APP_ID="app_YOUR_PUBLIC_ID"
export KDP_ENVIRONMENT="SANDBOX"
python desktop.py

$env:KDP_APP_ID="app_YOUR_PUBLIC_ID"
$env:KDP_ENVIRONMENT="SANDBOX"
python desktop.py

"""Shared application code. Real licensing calls; never a developer management key."""
import csv
import os
from pathlib import Path

from kdpika_licensing import LicenseClient


class AccessDenied(Exception):
    pass


def application_folder() -> Path:
    # Use one private, persistent folder for each app + machine + environment.
    product = os.environ.get("KDP_APP_ID", "").strip()
    if not product.startswith("app_") or not product.replace("_", "").isalnum():
        raise ValueError("Set KDP_APP_ID to your product's public App ID first.")
    environment = os.environ.get("KDP_ENVIRONMENT", "SANDBOX").upper()
    if environment not in {"SANDBOX", "LIVE"}:
        raise ValueError("KDP_ENVIRONMENT must be SANDBOX or LIVE.")
    folder = Path.home() / ".kdpika-python-tutorial" / product / environment.lower()
    folder.mkdir(parents=True, exist_ok=True, mode=0o700)
    return folder


def make_client() -> LicenseClient:
    folder = application_folder()
    return LicenseClient(
        product=os.environ["KDP_APP_ID"],
        environment=os.environ.get("KDP_ENVIRONMENT", "SANDBOX"),
        base_url="https://licensing.kdpika.com",
        application_name="Python Licensing Tutorial",
        application_version="1.0.0",
        storage_dir=folder / "license-state",
    )


def require_csv_access(client: LicenseClient):
    result = client.check()
    if not result.valid or not result.signature_valid:
        raise AccessDenied("Check your license or reconnect, then try again.")
    if result.payload.get("metered"):
        # A normal feature check does NOT charge a credit. Never offer free work by mistake.
        raise AccessDenied("This license uses credits. Run credit_export.py for paid exports.")
    if not result.feature("csv_export"):
        raise AccessDenied("CSV export is not included in this license.")
    return result


def export_csv(client: LicenseClient, text: str) -> Path:
    require_csv_access(client)  # Check again at the protected action, not only at startup.
    path = application_folder() / "export.csv"
    with path.open("w", encoding="utf-8", newline="") as output:
        writer = csv.writer(output)
        writer.writerow(["Message"])
        # Keep spreadsheet applications from interpreting user text as a formula.
        writer.writerow(["'" + text if text.lstrip().startswith(("=", "+", "-", "@")) else text])
    return path

from kdpika_licensing import AllSeatsInUseError

try:
    result = client.checkout()
    if not result.valid or not result.signature_valid:
        show_access_unavailable()
    elif result.feature("csv_export"):
        enable_export_button()
except AllSeatsInUseError:
    show_message("All seats are in use. Close the app on another machine and try again.")
# These UI callback names belong to your own application.
# Still check access again immediately before protected work.

"""Small real desktop app. SDK/network work stays off the Tk event thread."""
from concurrent.futures import ThreadPoolExecutor

from common import AccessDenied, export_csv, make_client


def main():
    import tkinter as tk
    from tkinter import ttk

    client = make_client()
    root = tk.Tk()
    root.title("My licensed Python app")
    root.geometry("560x600")
    root.minsize(360, 560)
    frame = ttk.Frame(root, padding=24)
    frame.pack(fill="both", expand=True)
    frame.columnconfigure(0, weight=1)
    ttk.Label(frame, text="Make your first licensed export", font=("", 16, "bold")).grid(sticky="w")
    status = tk.StringVar(value="Paste your test license and its matching email.")
    status_label = ttk.Label(frame, textvariable=status, wraplength=460)
    status_label.grid(sticky="ew", pady=(10, 18))
    frame.bind("<Configure>", lambda event: status_label.configure(wraplength=max(250, event.width - 48)))
    ttk.Label(frame, text="License key").grid(sticky="w")
    key = ttk.Entry(frame, show="*", font=("", 12))
    key.grid(sticky="ew", pady=(4, 10))
    ttk.Label(frame, text="Email on this license").grid(sticky="w")
    email = ttk.Entry(frame, font=("", 12))
    email.grid(sticky="ew", pady=(4, 10))
    message = ttk.Entry(frame, font=("", 12))
    message.insert(0, "My first licensed export")
    message.grid(sticky="ew", pady=(10, 12))
    actions = ttk.Frame(frame)
    actions.grid(sticky="ew")
    executor = ThreadPoolExecutor(max_workers=1)
    buttons, pending = [], []
    closing = False

    def show_result(result):
        status.set("License accepted. Try Export CSV." if result.valid and result.signature_valid
                   else "Access is unavailable. Check your license or connection.")
        key.delete(0, "end")

    def run(work, done):
        if pending or closing:
            return
        status.set("Checking…")
        for button in buttons:
            button.state(["disabled"])
        pending.append((executor.submit(work), done))

    def show_released(result):
        status.set("Machine released. Activate again before exporting." if result.valid
                   else "The machine could not be released. Reconnect and try again.")

    def activate():
        license_key, customer_email = key.get().strip(), email.get().strip()
        run(lambda: client.activate(license_key, customer_email or None), show_result)

    def export():
        text = message.get()
        run(lambda: export_csv(client, text), lambda path: status.set(f"Saved: {path}"))

    for label, command in [("Activate", activate), ("Check now", lambda: run(client.validate, show_result)),
                           ("Export CSV", export), ("Release this machine", lambda: run(client.deactivate, show_released))]:
        button = ttk.Button(actions, text=label, command=command)
        button.pack(side="top", fill="x", pady=3)
        buttons.append(button)

    def finish_close(_):
        executor.shutdown(wait=False)
        root.destroy()

    def close():
        nonlocal closing
        if closing:
            return
        closing = True
        status.set("Closing and releasing any shared seat…")
        for button in buttons:
            button.state(["disabled"])
        # Queued after any in-progress action. close() releases a floating seat, not the machine.
        pending.append((executor.submit(client.close), finish_close))

    def poll():
        if pending and pending[0][0].done():
            future, done = pending.pop(0)
            try:
                done(future.result())
            except AccessDenied as error:
                status.set(str(error))
            except Exception:
                status.set("The action could not be completed. Check your connection and try again.")
                if closing:
                    finish_close(None)
                    return
            if not closing:
                for button in buttons:
                    button.state(["!disabled"])
            elif not pending:
                return
        root.after(75, poll)

    root.protocol("WM_DELETE_WINDOW", close)
    root.after(75, poll)
    run(client.check, show_result)
    root.mainloop()


if __name__ == "__main__":
    main()