Initial logic for Model Derivative client.

Author: petrbroz

src/autodesk_forge_sdk/__init__.py

@@ -1,2 +1,3 @@
 from .auth import AuthenticationClient, PassiveTokenProvider, ActiveTokenProvider
-from .oss import OSSClient
+from .oss import OSSClient
+from .md import ModelDerivativeClient, urnify

src/autodesk_forge_sdk/md.py

@@ -0,0 +1,65 @@
+import base64
+from urllib.parse import quote
+from .auth import BaseOAuthClient
+
+BASE_URL = 'https://developer.api.autodesk.com/modelderivative/v2'
+WRITE_SCOPES = ['data:create', 'data:write', 'data:read']
+
+def urnify(str):
+    """Converts input string into base64-encoded string (without padding '=' characters)
+    that can be used as Model Derivative service URN.
+    """
+    base64_bytes = base64.b64encode(str.encode('ascii'))
+    ascii_output = base64_bytes.decode('ascii')
+    return ascii_output.rstrip('=')
+
+class ModelDerivativeClient(BaseOAuthClient):
+    def __init__(self, token_provider, base_url=BASE_URL):
+        BaseOAuthClient.__init__(self, token_provider, base_url)
+
+    def get_formats(self):
+        """Returns an up-to-date list of Forge-supported translations, that you can use to identify
+        which types of derivatives are supported for each source file type.
+
+        Returns:
+        object: Parsed response object. For more information, see https://forge.autodesk.com/en/docs/model-derivative/v2/reference/http/formats-GET/#body-structure-200.
+        """
+        return self._get('/designdata/formats', []).json()
+
+    def submit_job(self, urn, output_formats, output_region, root_filename=None, workflow_id=None, workflow_attr=None, force=False):
+        """Translate a design from one format to another format.
+
+        Parameters:
+        urn (string): Base64-encoded ID of the object to translate.
+        output_formats (list): List of objects representing all the requested outputs. Each object should have at least 'type' property set to 'svf', 'svf2', etc.
+        output_region (string): Output region. Allowed values: 'US', 'EMEA'.
+        root_filename (string): Optional starting filename if the converted file is a ZIP archive.
+        workflow_id (string): Optional workflow ID.
+        workflow_attr (object): Optional workflow payload.
+
+        Returns:
+        object: Parsed response object. For more information, see https://forge.autodesk.com/en/docs/model-derivative/v2/reference/http/job-POST/#body-structure-200-201.
+        """
+        json = {
+            'input': {
+                'urn': urn
+            },
+            'output': {
+                'formats': output_formats,
+                'destination': {
+                    'region': output_region
+                }
+            }
+        }
+        if root_filename:
+            json['input']['compressedUrn'] = True
+            json['input']['rootFilename'] = root_filename
+        if workflow_id:
+            json['misc'] = { 'workflowId': workflow_id }
+            if workflow_attr:
+                json['misc']['workflowAttribute'] = workflow_attr
+        headers = {}
+        if force:
+            headers['x-ads-force'] = 'true'
+        # TODO: what about the EMEA endpoint?
+        return self._post('/designdata/job', WRITE_SCOPES, json=json, headers=headers).json()

tests/context.py

@@ -2,7 +2,7 @@
 import sys
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'src')))
-from autodesk_forge_sdk import AuthenticationClient, OSSClient, ActiveTokenProvider, PassiveTokenProvider
+from autodesk_forge_sdk import AuthenticationClient, OSSClient, ModelDerivativeClient, urnify, ActiveTokenProvider, PassiveTokenProvider
 
 FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
 FORGE_CLIENT_SECRET = os.environ["FORGE_CLIENT_SECRET"]

tests/test_md.py

@@ -0,0 +1,19 @@
+import unittest
+from .context import ModelDerivativeClient, urnify, ActiveTokenProvider, FORGE_CLIENT_ID, FORGE_CLIENT_SECRET, FORGE_BUCKET
+
+class ModelDerivativeClientTestSuite(unittest.TestCase):
+    """Forge Model Derivative client test cases."""
+
+    def setUp(self):
+        self.client = ModelDerivativeClient(ActiveTokenProvider(FORGE_CLIENT_ID, FORGE_CLIENT_SECRET))
+
+    def test_get_formats(self):
+        formats = self.client.get_formats()
+        assert formats
+
+    def test_urnify(self):
+        urn = urnify('test')
+        assert urn == 'dGVzdA'
+
+if __name__ == "__main__":
+    unittest.main()