Add defaults for demo purposes

Author: whoisarpit

patchwork/steps/ManageEngine/ManageEngineStep.py

@@ -5,13 +5,18 @@
 )
 from patchwork.common.tools.api_tool import APIRequestTool
 from patchwork.step import Step
-from patchwork.steps.ManageEngine.typed import ManageEngineInputs, ManageEngineOutputs
+
+from .demo import get_demo_defaults
+from .typed import ManageEngineInputs, ManageEngineOutputs
 
 
 class ManageEngineStep(
     Step, input_class=ManageEngineInputs, output_class=ManageEngineOutputs
 ):
     def __init__(self, inputs: ManageEngineInputs):
+        # ! TODO: Remove this after demo
+        inputs["access_token"] = get_demo_defaults().get("access_token")
+
         super().__init__(inputs)
 
         if not inputs.get("access_token"):

patchwork/steps/ManageEngine/demo.py

@@ -0,0 +1,114 @@
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Optional
+
+import requests
+import yaml
+
+from patchwork.logger import logger
+
+
+demo_defaults: Optional[dict] = None
+
+
+def get_demo_defaults() -> dict:
+    global demo_defaults
+    if demo_defaults is None:
+        demo_defaults = _load_demo_defaults()
+
+    if not demo_defaults.get("access_token"):
+        logger.debug("No access token found, getting a new one")
+        get_access_token()
+    elif (
+        demo_defaults.get("expires_at")
+        and datetime.now() + timedelta(minutes=5) > demo_defaults["expires_at"]
+    ):
+        logger.debug("Token expired, refreshing")
+        refresh_access_token()
+
+    return demo_defaults
+
+
+def _load_demo_defaults() -> dict:
+    """Loads demo defaults from the YAML file."""
+    try:
+        with open(Path(__file__).parent / "demo_defaults.yml", "r") as f:
+            defaults = yaml.safe_load(f)
+            # Convert expires_at to datetime object if it exists and is a string
+            if isinstance(defaults.get("expires_at"), str):
+                defaults["expires_at"] = datetime.fromisoformat(defaults["expires_at"])
+            return defaults
+    except FileNotFoundError:
+        logger.warning("demo_defaults.yml not found.  Returning empty dict.")
+        return {}
+
+
+def save_demo_defaults():
+    global demo_defaults
+    _save_demo_defaults(demo_defaults)
+
+
+def _save_demo_defaults(data: dict):
+    """Saves demo defaults to the YAML file."""
+    with open(Path(__file__).parent / "demo_defaults.yml", "w") as f:
+        yaml.dump(data, f)
+
+
+def get_access_token():
+    global demo_defaults
+    if not demo_defaults or not demo_defaults.get("grant_token"):
+        raise ValueError("grant_token is not set")
+
+    url = "https://accounts.zoho.com/oauth/v2/token"
+    data = {
+        "grant_type": "authorization_code",
+        "code": demo_defaults["grant_token"],
+        "client_id": demo_defaults["client_id"],
+        "client_secret": demo_defaults["client_secret"],
+        "redirect_uri": "https://www.zoho.com",
+    }
+    try:
+        response = requests.post(url, data=data)
+        response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
+        res_data = response.json()
+        demo_defaults["access_token"] = res_data["access_token"]
+        demo_defaults["expires_at"] = datetime.now() + timedelta(
+            seconds=int(res_data["expires_in"])
+        )
+        demo_defaults["refresh_token"] = res_data["refresh_token"]
+        demo_defaults["grant_token"] = None
+        save_demo_defaults()
+        logger.debug("Access token retrieved and demo_defaults updated")
+        return True
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error getting access token: {e}")
+        return False
+
+
+def refresh_access_token():
+    global demo_defaults
+    if not demo_defaults:
+        raise ValueError("demo_defaults is not initialized")
+
+    url = "https://accounts.zoho.com/oauth/v2/token"
+    data = {
+        "grant_type": "refresh_token",
+        "refresh_token": demo_defaults["refresh_token"],
+        "client_id": demo_defaults["client_id"],
+        "client_secret": demo_defaults["client_secret"],
+        "redirect_uri": "https://www.zoho.com",
+    }
+    try:
+        response = requests.post(url, data=data)
+        response.raise_for_status()
+        res_data = response.json()
+        demo_defaults["access_token"] = res_data["access_token"]
+        demo_defaults["expires_at"] = datetime.now() + timedelta(
+            seconds=int(res_data["expires_in"])
+        )
+        save_demo_defaults()
+        logger.debug("Access token refreshed and demo_defaults updated")
+        return True
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error refreshing access token: {e}")
+        return False

patchwork/steps/ManageEngine/demo_defaults.yml

@@ -0,0 +1,6 @@
+access_token: 1000.aea25f21668f7f25941b2ad7c5f6b662.1ba06be53aaf2dc1a8ddf23569b51d3b
+client_id: 1000.BZOK84OAU6LL168UUUXC19D9D1TYWP
+client_secret: 0bd251ab758c074a4016b49211c734cbad2d205cf0
+expires_at: 2025-03-05 18:09:43.659864
+grant_token: null
+refresh_token: 1000.29d59726eae5b800e14709baa6d90242.6066775c3a262ad3a42cc3ff9e3609a2