From e5f0e05fbd5922bb2f349f79b19264bfc516b3f9 Mon Sep 17 00:00:00 2001 From: Chris Lambert Date: Tue, 31 Dec 2024 10:22:35 -0600 Subject: [PATCH] Updated version to 0.2.01. Added workflows_list() and workflows_list_pd(). --- nacwrap/data_model.py | 20 +++++++++++- nacwrap/exceptions.py | 2 ++ nacwrap/nacwrap.py | 2 ++ nacwrap/workflows.py | 69 ++++++++++++++++++++++++++++++++++++++++++ setup.py | 2 +- tests/workflow_test.py | 41 +++++++++++++++++++++++++ 6 files changed, 134 insertions(+), 2 deletions(-) create mode 100644 nacwrap/exceptions.py create mode 100644 tests/workflow_test.py diff --git a/nacwrap/data_model.py b/nacwrap/data_model.py index 9b12c4e..616ecb9 100644 --- a/nacwrap/data_model.py +++ b/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 diff --git a/nacwrap/exceptions.py b/nacwrap/exceptions.py new file mode 100644 index 0000000..bc48d6a --- /dev/null +++ b/nacwrap/exceptions.py @@ -0,0 +1,2 @@ +class WorkflowApiError(Exception): + pass diff --git a/nacwrap/nacwrap.py b/nacwrap/nacwrap.py index 4f1fa35..aa7e005 100644 --- a/nacwrap/nacwrap.py +++ b/nacwrap/nacwrap.py @@ -6,3 +6,5 @@ from .tasks import * from .users import * from .data_model import * +from .workflows import * +from .exceptions import * diff --git a/nacwrap/workflows.py b/nacwrap/workflows.py index 4f8963b..ebd6400 100644 --- a/nacwrap/workflows.py +++ b/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 diff --git a/setup.py b/setup.py index ccdfab2..33e41fe 100644 --- a/setup.py +++ b/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", diff --git a/tests/workflow_test.py b/tests/workflow_test.py new file mode 100644 index 0000000..de95295 --- /dev/null +++ b/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()