Merge branch 'experiment/typings' into develop.

Author: petrbroz

NOTES.md

@@ -0,0 +1,4 @@
+# Ideas
+
+- use something like [colander](https://docs.pylonsproject.org/projects/colander/en/latest/basics.html) to deserialize JSON into Python classes (and potentially validate it, too)
+- how about _async_?

src/autodesk_forge_sdk/auth.py

@@ -48,7 +48,7 @@ class AuthenticationClient(BaseClient):
     **Documentation**: https://forge.autodesk.com/en/docs/oauth/v2/reference/http
     """
 
-    def __init__(self, base_url=BASE_URL):
+    def __init__(self, base_url: str=BASE_URL):
         """
         Create new instance of the client.
 
@@ -57,7 +57,7 @@ def __init__(self, base_url=BASE_URL):
         """
         BaseClient.__init__(self, base_url)
 
-    def authenticate(self, client_id, client_secret, scopes):
+    def authenticate(self, client_id: str, client_secret: str, scopes: list[Scope]) -> dict:
         """
         Generate a two-legged access token for specific set of scopes.
 
@@ -88,7 +88,7 @@ def authenticate(self, client_id, client_secret, scopes):
         }
         return self._post('/authenticate', form=form).json()
 
-    def get_authorization_url(self, client_id, response_type, redirect_uri, scopes, state=None):
+    def get_authorization_url(self, client_id: str, response_type: str, redirect_uri: str, scopes: list[Scope], state: str=None) -> str:
         """
         Generate a URL to redirect an end user to in order to acquire the user’s consent
         for your app to access the specified resources.
@@ -122,7 +122,7 @@ def get_authorization_url(self, client_id, response_type, redirect_uri, scopes,
             url += '&state={}'.format(quote(state))
         return url
 
-    def get_token(self, client_id, client_secret, code, redirect_uri):
+    def get_token(self, client_id: str, client_secret: str, code: str, redirect_uri: str) -> dict:
         """
         Exchange an authorization code extracted from `get_authorization_url` callback for a three-legged access token.
         This API will only be used when the 'Authorization Code' grant type is being adopted.
@@ -159,7 +159,7 @@ def get_token(self, client_id, client_secret, code, redirect_uri):
         }
         return self._post('/gettoken', form=form).json()
 
-    def refresh_token(self, client_id, client_secret, refresh_token, scopes):
+    def refresh_token(self, client_id: str, client_secret: str, refresh_token: str, scopes: list[Scope]) -> dict:
         """
         Acquire a new access token by using the refresh token provided by `get_token`.
 
@@ -193,7 +193,7 @@ def refresh_token(self, client_id, client_secret, refresh_token, scopes):
         }
         return self._post('/refreshtoken', form=form).json()
 
-    def get_user_profile(self, access_token):
+    def get_user_profile(self, access_token: str) -> dict:
         """
         Get the profile information of an authorizing end user in a three-legged context.
 
@@ -217,7 +217,7 @@ def get_user_profile(self, access_token):
         return self._get('/users/@me', headers=headers).json()
 
 class TokenProviderInterface:
-    def get_token(self, scopes):
+    def get_token(self, scopes: list[Scope]) -> str:
         """
         Generates access token for given set of scopes.
 
@@ -235,7 +235,7 @@ class SimpleTokenProvider(TokenProviderInterface):
     When using this approach, make sure that the hard-coded access token supports all the scopes that may be needed.
     """
 
-    def __init__(self, access_token):
+    def __init__(self, access_token: str):
         """
         Create new instance of the provider.
 
@@ -244,15 +244,15 @@ def __init__(self, access_token):
         """
         self.access_token = access_token
 
-    def get_token(self, scopes):
+    def get_token(self, scopes: list[Scope]) -> str:
         return self.access_token
 
 class OAuthTokenProvider(TokenProviderInterface):
     """
     Helper class that automatically generates (and caches) access tokens using specific app credentials.
     """
 
-    def __init__(self, client_id, client_secret):
+    def __init__(self, client_id: str, client_secret: str):
         """
         Create new instance of the provider.
 
@@ -265,7 +265,7 @@ def __init__(self, client_id, client_secret):
         self.auth_client = AuthenticationClient()
         self.cache = {}
 
-    def get_token(self, scopes):
+    def get_token(self, scopes: list[Scope]) -> str:
         cache_key = '+'.join(map(lambda s: s.value, scopes))
         now = datetime.now()
         if cache_key in self.cache:
@@ -277,7 +277,7 @@ def get_token(self, scopes):
         return auth
 
 class BaseOAuthClient(BaseClient):
-    def __init__(self, token_provider, base_url):
+    def __init__(self, token_provider: TokenProviderInterface, base_url: str):
         """
         Create new instance of the client.
 
@@ -288,31 +288,31 @@ def __init__(self, token_provider, base_url):
         BaseClient.__init__(self, base_url)
         self.token_provider = token_provider
 
-    def _get(self, url, scopes, params=None, headers=None):
+    def _get(self, url: str, scopes: list[Scope], params: dict=None, headers: dict=None):
         if not headers:
             headers = {}
         self._set_auth_headers(headers, scopes)
         return BaseClient._get(self, url, params, headers)
 
-    def _post(self, url, scopes, form=None, json=None, buff=None, params=None, headers=None):
+    def _post(self, url: str, scopes: list[Scope], form: dict=None, json: dict=None, buff=None, params: dict=None, headers: dict=None):
         if not headers:
             headers = {}
         self._set_auth_headers(headers, scopes)
         return BaseClient._post(self, url, form, json, buff, params, headers)
 
-    def _put(self, url, scopes, form=None, json=None, buff=None, params=None, headers=None):
+    def _put(self, url: str, scopes: list[Scope], form: dict=None, json: dict=None, buff=None, params: dict=None, headers: dict=None):
         if not headers:
             headers = {}
         self._set_auth_headers(headers, scopes)
         return BaseClient._put(self, url, form, json, buff, params, headers)
 
-    def _delete(self, url, scopes, params=None, headers=None):
+    def _delete(self, url: str, scopes: list[Scope], params: dict=None, headers: dict=None):
         if not headers:
             headers = {}
         self._set_auth_headers(headers, scopes)
         return BaseClient._delete(self, url, params, headers)
 
-    def _set_auth_headers(self, headers, scopes):
+    def _set_auth_headers(self, headers: dict, scopes: list[Scope]):
         if not 'Authorization' in headers:
             auth = self.token_provider.get_token(scopes)
             headers['Authorization'] = 'Bearer {}'.format(auth['access_token'])

src/autodesk_forge_sdk/base.py

@@ -1,21 +1,21 @@
 import requests
 
 class BaseClient(object):
-    def __init__(self, base_url):
+    def __init__(self, base_url: str):
         self.base_url = base_url
 
-    def _resolve_url(self, url):
+    def _resolve_url(self, url: str) -> str:
         if url.startswith('/'):
             url = self.base_url + url
         return url
 
-    def _get(self, url, params=None, headers=None):
+    def _get(self, url: str, params: dict=None, headers: dict=None) -> requests.Response:
         url = self._resolve_url(url)
         response = requests.get(url, params=params, headers=headers)
         response.raise_for_status()
         return response
 
-    def _post(self, url, form=None, json=None, buff=None, params=None, headers=None):
+    def _post(self, url: str, form: dict=None, json: dict=None, buff=None, params: dict=None, headers: dict=None) -> requests.Response:
         url = self._resolve_url(url)
         response = None
         if form:
@@ -29,7 +29,7 @@ def _post(self, url, form=None, json=None, buff=None, params=None, headers=None)
         response.raise_for_status()
         return response
 
-    def _put(self, url, form=None, json=None, buff=None, params=None, headers=None):
+    def _put(self, url: str, form: dict=None, json: dict=None, buff=None, params: dict=None, headers: dict=None) -> requests.Response:
         url = self._resolve_url(url)
         response = None
         if form:
@@ -43,7 +43,7 @@ def _put(self, url, form=None, json=None, buff=None, params=None, headers=None):
         response.raise_for_status()
         return response
 
-    def _delete(self, url, params=None, headers=None):
+    def _delete(self, url: str, params: dict=None, headers: dict=None) -> requests.Response:
         url = self._resolve_url(url)
         response = requests.delete(url, params=params, headers=headers)
         response.raise_for_status()

src/autodesk_forge_sdk/md.py

@@ -33,7 +33,7 @@ class ModelDerivativeClient(BaseOAuthClient):
     **Documentation**: https://forge.autodesk.com/en/docs/model-derivative/v2/reference/http
     """
 
-    def __init__(self, token_provider, base_url=BASE_URL):
+    def __init__(self, token_provider: TokenProviderInterface(), base_url: str = BASE_URL):
         """
         Create new instance of the client.
 
@@ -62,7 +62,7 @@ def get_token(self, scopes):
         """
         BaseOAuthClient.__init__(self, token_provider, base_url)
 
-    def get_formats(self):
+    def get_formats(self) -> dict:
         """
         Return 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.
@@ -77,13 +77,13 @@ def get_formats(self):
             FORGE_CLIENT_ID = os.environ["FORGE_CLIENT_ID"]
             FORGE_CLIENT_SECRET = os.environ["FORGE_CLIENT_SECRET"]
             client = ModelDerivativeClient(OAuthTokenProvider(FORGE_CLIENT_ID, FORGE_CLIENT_SECRET))
-            formats = client.get_formats()
-            print(formats)
+            resp = client.get_formats()
+            print(resp.formats)
             ```
         """
         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):
+    def submit_job(self, urn: str, output_formats: list[dict], output_region: str, root_filename: str=None, workflow_id: str=None, workflow_attr: str=None, force: bool = False) -> dict:
         """
         Translate a design from one format to another format.
 

src/autodesk_forge_sdk/oss.py

@@ -39,7 +39,7 @@ class OSSClient(BaseOAuthClient):
     **Documentation**: https://forge.autodesk.com/en/docs/data/v2/reference/http
     """
 
-    def __init__(self, token_provider, base_url=BASE_URL):
+    def __init__(self, token_provider: TokenProviderInterface(), base_url: str = BASE_URL):
         """
         Create new instance of the client.
 
@@ -68,7 +68,7 @@ def get_token(self, scopes):
         """
         BaseOAuthClient.__init__(self, token_provider, base_url)
 
-    def _get_paginated(self, url, scopes, params=None, headers=None):
+    def _get_paginated(self, url: str, scopes: list[Scope], params: dict=None, headers: dict=None) -> list:
         items = []
         while url:
             json = self._get(url, scopes, params, headers).json()
@@ -79,7 +79,7 @@ def _get_paginated(self, url, scopes, params=None, headers=None):
                 url = None
         return items
 
-    def get_buckets(self, region=None, limit=None, start_at=None):
+    def get_buckets(self, region: str=None, limit: int=None, start_at: str=None) -> dict:
         """
         List buckets owned by the application, using pagination.
 
@@ -112,7 +112,7 @@ def get_buckets(self, region=None, limit=None, start_at=None):
             params['startAt'] = start_at
         return self._get('/buckets', READ_SCOPES, params).json()
 
-    def get_all_buckets(self, region=None):
+    def get_all_buckets(self, region: str=None) -> list:
         """
         List all buckets owned by the application. Similar to `OSSClient.get_buckets` but returning all results without pagination.
 
@@ -138,7 +138,7 @@ def get_all_buckets(self, region=None):
             params['region'] = region
         return self._get_paginated('/buckets', READ_SCOPES, params)
 
-    def get_bucket_details(self, bucket_key):
+    def get_bucket_details(self, bucket_key: str) -> dict:
         """
         Get bucket details in JSON format if the caller is the owner of the bucket.
         A request by any other application will result in a response of 403 Forbidden.
@@ -160,7 +160,7 @@ def get_bucket_details(self, bucket_key):
         """
         return self._get('/buckets/{}/details'.format(quote(bucket_key)), READ_SCOPES).json()
 
-    def create_bucket(self, bucket_key, data_retention_policy, region):
+    def create_bucket(self, bucket_key: str, data_retention_policy: DataRetention, region: str) -> dict:
         """
         Create a bucket. Buckets are arbitrary spaces that are created by applications
         and are used to store objects for later retrieval. A bucket is owned by the application
@@ -197,7 +197,7 @@ def create_bucket(self, bucket_key, data_retention_policy, region):
         }
         return self._post('/buckets', WRITE_SCOPES, json=json, headers=headers).json()
 
-    def delete_bucket(self, bucket_key):
+    def delete_bucket(self, bucket_key: str):
         """
         Delete a bucket. The bucket must be owned by the application.
 
@@ -206,9 +206,9 @@ def delete_bucket(self, bucket_key):
         Args:
             bucket_key (str): Name of the bucket to be deleted.
         """
-        return self._delete('/buckets/{}'.format(quote(bucket_key)), DELETE_SCOPES)
+        self._delete('/buckets/{}'.format(quote(bucket_key)), DELETE_SCOPES)
 
-    def get_objects(self, bucket_key, limit=None, begins_with=None, start_at=None):
+    def get_objects(self, bucket_key: str, limit: int=None, begins_with: str=None, start_at: str=None) -> dict:
         """
         List objects in bucket, using pagination. It is only available to the bucket creator.
 
@@ -243,7 +243,7 @@ def get_objects(self, bucket_key, limit=None, begins_with=None, start_at=None):
             params['startAt'] = start_at
         return self._get('/buckets/{}/objects'.format(quote(bucket_key)), READ_SCOPES, params).json()
 
-    def get_all_objects(self, bucket_key, begins_with=None):
+    def get_all_objects(self, bucket_key: str, begins_with: str=None) -> list:
         """
         List all objects in bucket. Similar to `OSSClient.get_objects` but returning all results without pagination.
 
@@ -268,7 +268,7 @@ def get_all_objects(self, bucket_key, begins_with=None):
             params['beginsWith'] = begins_with
         return self._get_paginated('/buckets/{}/objects'.format(quote(bucket_key)), READ_SCOPES, params)
 
-    def get_object_details(self, bucket_key, object_key):
+    def get_object_details(self, bucket_key: str, object_key: str) -> dict:
         """
         Get object details in JSON format.
 
@@ -294,7 +294,7 @@ def get_object_details(self, bucket_key, object_key):
         url = '/buckets/{}/objects/{}'.format(quote(bucket_key), quote(object_key))
         return self._get(url, READ_SCOPES).json()
 
-    def upload_object(self, bucket_key, object_key, buff):
+    def upload_object(self, bucket_key: str, object_key: str, buff) -> list:
         """
         Upload an object. If the specified object name already exists in the bucket,
         the uploaded content will overwrite the existing content for the bucket name/object name combination.
@@ -326,7 +326,7 @@ def upload_object(self, bucket_key, object_key, buff):
         url = '/buckets/{}/objects/{}'.format(quote(bucket_key), quote(object_key))
         return self._put(url, WRITE_SCOPES, buff=buff).json()
 
-    def delete_object(self, bucket_key, object_key):
+    def delete_object(self, bucket_key: str, object_key: str):
         """
         Delete an object from bucket.
 
@@ -337,4 +337,4 @@ def delete_object(self, bucket_key, object_key):
             object_key (str): Name of the object to be removed.
         """
         url = '/buckets/{}/objects/{}'.format(quote(bucket_key), quote(object_key))
-        return self._delete(url, DELETE_SCOPES)
+        self._delete(url, DELETE_SCOPES)