anvilco · aalmazan · Aug 5, 2022 · Aug 3, 2022 · Aug 3, 2022 · Aug 4, 2022
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# 1.5.0 (2022-08-05)
+
+- Added support for `ForgeSubmit` mutation.
+
 # 1.4.1 (2022-05-11)
 
 - Updated `mkdocs` dependency to fix issue with Read the Docs.

diff --git a/docs/api_usage.md b/docs/api_usage.md
@@ -150,6 +150,15 @@ for more details the creation process.
   `CreateEtchPacket` and `CreateEtchPacketPayload`.
 * `json` - Raw JSON payload of the etch packet
 
+### Anvil.forge_submit
+
+Creates an Anvil submission
+object. [See documentation](https://www.useanvil.com/docs/api/graphql/reference/#operation-forgesubmit-Mutations) for
+more details.
+
+* `payload` - Payload to use for the submission. Accepted types are `dict`,
+  `ForgeSubmit` and `ForgeSubmitPayload`.
+* `json` - Raw JSON payload of the `forgeSubmit` mutation.
 
 ### Data Types
 

diff --git a/examples/create_workflow_submission.py b/examples/create_workflow_submission.py
@@ -0,0 +1,54 @@
+from datetime import datetime
+
+from python_anvil.api import Anvil
+from python_anvil.api_resources.payload import ForgeSubmitPayload
+
+
+API_KEY = 'my-api-key'
+
+
+def main():
+    # Use https://app.useanvil.com/org/YOUR_ORG_HERE/w/WORKFLOW_NAME/api
+    # to get a detailed list and description of which fields, eids, etc.
+    # are available to use.
+    forge_eid = ""
+
+    anvil = Anvil(api_key=API_KEY)
+
+    # Create a payload with the payload model.
+    payload = ForgeSubmitPayload(
+        forge_eid=forge_eid, payload=dict(field1="Initial forgeSubmit")
+    )
+
+    res = anvil.forge_submit(payload=payload)
+
+    data = res["data"]["forgeSubmit"]
+
+    print(data)
+
+    # Get submission and weld_data eids from the initial response
+    submission_eid = data["eid"]
+    weld_data_eid = data["weldData"]["eid"]
+
+    payload = ForgeSubmitPayload(
+        forge_eid=forge_eid,
+        # If submission and weld_data eids are provided, you will be _editing_
+        # an existing submission.
+        submission_eid=submission_eid,
+        weld_data_eid=weld_data_eid,
+        # NOTE: If using a development key, this will `is_test` will always
+        # be `True` even if it's set as `False` here.
+        is_test=False,
+        payload=dict(
+            field1=f"Edited this field {datetime.now()}",
+        ),
+    )
+
+    res = anvil.forge_submit(payload=payload)
+
+    data = res["data"]["forgeSubmit"]
+    print(data)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/pyproject.toml b/pyproject.toml
@@ -1,7 +1,7 @@
 [tool.poetry]
 
 name = "python_anvil"
-version = "1.4.1"
+version = "1.5.0"
 description = "Anvil API"
 
 license = "MIT"

diff --git a/python_anvil/api.py b/python_anvil/api.py
@@ -1,10 +1,11 @@
 from logging import getLogger
-from typing import AnyStr, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, AnyStr, Callable, Dict, List, Optional, Text, Tuple, Union
 
 from .api_resources.mutations import *
 from .api_resources.payload import (
     CreateEtchPacketPayload,
     FillPDFPayload,
+    ForgeSubmitPayload,
     GeneratePDFPayload,
 )
 from .api_resources.requests import GraphqlRequest, PlainRequest, RestRequest
@@ -37,7 +38,12 @@ class Anvil:
     def __init__(self, api_key=None, environment='dev'):
         self.client = HTTPClient(api_key=api_key, environment=environment)
 
-    def query(self, query: str, variables: Optional[str] = None, **kwargs):
+    def query(
+        self,
+        query: str,
+        variables: Union[Optional[Text], Dict[Text, Any]] = None,
+        **kwargs,
+    ):
         gql = GraphqlRequest(client=self.client)
         return gql.post(query, variables=variables, **kwargs)
 
@@ -187,6 +193,10 @@ def get_welds(self, **kwargs) -> Union[List, Tuple[List, Dict]]:
                     eid
                     slug
                     title
+                    forges {
+                      eid
+                      name
+                    }
                   }
                 }
               }
@@ -200,6 +210,38 @@ def get_data(r):
 
         return _get_return(res, get_data=get_data)
 
+    def get_weld(self, eid: Text, **kwargs):
+        res = self.query(
+            """
+            query WeldQuery(
+                #$organizationSlug: String!,
+                #$slug: String!
+                $eid: String!
+            ) {
+                weld(
+                    #organizationSlug: $organizationSlug,
+                    #slug: $slug
+                    eid: $eid
+                ) {
+                    eid
+                    slug
+                    name
+                    forges {
+                        eid
+                        name
+                        slug
+                    }
+                }
+            }""",
+            variables=dict(eid=eid),
+            **kwargs,
+        )
+
+        def get_data(r):
+            return r["data"]["weld"]
+
+        return _get_return(res, get_data=get_data)
+
     def create_etch_packet(
         self,
         payload: Optional[
@@ -253,3 +295,33 @@ def download_documents(self, document_group_eid: str, **kwargs):
         """Retrieve all completed documents in zip form."""
         api = PlainRequest(client=self.client)
         return api.get(f"document-group/{document_group_eid}.zip", **kwargs)
+
+    def forge_submit(
+        self,
+        payload: Optional[Union[Dict[Text, Any], ForgeSubmitPayload]] = None,
+        json=None,
+        **kwargs,
+    ):
+        """Create a Webform (forge) submission via a graphql mutation."""
+        if not any([json, payload]):
+            raise TypeError('One of arguments `json` or `payload` are required')
+
+        if json:
+            payload = ForgeSubmitPayload.parse_raw(
+                json, content_type="application/json"
+            )
+
+        if isinstance(payload, dict):
+            mutation = ForgeSubmit.create_from_dict(payload)
+        elif isinstance(payload, ForgeSubmitPayload):
+            mutation = ForgeSubmit(payload=payload)
+        else:
+            raise ValueError(
+                "`payload` must be a valid ForgeSubmitPayload instance or dict"
+            )
+
+        return self.mutate(
+            mutation,
+            variables=mutation.create_payload().dict(by_alias=True, exclude_none=True),
+            **kwargs,
+        )
diff --git a/python_anvil/api_resources/mutations/__init__.py b/python_anvil/api_resources/mutations/__init__.py
@@ -1,3 +1,4 @@
 from .base import BaseQuery
 from .create_etch_packet import CreateEtchPacket
+from .forge_submit import ForgeSubmit
 from .generate_etch_signing_url import GenerateEtchSigningURL
diff --git a/python_anvil/api_resources/mutations/create_etch_packet.py b/python_anvil/api_resources/mutations/create_etch_packet.py
@@ -120,7 +120,7 @@ def __init__(
         self.merge_pdfs = merge_pdfs
 
     @classmethod
-    def create_from_dict(cls, payload: dict):
+    def create_from_dict(cls, payload: Dict) -> 'CreateEtchPacket':
         """Create a new instance of `CreateEtchPacket` from a dict payload."""
         try:
             mutation = cls(

diff --git a/python_anvil/api_resources/mutations/forge_submit.py b/python_anvil/api_resources/mutations/forge_submit.py
@@ -0,0 +1,129 @@
+from typing import Any, Dict, Optional, Text, Union
+
+from python_anvil.api_resources.mutations.base import BaseQuery
+from python_anvil.api_resources.mutations.helpers import get_payload_attrs
+from python_anvil.api_resources.payload import ForgeSubmitPayload
+
+
+DEFAULT_RESPONSE_QUERY = """
+{
+  id
+  eid
+  status
+  resolvedPayload
+  currentStep
+  completedAt
+  createdAt
+  updatedAt
+  signer {
+    name
+    email
+    status
+    routingOrder
+  }
+  weldData {
+    id
+    eid
+    status
+    isTest
+    isComplete
+    agents
+  }
+}
+"""
+
+# NOTE: Since the below will be used as a formatted string (this also applies
+#   to f-strings) any literal curly braces need to be doubled, else they'll be
+#   interpreted as string replacement tokens.
+FORGE_SUBMIT = """
+mutation ForgeSubmit(
+    $forgeEid: String!,
+    $weldDataEid: String,
+    $submissionEid: String,
+    $payload: JSON!,
+    $currentStep: Int,
+    $complete: Boolean,
+    $isTest: Boolean,
+    $timezone: String,
+    $groupArrayId: String,
+    $groupArrayIndex: Int,
+    $errorType: String,
+) {{
+    forgeSubmit (
+        forgeEid: $forgeEid,
+        weldDataEid: $weldDataEid,
+        submissionEid: $submissionEid,
+        payload: $payload,
+        currentStep: $currentStep,
+        complete: $complete,
+        isTest: $isTest,
+        timezone: $timezone,
+        groupArrayId: $groupArrayId,
+        groupArrayIndex: $groupArrayIndex,
+        errorType: $errorType
+    ) {query}
+}}
+"""
+
+
+class ForgeSubmit(BaseQuery):
+    mutation = FORGE_SUBMIT
+    mutation_res_query = DEFAULT_RESPONSE_QUERY
+
+    def __init__(
+        self,
+        payload: Union[Dict[Text, Any], ForgeSubmitPayload],
+        forge_eid: Optional[Text] = None,
+        weld_data_eid: Optional[Text] = None,
+        submission_eid: Optional[Text] = None,
+        is_test: Optional[bool] = None,
+        **kwargs,
+    ):
+        """
+        :param forge_eid:
+        :param payload:
+        :param weld_data_eid:
+        :param submission_eid:
+        :param is_test:
+        :param kwargs: kwargs may contain other fields defined in
+              `ForgeSubmitPayload` if not explicitly in the `__init__` args.
+        """
+        if not forge_eid and not isinstance(payload, ForgeSubmitPayload):
+            raise ValueError(
+                "`forge_eid` is required if `payload` is not a "
+                "`ForgeSubmitPayload` instance"
+            )
+
+        self.payload = payload
+        self.forge_eid = forge_eid
+        self.weld_data_eid = weld_data_eid
+        self.submission_eid = submission_eid
+        self.is_test = is_test
+
+        # Get other attrs from the model and set on the instance
+        model_attrs = get_payload_attrs(ForgeSubmitPayload)
+        for attr in model_attrs:
+            if attr in kwargs:
+                setattr(self, attr, kwargs[attr])
+
+    @classmethod
+    def create_from_dict(cls, payload: Dict[Text, Any]):
+        # Parse the data through the model class to validate and pass it back
+        # as variables in this class.
+        return cls(**payload)
+
+    def create_payload(self):
+        # If provided a payload and no forge_eid, we'll assume that it's the
+        # full thing. Return that instead.
+        if not self.forge_eid and self.payload:
+            return self.payload
+
+        model_attrs = get_payload_attrs(ForgeSubmitPayload)
+
+        for_payload = {}
+        for attr in model_attrs:
+            obj = getattr(self, attr, None)
+            if obj is not None:
+                for_payload[attr] = obj
+
+        return ForgeSubmitPayload(**for_payload)
diff --git a/python_anvil/api_resources/mutations/helpers.py b/python_anvil/api_resources/mutations/helpers.py
@@ -0,0 +1,7 @@
+from typing import List, Text, Type
+
+from python_anvil.api_resources.base import BaseModel
+
+
+def get_payload_attrs(payload_model: Type[BaseModel]) -> List[Text]:
+    return list(payload_model.__fields__.keys())