feat: force option for cli auth + get process by package version

Author: radu-mocanu

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.0.50"
+version = "2.0.51"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.10"

src/uipath/_cli/_utils/_processes.py

@@ -8,28 +8,38 @@
 
 console = ConsoleLogger()
 client = httpx.Client(follow_redirects=True, timeout=30.0)
+odata_top_filter = 25
 
 
 def get_release_info(
-    base_url: str, token: str, package_name: str, folder_id: str
+    base_url: str,
+    token: str,
+    package_name: str,
+    package_version: str,
+    folder_id: str,
 ) -> None | tuple[Any, Any] | tuple[None, None]:
     headers = {
         "Authorization": f"Bearer {token}",
         "x-uipath-organizationunitid": str(folder_id),
     }
 
-    release_url = f"{base_url}/orchestrator_/odata/Releases/UiPath.Server.Configuration.OData.ListReleases?$select=Id,Key&$top=1&$filter=Name%20eq%20%27{urllib.parse.quote(package_name)}%27"
+    release_url = f"{base_url}/orchestrator_/odata/Releases/UiPath.Server.Configuration.OData.ListReleases?$select=Id,Key,ProcessVersion&$top={odata_top_filter}&$filter=ProcessKey%20eq%20%27{urllib.parse.quote(package_name)}%27"
     response = client.get(release_url, headers=headers)
     if response.status_code == 200:
         try:
             data = json.loads(response.text)
-            release_id = data["value"][0]["Id"]
-            release_key = data["value"][0]["Key"]
+            process = next(
+                process
+                for process in data["value"]
+                if process["ProcessVersion"] == package_version
+            )
+            release_id = process["Id"]
+            release_key = process["Key"]
             return release_id, release_key
         except KeyError:
             console.warning("Warning: Failed to deserialize release data")
             return None, None
-        except IndexError:
+        except StopIteration:
             console.error(
                 f"Error: No process with name '{package_name}' found in your workspace. Please publish the process first."
             )

src/uipath/_cli/cli_auth.py

@@ -56,25 +56,34 @@ def set_port():
 
 @click.command()
 @environment_options
-def auth(domain="alpha"):
+@click.option(
+    "-f",
+    "--force",
+    is_flag=True,
+    required=False,
+    help="Force new token",
+)
+def auth(domain, force: None | bool = False):
     """Authenticate with UiPath Cloud Platform."""
     with console.spinner("Authenticating with UiPath ..."):
         portal_service = PortalService(domain)
-        if (
-            os.getenv("UIPATH_URL")
-            and os.getenv("UIPATH_TENANT_ID")
-            and os.getenv("UIPATH_ORGANIZATION_ID")
-        ):
-            try:
-                portal_service.ensure_valid_token()
-                console.success(
-                    "Authentication successful.",
-                )
-                return
-            except Exception:
-                console.info(
-                    "Authentication token is invalid. Please reauthenticate.",
-                )
+
+        if not force:
+            if (
+                os.getenv("UIPATH_URL")
+                and os.getenv("UIPATH_TENANT_ID")
+                and os.getenv("UIPATH_ORGANIZATION_ID")
+            ):
+                try:
+                    portal_service.ensure_valid_token()
+                    console.success(
+                        "Authentication successful.",
+                    )
+                    return
+                except Exception:
+                    console.info(
+                        "Authentication token is invalid. Please reauthenticate.",
+                    )
 
         auth_url, code_verifier, state = get_auth_url(domain)
 

src/uipath/_cli/cli_invoke.py

@@ -24,7 +24,7 @@
 client = httpx.Client(follow_redirects=True, timeout=30.0)
 
 
-def _read_project_name() -> str:
+def _read_project_details() -> [str, str]:
     current_path = os.getcwd()
     toml_path = os.path.join(current_path, "pyproject.toml")
     if not os.path.isfile(toml_path):
@@ -37,7 +37,7 @@ def _read_project_name() -> str:
         if "name" not in content["project"]:
             console.error("pyproject.toml is missing the required field: project.name.")
 
-        return content["project"]["name"]
+        return content["project"]["name"], content["project"]["version"]
 
 
 @click.command()
@@ -67,15 +67,15 @@ def invoke(
 
         url = f"{base_url}/orchestrator_/odata/Jobs/UiPath.Server.Configuration.OData.StartJobs"
         _, personal_workspace_folder_id = get_personal_workspace_info(base_url, token)
-        project_name = _read_project_name()
+        project_name, project_version = _read_project_details()
 
         if not personal_workspace_folder_id:
             console.error(
                 "No personal workspace found for user. Please try reauthenticating."
             )
 
         _, release_key = get_release_info(
-            base_url, token, project_name, personal_workspace_folder_id
+            base_url, token, project_name, project_version, personal_workspace_folder_id
         )
         payload = {
             "StartInfo": {

src/uipath/_cli/cli_publish.py

@@ -129,15 +129,22 @@ def publish(feed):
         console.success("Package published successfully!")
 
         if is_personal_workspace:
+            package_name = None
+            package_version = None
             try:
-                data = json.loads(response.text)
-                package_name = json.loads(data["value"][0]["Body"])["Id"]
+                data = json.loads(response.text)["value"][0]["Body"]
+                package_name = json.loads(data)["Id"]
+                package_version = json.loads(data)["Version"]
             except json.decoder.JSONDecodeError:
                 console.warning("Failed to deserialize package name")
             if package_name is not None:
                 with console.spinner("Getting process information ..."):
                     release_id, _ = get_release_info(
-                        base_url, token, package_name, personal_workspace_feed_id
+                        base_url,
+                        token,
+                        package_name,
+                        package_version,
+                        personal_workspace_feed_id,
                     )
                 if release_id:
                     process_url = f"{base_url}/orchestrator_/processes/{release_id}/edit?fid={personal_workspace_folder_id}"