SDK Integration Best Practices

import os

key = (
    args.license_key
    or os.environ.get("AUTHFORGE_LICENSE_KEY")
    or load_from_config()
    or input("Enter license key: ")
)

import time

MAX_RETRIES = 3
for attempt in range(MAX_RETRIES):
    if client.login(license_key):
        break
    if attempt < MAX_RETRIES - 1:
        time.sleep(2 ** attempt)  # 1s, 2s, 4s
else:
    print("Could not connect to authentication server.")
    exit(1)

import sys
import threading
import time

from authforge import AuthForgeError

HEARTBEAT_INTERVAL = 900               # must match heartbeat_interval on the client
RETRY_WINDOW = 2 * HEARTBEAT_INTERVAL  # allow two more check-ins before giving up
SLACK = 120                            # covers SDK retries and request timeouts

license_lost = threading.Event()  # the main loop checks this, saves, and exits
lock = threading.Lock()
outage_started = None   # time.monotonic() of the first transient failure
last_failure = None     # time.monotonic() of the latest transient failure
watchdog_running = False

def on_failure(reason, exception):
    global outage_started, last_failure, watchdog_running
    if reason == "heartbeat_failed" and isinstance(exception, AuthForgeError) and exception.transient:
        # The SDK kept the session and checks in again on the next interval.
        now = time.monotonic()
        with lock:
            last_failure = now
            if outage_started is None:
                outage_started = now
                show_warning("License verification failed. "
                             "The app will close in 30 minutes unless connectivity is restored.")
            if not watchdog_running:
                watchdog_running = True
                threading.Thread(target=watch_retry_window, daemon=True).start()
        return
    # Definitive (the SDK already cleared the session and stopped check-ins) or login failure.
    force_shutdown()

def watch_retry_window():
    global outage_started, last_failure, watchdog_running
    while True:
        time.sleep(30)
        now = time.monotonic()
        with lock:
            if now - last_failure > HEARTBEAT_INTERVAL + SLACK:
                # A check-in has run since the last failure without reporting one.
                outage_started = last_failure = None
                watchdog_running = False
                hide_warning()
                return
            if now - outage_started < RETRY_WINDOW:
                continue
        force_shutdown()
        return

def force_shutdown():
    # Runs on a background thread, where sys.exit() would only end that thread:
    # signal the main loop instead of exiting here.
    license_lost.set()

# Main thread
while not license_lost.is_set():
    do_one_unit_of_work()  # keep units short so the loop notices the signal quickly

save_application_state()
show_message("License verification failed. Your work has been saved.")
client.logout()
sys.exit(1)

def on_failure(reason, exception):
    save_application_state()  # Always save first
    log_failure(reason, exception)  # Log for debugging
    show_user_message(reason)  # Inform the user
    # Then signal the main thread to exit after a delay or user acknowledgment

import os
import sys

LOCK_FILE = "/tmp/yourapp.lock"

def acquire_lock():
    if os.path.exists(LOCK_FILE):
        return False  # Another instance is running
    with open(LOCK_FILE, "w") as f:
        f.write(str(os.getpid()))
    return True

def release_lock():
    if os.path.exists(LOCK_FILE):
        os.remove(LOCK_FILE)

if not acquire_lock():
    # Secondary instance; skip auth, connect to primary via IPC
    connect_to_primary()
else:
    # Primary instance; authenticate
    if client.login(license_key):
        run_app()
    release_lock()

if client.login(license_key):
    min_version = client.app_variables.get("minVersion")
    if min_version and current_version < min_version:
        print(f"Please update to version {min_version} or later.")
        exit(1)