✨ add support for workflows

Author: sebastianMindee

.github/workflows/test-regression.yml

@@ -62,3 +62,4 @@ jobs:
         notification_title: "Regression test workflow {workflow} is failing"
       env:
         SLACK_WEBHOOK_URL: ${{ secrets.PRODUCTION_ISSUES_SLACK_HOOK_URL }}
+        WORKFLOW_ID: ${{ secrets.WORKFLOW_ID_SE_TESTS }}

docs/extras/code_samples/workflow_execution.txt

@@ -0,0 +1,21 @@
+from mindee import Client, WorkflowResponse
+
+# Init a new client
+mindee_client = Client(api_key: "my-api-key")
+
+workflow_id = "workflow-id"
+
+# Load a file from disk
+input_doc = mindee_client.source_from_path("/path/to/the/file.ext")
+
+# Send the file to the workflow.
+result: WorkflowResponse = mindee_client.execute_workflow(
+    input_doc,
+    workflow_id,
+    # Optionally, add an alias and a priority to the workflow.
+    # alias="my-alias",
+    # priority=ExecutionPriority.LOW
+)
+
+# Print the ID of the execution to make sure it worked.
+print(result.execution.id)

mindee/__init__.py

@@ -4,3 +4,4 @@
 from mindee.parsing.common.async_predict_response import AsyncPredictResponse, Job
 from mindee.parsing.common.feedback_response import FeedbackResponse
 from mindee.parsing.common.predict_response import PredictResponse
+from mindee.parsing.common.workflow_response import WorkflowResponse

mindee/client.py

@@ -22,11 +22,18 @@
     is_valid_async_response,
     is_valid_sync_response,
 )
+from mindee.mindee_http.endpoint import CustomEndpoint, Endpoint
+from mindee.mindee_http.mindee_api import MindeeApi
+from mindee.mindee_http.workflow_endpoint import WorkflowEndpoint
+from mindee.mindee_http.workflow_settings import WorkflowSettings
 from mindee.parsing.common.async_predict_response import AsyncPredictResponse
+from mindee.parsing.common.execution_priority import ExecutionPriority
 from mindee.parsing.common.feedback_response import FeedbackResponse
 from mindee.parsing.common.inference import Inference
 from mindee.parsing.common.predict_response import PredictResponse
 from mindee.parsing.common.string_dict import StringDict
+from mindee.parsing.common.workflow_response import WorkflowResponse
+from mindee.product import GeneratedV1
 
 OTS_OWNER = "mindee"
 
@@ -230,6 +237,42 @@ def parse_queued(
 
         return self._get_queued_document(product_class, endpoint, queue_id)
 
+    def execute_workflow(
+        self,
+        input_source: Union[LocalInputSource, UrlInputSource],
+        workflow_id: str,
+        page_options: Optional[PageOptions] = None,
+        alias: Optional[str] = None,
+        priority: Optional[ExecutionPriority] = None,
+        full_text: bool = False,
+    ) -> WorkflowResponse:
+        """
+        Send the document to an asynchronous endpoint and return its ID in the queue.
+
+        :param input_source: The document/source file to use.
+            Has to be created beforehand.
+        :param workflow_id: ID of the workflow.
+        :param page_options: If set, remove pages from the document as specified. This is done before sending the file\
+ to the server. It is useful to avoid page limitations.
+        :param alias: Optional alias for the document.
+        :param priority: Optional priority for the document.
+        :param full_text: Whether to include the full OCR text response in compatible APIs.
+        :return:
+        """
+        if isinstance(input_source, LocalInputSource):
+            if page_options and input_source.is_pdf():
+                input_source.process_pdf(
+                    page_options.operation,
+                    page_options.on_min_pages,
+                    page_options.page_indexes,
+                )
+
+        logger.debug("Sending document to workflow: %s", workflow_id)
+
+        return self._send_to_workflow(
+            GeneratedV1, input_source, workflow_id, alias, priority, full_text
+        )
+
     def _validate_async_params(
         self, initial_delay_sec: float, delay_sec: float, max_retries: int
     ) -> None:
@@ -438,6 +481,50 @@ def _get_queued_document(
 
         return AsyncPredictResponse(product_class, queue_response.json())
 
+    def _send_to_workflow(
+        self,
+        product_class: Type[Inference],
+        input_source: Union[LocalInputSource, UrlInputSource],
+        workflow_id: str,
+        alias: Optional[str] = None,
+        priority: Optional[ExecutionPriority] = None,
+        full_text: bool = False,
+    ) -> WorkflowResponse:
+        """
+        Sends a document to a workflow.
+
+        :param product_class: The document class to use.
+            The response object will be instantiated based on this parameter.
+
+        :param input_source: The document/source file to use.
+            Has to be created beforehand.
+        :param workflow_id: ID of the workflow.
+        :param alias: Optional alias for the document.
+        :param priority: Priority for the document.
+        :param full_text: Whether to include the full OCR text response in compatible APIs.
+        :return:
+        """
+        if input_source is None:
+            raise MindeeClientError("No input document provided")
+
+        workflow_endpoint = WorkflowEndpoint(
+            WorkflowSettings(api_key=self.api_key, workflow_id=workflow_id)
+        )
+
+        response = workflow_endpoint.workflow_execution_post(
+            input_source, alias, priority, full_text
+        )
+
+        dict_response = response.json()
+
+        if not is_valid_async_response(response):
+            clean_response = clean_request_json(response)
+            raise handle_error(
+                str(product_class.endpoint_name),
+                clean_response,
+            )
+        return WorkflowResponse(product_class, dict_response)
+
     def _initialize_ots_endpoint(self, product_class: Type[Inference]) -> Endpoint:
         if product_class.__name__ == "CustomV1":
             raise MindeeClientError("Missing endpoint specifications for custom build.")

mindee/mindee_http/__init__.py

@@ -6,3 +6,5 @@
     is_valid_async_response,
     is_valid_sync_response,
 )
+from mindee.mindee_http.workflow_endpoint import WorkflowEndpoint
+from mindee.mindee_http.workflow_settings import WorkflowSettings

mindee/mindee_http/base_endpoint.py

@@ -1,12 +1,12 @@
 from abc import ABC
 
-from mindee.mindee_http.mindee_api import MindeeApi
+from mindee.mindee_http.base_settings import BaseSettings
 
 
 class BaseEndpoint(ABC):
     """Base endpoint class for the Mindee API."""
 
-    def __init__(self, settings: MindeeApi) -> None:
+    def __init__(self, settings: BaseSettings) -> None:
         """
         Base API endpoint class for all endpoints.
 

mindee/mindee_http/base_settings.py

@@ -0,0 +1,71 @@
+import os
+from dataclasses import dataclass
+from typing import Dict, Optional, Union
+
+from mindee.logger import logger
+from mindee.versions import __version__, get_platform, python_version
+
+API_KEY_ENV_NAME = "MINDEE_API_KEY"
+API_KEY_DEFAULT = ""
+
+BASE_URL_ENV_NAME = "MINDEE_BASE_URL"
+BASE_URL_DEFAULT = "https://api.mindee.net/v1"
+
+REQUEST_TIMEOUT_ENV_NAME = "MINDEE_REQUEST_TIMEOUT"
+TIMEOUT_DEFAULT = 120
+
+PLATFORM = get_platform()
+USER_AGENT = f"mindee-api-python@v{__version__} python-v{python_version} {PLATFORM}"
+
+
+@dataclass
+class BaseSettings:
+    """Settings class relating to API requests."""
+
+    api_key: Optional[str]
+    """API Key for the client."""
+    base_url: str
+    request_timeout: int
+
+    def __init__(self, api_key: Optional[str]):
+        self._set_api_key(api_key)
+        self.request_timeout = TIMEOUT_DEFAULT
+        self.set_base_url(BASE_URL_DEFAULT)
+        self.set_from_env()
+
+    @property
+    def base_headers(self) -> Dict[str, str]:
+        """Base headers to send with all API requests."""
+        return {
+            "Authorization": f"Token {self.api_key}",
+            "User-Agent": USER_AGENT,
+        }
+
+    def _set_api_key(self, api_key: Optional[str]) -> None:
+        """Set the endpoint's API key from an environment variable, if present."""
+        env_val = os.getenv(API_KEY_ENV_NAME, "")
+        if env_val and (not api_key or len(api_key) == 0):
+            logger.debug("API key set from environment")
+            self.api_key = env_val
+            return
+        self.api_key = api_key
+
+    def set_from_env(self) -> None:
+        """Set various parameters from environment variables, if present."""
+        env_vars = {
+            BASE_URL_ENV_NAME: self.set_base_url,
+            REQUEST_TIMEOUT_ENV_NAME: self.set_timeout,
+        }
+        for name, func in env_vars.items():
+            env_val = os.getenv(name, "")
+            if env_val:
+                func(env_val)
+                logger.debug("Value was set from env: %s", name)
+
+    def set_timeout(self, value: Union[str, int]) -> None:
+        """Set the timeout for all requests."""
+        self.request_timeout = int(value)
+
+    def set_base_url(self, value: str) -> None:
+        """Set the base URL for all requests."""
+        self.base_url = value

mindee/mindee_http/endpoint.py

@@ -13,6 +13,8 @@
 class Endpoint(BaseEndpoint):
     """Generic API endpoint for a product."""
 
+    settings: MindeeApi
+
     def __init__(
         self, url_name: str, owner: str, version: str, settings: MindeeApi
     ) -> None:

mindee/mindee_http/mindee_api.py

@@ -1,41 +1,22 @@
-import os
 from dataclasses import dataclass
-from typing import Dict, Optional, Union
+from typing import Optional
 
 from mindee.error.mindee_error import MindeeApiError
-from mindee.logger import logger
-from mindee.versions import __version__, get_platform, python_version
-
-API_KEY_ENV_NAME = "MINDEE_API_KEY"
-API_KEY_DEFAULT = ""
-
-BASE_URL_ENV_NAME = "MINDEE_BASE_URL"
-BASE_URL_DEFAULT = "https://api.mindee.net/v1"
-
-REQUEST_TIMEOUT_ENV_NAME = "MINDEE_REQUEST_TIMEOUT"
-TIMEOUT_DEFAULT = 120
-
-PLATFORM = get_platform()
-USER_AGENT = f"mindee-api-python@v{__version__} python-v{python_version} {PLATFORM}"
+from mindee.mindee_http.base_settings import API_KEY_ENV_NAME, BaseSettings
 
 
 @dataclass
-class MindeeApi:
+class MindeeApi(BaseSettings):
     """Settings class relating to API requests."""
 
-    api_key: Optional[str]
-    """API Key for the client."""
-    base_url: str
-    request_timeout: int
-
     def __init__(
         self,
         api_key: Optional[str],
         endpoint_name: str,
         account_name: str,
         version: str,
     ):
-        self._set_api_key(api_key)
+        super().__init__(api_key)
         if not self.api_key or len(self.api_key) == 0:
             raise MindeeApiError(
                 (
@@ -48,44 +29,4 @@ def __init__(
         self.endpoint_name = endpoint_name
         self.account_name = account_name
         self.version = version
-        self.request_timeout = TIMEOUT_DEFAULT
-        self.set_base_url(BASE_URL_DEFAULT)
-        self.set_from_env()
         self.url_root = f"{self.base_url}/products/{self.account_name}/{self.endpoint_name}/v{self.version}"
-
-    @property
-    def base_headers(self) -> Dict[str, str]:
-        """Base headers to send with all API requests."""
-        return {
-            "Authorization": f"Token {self.api_key}",
-            "User-Agent": USER_AGENT,
-        }
-
-    def _set_api_key(self, api_key: Optional[str]) -> None:
-        """Set the endpoint's API key from an environment variable, if present."""
-        env_val = os.getenv(API_KEY_ENV_NAME, "")
-        if env_val and (not api_key or len(api_key) == 0):
-            logger.debug("API key set from environment")
-            self.api_key = env_val
-            return
-        self.api_key = api_key
-
-    def set_from_env(self) -> None:
-        """Set various parameters from environment variables, if present."""
-        env_vars = {
-            BASE_URL_ENV_NAME: self.set_base_url,
-            REQUEST_TIMEOUT_ENV_NAME: self.set_timeout,
-        }
-        for name, func in env_vars.items():
-            env_val = os.getenv(name, "")
-            if env_val:
-                func(env_val)
-                logger.debug("Value was set from env: %s", name)
-
-    def set_timeout(self, value: Union[str, int]) -> None:
-        """Set the timeout for all requests."""
-        self.request_timeout = int(value)
-
-    def set_base_url(self, value: str) -> None:
-        """Set the base URL for all requests."""
-        self.base_url = value

mindee/mindee_http/response_validation.py

@@ -39,14 +39,15 @@ def is_valid_async_response(response: requests.Response) -> bool:
     ):
         return False
     # Async errors.
-    if "job" not in response_json:
+    if "job" not in response_json and "execution" not in response_json:
         return False
     if (
         "job" in response_json
         and "error" in response_json["job"]
         and response_json["job"]["error"]
     ):
         return False
+
     return True