파일 작업 공간

import os
import threading

import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")


def login(email, password):
    response = requests.post(f"{BASE_URL}/api/auth/login",
                             json={"email": email, "password": password},
                             timeout=30)
    response.raise_for_status()

    # the login response nests the tokens under data.user
    user = response.json()["data"]["user"]
    return user["accessToken"], user.get("refreshToken")


def refresh(refresh_token):
    response = requests.post(f"{BASE_URL}/api/auth/token/refresh",
                             headers={"X-Refresh-Token": refresh_token},
                             timeout=30)
    response.raise_for_status()

    # the refresh response nests the tokens directly under data
    data = response.json()["data"]
    return data["accessToken"], data.get("refreshToken")

class Session:
    def __init__(self, email, password):
        self._lock = threading.Lock()
        self._access, self._refresh = login(email, password)

    @property
    def access_token(self):
        with self._lock:
            return self._access

    def renew(self):
        with self._lock:
            self._access, self._refresh = refresh(self._refresh)
            return self._access

class Client:
    def __init__(self, session, workspace_id=None):
        self.session = session
        self.workspace_id = workspace_id

    def request(self, method, path, body=None, params=None, retried=False):
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.session.access_token}",
        }

        # when omitted the account's current workspace is used
        if self.workspace_id:
            headers["x-workspace-id"] = self.workspace_id

        response = requests.request(method, BASE_URL + path, headers=headers,
                                    json=body, params=params, timeout=30)

        if response.status_code == 401 and not retried:
            self.session.renew()
            return self.request(method, path, body, params, retried=True)

        return response

me = api("GET", "/api/auth/me") or {}
print(me)

result = api("GET", "/api/devices", params={"page": 1, "size": 20}) or {}

if result.get("viewScope"):
    print("view scope:", result["viewScope"])

for device in result.get("devices") or []:
    print(device["deviceId"], device["name"])

class WorkspaceForbidden(Exception):
    pass


def call(client, method, path, body=None, params=None):
    response = client.request(method, path, body, params)

    if response.status_code == 403:
        raise WorkspaceForbidden(path)

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


try:
    devices = call(client, "GET", "/api/devices")
except WorkspaceForbidden:
    devices, notice = [], "no permission to access this workspace"

clients = {
    workspace_id: Client(session, workspace_id)
    for workspace_id in WORKSPACE_IDS
}


def collect_devices():
    result = {}

    for workspace_id, client in clients.items():
        try:
            result[workspace_id] = call(client, "GET", "/api/devices")
        except WorkspaceForbidden:
            continue

    return result

def build_client(config):
    session = Session(config["email"], config["password"])
    return Client(session, config["workspace_id"])


clients = {name: build_client(cfg) for name, cfg in TENANTS.items()}

failed = []

for name, client in clients.items():
    try:
        run_for(client)
    except Exception as error:
        failed.append((name, error))

for name, error in failed:
    print(f"{name}: {error}")