Merge pull request #182 from denisneuf/wp/issue-180/0

Fixed attribute error and added annotations

ad_api/api/dsp/client.py

@@ -1,12 +1,13 @@
+from typing import Dict
 from ad_api.base import Client
 
 
 class DspClient(Client):
     @property
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         return {
             'User-Agent': self.user_agent,
-            'Amazon-Advertising-API-ClientId': self.credentials.client_id,
-            'Authorization': 'Bearer %s' % self.auth.access_token,
+            'Amazon-Advertising-API-ClientId': self.credentials['client_id'],
+            'Authorization': f'Bearer {self.auth.access_token}',
             'Content-Type': 'application/json',
         }

ad_api/base/client.py

@@ -17,23 +17,25 @@
 from urllib.parse import urlparse, quote
 from ad_api.base.credential_provider import CredentialProvider
 import ad_api.version as vd
+from typing import Any, Dict, Optional
 
 log = logging.getLogger(__name__)
 role_cache = TTLCache(maxsize=int(os.environ.get('AD_API_AUTH_CACHE_SIZE', 10)), ttl=3200)
+_DEFAULT_MARKETPLACE = Marketplaces[os.environ['AD_API_DEFAULT_MARKETPLACE']] if 'AD_API_DEFAULT_MARKETPLACE' in os.environ else Marketplaces.EU
 
 
 class Client(BaseClient):
     def __init__(
         self,
-        account='default',
-        marketplace: Marketplaces = Marketplaces[os.environ['AD_API_DEFAULT_MARKETPLACE']] if 'AD_API_DEFAULT_MARKETPLACE' in os.environ else Marketplaces.EU,
-        credentials=None,
-        proxies=None,
-        verify=True,
-        timeout=None,
-        debug=False,
-        access_token=None,
-        verify_additional_credentials=True,
+        account: str = 'default',
+        marketplace: Marketplaces = _DEFAULT_MARKETPLACE,
+        credentials: Optional[Dict[str, str]] = None,
+        proxies: Optional[Dict[str, str]] = None,
+        verify: bool = True,
+        timeout: Optional[int] = None,
+        debug: bool = False,
+        access_token: Optional[str] = None,
+        verify_additional_credentials: bool = True,
     ):
         self.credentials = CredentialProvider(account, credentials, verify_additional_credentials).credentials
         self._auth = AccessTokenClient(
@@ -53,7 +55,7 @@ def __init__(
         self.user_agent += f'-{version}'
 
     @property
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         data = {
             'User-Agent': self.user_agent,
             'Amazon-Advertising-API-ClientId': self.credentials['client_id'],
@@ -69,7 +71,7 @@ def auth(self) -> AccessTokenResponse:
         return self._auth.get_auth() if self._access_token is None else AccessTokenResponse(access_token=self._access_token)
 
     @staticmethod
-    def _download(self, params: dict = None, headers=None) -> ApiResponse:
+    def _download(self: Any, params: Optional[dict] = None, headers: Optional[dict] = None) -> ApiResponse:
         location = params.get("url")
 
         try:
@@ -211,9 +213,9 @@ def _download(self, params: dict = None, headers=None) -> ApiResponse:
     def _request(
         self,
         path: str,
-        data: str = None,
-        params: dict = None,
-        headers=None,
+        data: Optional[str] = None,
+        params: Optional[dict] = None,
+        headers: Optional[dict] = None,
     ) -> ApiResponse:
         if params is None:
             params = {}
@@ -262,7 +264,7 @@ def _request(
         return self._check_response(res)
 
     @staticmethod
-    def _check_response(res) -> ApiResponse:
+    def _check_response(res: requests.Response) -> ApiResponse:
         headers = vars(res).get('headers')
         status_code = vars(res).get('status_code')
 
@@ -294,4 +296,3 @@ def _check_response(res) -> ApiResponse:
                 js = res.content
 
             raise exception(status_code, js, headers)
-            exit(res.status_code)

ad_api/base/credential_provider.py

@@ -1,7 +1,7 @@
 import abc
 import os
 
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 import confuse
 import logging
@@ -26,17 +26,17 @@ class MissingCredentials(Exception):
 
 
 class BaseCredentialProvider(abc.ABC):
-    def __init__(self, credentials: dict or str, *args, **kwargs):
+    def __init__(self, credentials: Dict[str, str], *args: Any, **kwargs: Any):
         self.credentials = credentials
 
     @abc.abstractmethod
-    def load_credentials(self, verify_additional_credentials):
+    def load_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
         pass
 
-    def _get_env(self, key):
-        return os.environ.get(f'{key}', os.environ.get(key))
+    def _get_env(self, key: str) -> str:
+        return os.environ.get(key)
 
-    def check_credentials(self, verify_additional_credentials):
+    def check_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
         creds_to_check = REQUIRED_CREDENTIALS[:]
         if verify_additional_credentials:
             creds_to_check += ADDITIONAL_CREDENTIALS
@@ -50,10 +50,10 @@ def check_credentials(self, verify_additional_credentials):
 
 
 class FromEnvCredentialProvider(BaseCredentialProvider):
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any):
         pass
 
-    def load_credentials(self, verify_additional_credentials):
+    def load_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
         account_data = dict(
             refresh_token=self._get_env('AD_API_REFRESH_TOKEN'),
             client_id=self._get_env('AD_API_CLIENT_ID'),
@@ -64,26 +64,23 @@ def load_credentials(self, verify_additional_credentials):
         self.check_credentials(verify_additional_credentials)
         return self.credentials
 
-    def _get_env(self, key):
-        return os.environ.get(f'{key}', os.environ.get(key))
-
 
 class FromCodeCredentialProvider(BaseCredentialProvider):
-    def __init__(self, credentials: dict, *args, **kwargs):
+    def __init__(self, credentials: Dict, *args: Any, **kwargs: Any):
         super().__init__(credentials)
 
-    def load_credentials(self, verify_additional_credentials):
+    def load_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
         self.check_credentials(verify_additional_credentials)
         return self.credentials
 
 
 class FromConfigFileCredentialProvider(BaseCredentialProvider):
     file = 'credentials.yml'  # Will moved to default confuse config.yaml
 
-    def __init__(self, account: str, *args, **kwargs):
+    def __init__(self, account: str, *args: Any, **kwargs: Any):
         self.account = account
 
-    def load_credentials(self, verify_additional_credentials):
+    def load_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
         try:
             config = confuse.Configuration('python-ad-api')
             config_filename = os.path.join(config.config_dir(), self.file)
@@ -104,10 +101,10 @@ def load_credentials(self, verify_additional_credentials):
 
 
 class CredentialProvider:
-    def load_credentials(self, verify_additional_credentials):
-        pass
+    def load_credentials(self, verify_additional_credentials: bool) -> Dict[str, str]:
+        return {}
 
-    def _get_env(self, key):
+    def _get_env(self, key: str) -> Optional[str]:
         return os.environ.get(f'{key}', os.environ.get(key))
 
     def __init__(
@@ -144,7 +141,7 @@ def __init__(
                 logging.error(MissingCredentials)
 
     class Config:
-        def __init__(self, **kwargs):
+        def __init__(self, **kwargs: Any):
             self.refresh_token = kwargs.get('refresh_token')
             self.client_id = kwargs.get('client_id')
             self.client_secret = kwargs.get('client_secret')