Updated version to 0.2.01.

Added workflows_list() and workflows_list_pd().

nacwrap/data_model.py

@@ -39,6 +39,7 @@ def _missing_(cls, value):
                 return member
         return None
 
+
 class ResolveType(Enum):
     RETRY = "1"
     FAIL = "2"
@@ -61,12 +62,16 @@ class Action(BaseModel):
         type: str
         parentId: Optional[str] = Field(default=None)
         startDateTime: Optional[datetime] = Field(default=None)
+        endDateTime: Optional[datetime] = Field(default=None)
         errorMessage: Optional[str] = Field(default=None)
         logMessage: Optional[str] = Field(default=None)
 
         @property
         def age(self) -> timedelta:
-            return datetime.now(timezone.utc) - self.startDateTime
+            if self.startDateTime:
+                return datetime.now(timezone.utc) - self.startDateTime
+            else:
+                return timedelta(days=0)
 
     # NintexInstance attributes
     instanceId: str
@@ -77,6 +82,7 @@ def age(self) -> timedelta:
     workflow: Workflow
     actions: List[Action]
 
+
 class NintexInstance(BaseModel):
     class Workflow(BaseModel):
         id: str
@@ -162,5 +168,17 @@ def name(self):
         return self.firstName + " " + self.lastName
 
 
+class NintexWorkflows(BaseModel):
+    """Response Data Model for Nintex Workflows API Endpoints."""
+
+    class Workflow(BaseModel):
+        id: str
+        name: str
+        lastModified: datetime
+
+    # NintexWorkflows Attributes
+    workflows: List[Workflow]
+
+
 class InstanceStartData(BaseModel):
     pass

nacwrap/exceptions.py

@@ -0,0 +1,2 @@
+class WorkflowApiError(Exception):
+    pass

nacwrap/nacwrap.py

@@ -6,3 +6,5 @@
 from .tasks import *
 from .users import *
 from .data_model import *
+from .workflows import *
+from .exceptions import *

nacwrap/workflows.py

@@ -2,7 +2,76 @@
 This module contains functions to interact with workflows
 """
 
+import json
+import os
+from datetime import datetime
+from typing import Literal, Optional, Union
+
+import requests
+
+from nacwrap._auth import Decorators
+from nacwrap._helpers import _fetch_page, _post
+from nacwrap.data_model import *
+from nacwrap.exceptions import *
+
+
 # TODO List workflows
+@Decorators.refresh_token
+def workflows_list(limit: int = 1000) -> dict:
+    """Get Nintex Workflow information.
+
+    Args:
+        limit (int, optional): Max number of workflows to return. Defaults to 1000.
+
+    Raises:
+        WorkflowApiError: Generic error when API call fails.
+
+    Returns:
+        list[dict]: _description_
+    """
+
+    base_url = (
+        os.environ["NINTEX_BASE_URL"] + f"/workflows/v1/designs/published?limit={limit}"
+    )
+    results = []
+    url = base_url
+
+    while url:
+
+        try:
+            response = _fetch_page(
+                url,
+                headers={
+                    "Authorization": "Bearer " + os.environ["NTX_BEARER_TOKEN"],
+                    "Content-Type": "application/json",
+                },
+            )
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            raise WorkflowApiError(
+                f"Error, could not get workflow data: {e.response.status_code} - {e.response.content}"
+            )
+        except requests.exceptions.RequestException as e:
+            raise WorkflowApiError(f"Error, could not get workflow data: {e}")
+
+        data = response.json()
+        results += data["workflows"]
+        url = data.get("nextLink")
+
+    return results
+
+
+def workflows_list_pd(limit: int = 1000) -> NintexWorkflows:
+    """Get Nintex Workflow information.
+    Returns data as a pydantic data model.
+
+    Returns:
+        NintexWorkflows: Pydantic data model of API response.
+    """
+
+    workflows = workflows_list(limit=limit)
+    return NintexWorkflows(workflows=workflows)
+
 
 # TODO List Workflow permissions
 

setup.py

@@ -8,7 +8,7 @@
 
 setup(
     name="nacwrap",
-    version="0.2.0",
+    version="0.2.01",
     description="Python package that acts as a wrapper for the Nintex Automation Cloud system.",
     long_description=long_description,
     long_description_content_type="text/markdown",

tests/workflow_test.py

@@ -0,0 +1,41 @@
+import os
+import sys
+from datetime import datetime, timedelta
+from pprint import pprint
+from pydantic import BaseModel
+
+sys.path.insert(0, "")
+from nacwrap import *
+
+
+class PurchDocStartData(InstanceStartData):
+    requestor: str = Field(alias="se_txtrequestor")
+    requestor_email: str = Field(alias="se_txtrequestoremail")
+    purchaser_code: str = Field(alias="se_txtpurchasercode")
+
+
+def test_list_instances():
+    instances = nacwrap.instances_list(
+        workflow_name="Purchase Approval", status=WorkflowStatus.RUNNING
+    )
+    # pprint(instances)
+
+    instances = nacwrap.instances_list_pd(
+        workflow_name="Purchase Approval", status=WorkflowStatus.RUNNING
+    )
+
+    for i in instances:
+        print(f"Name: {i.instanceName}")
+
+
+def test_list_workflows():
+    # workflows = nacwrap.workflows_list()
+
+    # print(workflows)
+
+    workflows = workflows_list_pd()
+
+    print(workflows)
+
+
+test_list_workflows()