mypy fixes

mypy.ini

@@ -0,0 +1,3 @@
+[mypy]
+show_error_codes = True
+ignore_missing_imports = True

netsuite/__main__.py

@@ -179,7 +179,7 @@ async def rest_api_openapi(config, args) -> str:
     return json.dumps_str(resp)
 
 
-async def rest_api_openapi_serve(config, args) -> str:
+async def rest_api_openapi_serve(config, args):
     rest_api = _get_rest_api_or_error(config)
     if len(args.record_types) == 0:
         logger.warning(
@@ -241,16 +241,18 @@ async def rest_api_openapi_serve(config, args) -> str:
         tempdir.rmdir()
 
 
-def _load_config_or_error(path: str, section: str) -> config.Config:
+def _load_config_or_error(path: str, section: str) -> config.Config:  # type: ignore[return]
     try:
-        return config.from_ini(path=path, section=section)
+        conf = config.from_ini(path=path, section=section)
     except FileNotFoundError:
         parser.error(f"Config file {path} not found")
     except KeyError as ex:
         if ex.args == (section,):
             parser.error(f"No config section `{section}` in file {path}")
         else:
             raise ex
+    else:
+        return conf
 
 
 def _get_rest_api_or_error(config: config.Config):

netsuite/client.py

@@ -4,7 +4,7 @@
 from contextlib import contextmanager
 from datetime import datetime
 from functools import wraps
-from typing import Any, Callable, Dict, List, Sequence, Union
+from typing import Any, Callable, Dict, List, Optional, Sequence, Union
 from urllib.parse import urlparse
 
 import requests
@@ -141,7 +141,7 @@ def __repr__(self) -> str:
 
     def __init__(
         self,
-        config: Union[Config, Dict],
+        config: Union[Config, Dict[str, Any]],
         *,
         version: str = None,
         wsdl_url: str = None,
@@ -216,7 +216,7 @@ def hostname(self) -> str:
     def service(self) -> zeep.client.ServiceProxy:
         return self.client.service
 
-    def _make_config(self, values_obj: Dict) -> Config:
+    def _make_config(self, values_obj: Union[Config, Dict[str, Any]]) -> Config:
         if isinstance(values_obj, Config):
             return values_obj
         return Config(**values_obj)
@@ -233,7 +233,7 @@ def _generate_wsdl_url(self) -> str:
         return self.wsdl_url_tmpl.format(
             underscored_version=self.underscored_version,
             # https://followingnetsuite.wordpress.com/2018/10/18/suitetalk-sandbox-urls-addendum/
-            account_id=self.config.account.lower().replace("_", "-"),
+            account_id=(self.config.account or "").lower().replace("_", "-"),
         )
 
     def _generate_cache(self) -> zeep.cache.Base:
@@ -483,7 +483,7 @@ def SupplyChain(self) -> zeep.client.Factory:
     def SupplyChainTypes(self) -> zeep.client.Factory:
         return self._type_factory("types.supplychain", "lists")
 
-    def request(self, service_name: str, *args, **kw) -> zeep.xsd.ComplexType:
+    def request(self, service_name: str, *args, **kw):
         """
         Make a web service request to NetSuite
 
@@ -504,13 +504,21 @@ def getList(
         self,
         recordType: str,
         *,
-        internalIds: Sequence[int] = (),
-        externalIds: Sequence[str] = (),
+        internalIds: Optional[Sequence[int]] = None,
+        externalIds: Optional[Sequence[str]] = None,
     ) -> List[CompoundValue]:
         """Get a list of records"""
+        if internalIds is None:
+            internalIds = []
+        else:
+            internalIds = list(internalIds)
+        if externalIds is None:
+            externalIds = []
+        else:
+            externalIds = list(externalIds)
 
-        if len(list(internalIds) + list(externalIds)) == 0:
-            raise ValueError("Please specify `internalId` and/or `externalId`")
+        if len(internalIds) + len(externalIds) == 0:
+            return []
 
         return self.request(
             "getList",
@@ -617,12 +625,21 @@ def upsertList(self, records: List[CompoundValue]) -> List[CompoundValue]:
     def getItemAvailability(
         self,
         *,
-        internalIds: Sequence[int] = (),
-        externalIds: Sequence[str] = (),
+        internalIds: Optional[Sequence[int]] = None,
+        externalIds: Optional[Sequence[str]] = None,
         lastQtyAvailableChange: datetime = None,
     ) -> List[Dict]:
-        if len(list(internalIds) + list(externalIds)) == 0:
-            raise ValueError("Please specify `internalId` and/or `externalId`")
+        if internalIds is None:
+            internalIds = []
+        else:
+            internalIds = list(internalIds)
+        if externalIds is None:
+            externalIds = []
+        else:
+            externalIds = list(externalIds)
+
+        if len(internalIds) + len(externalIds) == 0:
+            return []
 
         item_filters = [
             {"type": "inventoryItem", "internalId": internalId}

netsuite/config.py

@@ -1,5 +1,5 @@
 import configparser
-from typing import Dict
+from typing import Any, Dict, Optional, Tuple
 
 from .constants import DEFAULT_INI_PATH, DEFAULT_INI_SECTION, NOT_SET
 
@@ -17,37 +17,37 @@ class Config:
             Dictionary keys/values that will be set as attribute names/values
     """
 
-    auth_type = TOKEN
+    auth_type: str = TOKEN
     """The authentication type to use, either 'token' or 'credentials'"""
 
-    account = None
+    account: Optional[str] = None
     """The NetSuite account ID"""
 
-    consumer_key = None
+    consumer_key: Optional[str] = None
     """The OAuth 1.0 consumer key"""
 
-    consumer_secret = None
+    consumer_secret: Optional[str] = None
     """The OAuth 1.0 consumer secret"""
 
-    token_id = None
+    token_id: Optional[str] = None
     """The OAuth 1.0 token ID"""
 
-    token_secret = None
+    token_secret: Optional[str] = None
     """The OAuth 1.0 token secret"""
 
-    application_id = None
+    application_id: Optional[str] = None
     """Application ID, used with auth_type=credentials"""
 
-    email = None
+    email: Optional[str] = None
     """Account e-mail, used with auth_type=credentials"""
 
-    password = None
+    password: Optional[str] = None
     """Account password, used with auth_type=credentials"""
 
     preferences = None
     """Additional preferences"""
 
-    _settings_mapping = (
+    _settings_mapping: Tuple[Tuple[str, Dict[str, Any]], ...] = (
         (
             "account",
             {"type": str, "required": True},
@@ -86,13 +86,13 @@ class Config:
         ),
     )
 
-    def __init__(self, **opts) -> None:
+    def __init__(self, **opts):
         self._set(opts)
 
     def __contains__(self, key: str) -> bool:
         return hasattr(self, key)
 
-    def _set_auth_type(self, value: str) -> None:
+    def _set_auth_type(self, value: str):
         self._validate_attr("auth_type", value, str, True, {})
         self.auth_type = value
         assert self.auth_type in (TOKEN, CREDENTIALS)
@@ -103,7 +103,7 @@ def is_token_auth(self) -> bool:
     def is_credentials_auth(self) -> bool:
         return self.auth_type == CREDENTIALS
 
-    def _set(self, dct: Dict[str, object]) -> None:
+    def _set(self, dct: Dict[str, Any]):
         # As other setting validations depend on auth_type we set it first
         auth_type = dct.get("auth_type", self.auth_type)
         self._set_auth_type(auth_type)
@@ -124,8 +124,8 @@ def _set(self, dct: Dict[str, object]) -> None:
             setattr(self, attr, (None if value is NOT_SET else value))
 
     def _validate_attr(
-        self, attr: str, value: object, type_: object, required: bool, opts: dict
-    ) -> None:
+        self, attr: str, value: Any, type_: Any, required: bool, opts: Dict[str, Any]
+    ):
         if required and value is NOT_SET:
             required_for_auth_type = opts.get("required_for_auth_type")
             if required_for_auth_type:
@@ -146,7 +146,7 @@ def from_ini(
     with open(path) as fp:
         iniconf.read_file(fp)
 
-    config_dict = {"preferences": {}}
+    config_dict: Dict[str, Any] = {"preferences": {}}
 
     for key, val in iniconf[section].items():
         if key.startswith("preferences_"):

netsuite/passport.py

@@ -4,8 +4,6 @@
 from datetime import datetime
 from typing import Dict, TypeVar
 
-from zeep.xsd.valueobjects import CompoundValue
-
 from .config import Config
 
 NetSuite = TypeVar("NetSuite")
@@ -25,7 +23,7 @@ def __init__(
         self.email = email
         self.password = password
 
-    def get_element(self) -> CompoundValue:
+    def get_element(self):
         return self.ns.Core.Passport(
             account=self.account,
             email=self.email,
@@ -85,13 +83,13 @@ def _get_signature_value(self, nonce: str, timestamp: str) -> str:
         ).digest()
         return base64.b64encode(hashed).decode()
 
-    def _get_signature(self, nonce: str, timestamp: str) -> CompoundValue:
-        return self.ns.Core.TokenPassportSignature(
+    def _get_signature(self, nonce: str, timestamp: str):
+        return self.ns.Core.TokenPassportSignature(  # type: ignore[attr-defined]
             self._get_signature_value(nonce, timestamp),
             algorithm="HMAC-SHA256",
         )
 
-    def get_element(self) -> CompoundValue:
+    def get_element(self):
         nonce = self._generate_nonce()
         timestamp = self._generate_timestamp()
         signature = self._get_signature(nonce, timestamp)
@@ -107,6 +105,11 @@ def get_element(self) -> CompoundValue:
 
 def make(ns: NetSuite, config: Config) -> Dict:
     if config.auth_type == "token":
+        assert isinstance(config.account, str)
+        assert isinstance(config.consumer_key, str)
+        assert isinstance(config.consumer_secret, str)
+        assert isinstance(config.token_id, str)
+        assert isinstance(config.token_secret, str)
         token_passport = TokenPassport(
             ns,
             account=config.account,
@@ -117,6 +120,9 @@ def make(ns: NetSuite, config: Config) -> Dict:
         )
         return {"tokenPassport": token_passport.get_element()}
     elif config.auth_type == "credentials":
+        assert isinstance(config.account, str)
+        assert isinstance(config.email, str)
+        assert isinstance(config.password, str)
         passport = UserCredentialsPassport(
             ns,
             account=config.account,

netsuite/rest_api.py

@@ -1,16 +1,15 @@
 import asyncio
 import logging
-from typing import Iterable, Optional
+from typing import Sequence
 
 from . import json
-from .types import JsonDict
 
 try:
     import httpx
 except ImportError:
 
-    class httpx:
-        Response = None  # NOTE: For type hint to work
+    class httpx:  # type: ignore[no-redef]
+        Response = None
 
 
 try:
@@ -48,7 +47,7 @@ def __init__(
         consumer_secret: str,
         token_id: str,
         token_secret: str,
-        default_timeout: str = 60,
+        default_timeout: int = 60,
         concurrent_requests: int = 10,
     ):
         if not self.has_required_dependencies():
@@ -69,7 +68,7 @@ def __init__(
     def has_required_dependencies(cls) -> bool:
         return httpx is not None and OAuth1Auth is not None
 
-    async def get(self, subpath: str, **request_kw) -> JsonDict:
+    async def get(self, subpath: str, **request_kw):
         return await self.request("GET", subpath, **request_kw)
 
     async def post(self, subpath: str, **request_kw):
@@ -88,7 +87,7 @@ async def patch(self, subpath: str, **request_kw):
     async def delete(self, subpath: str, **request_kw):
         return await self.request("DELETE", subpath, **request_kw)
 
-    async def suiteql(self, q: str, limit: int = 10, offset: int = 0) -> JsonDict:
+    async def suiteql(self, q: str, limit: int = 10, offset: int = 0):
         return await self.request(
             "POST",
             "/query/v1/suiteql",
@@ -97,7 +96,7 @@ async def suiteql(self, q: str, limit: int = 10, offset: int = 0) -> JsonDict:
             params={"limit": limit, "offset": offset},
         )
 
-    async def jsonschema(self, record_type: str, **request_kw) -> JsonDict:
+    async def jsonschema(self, record_type: str, **request_kw):
         headers = {
             **request_kw.pop("headers", {}),
             "Accept": "application/schema+json",
@@ -109,7 +108,7 @@ async def jsonschema(self, record_type: str, **request_kw) -> JsonDict:
             **request_kw,
         )
 
-    async def openapi(self, record_types: Iterable[str] = (), **request_kw) -> JsonDict:
+    async def openapi(self, record_types: Sequence[str] = (), **request_kw):
         headers = {
             **request_kw.pop("headers", {}),
             "Accept": "application/swagger+json",
@@ -127,9 +126,7 @@ async def openapi(self, record_types: Iterable[str] = (), **request_kw) -> JsonD
             **request_kw,
         )
 
-    async def request(
-        self, method: str, subpath: str, **request_kw
-    ) -> Optional[JsonDict]:
+    async def request(self, method: str, subpath: str, **request_kw):
         resp = await self._raw_request(method, subpath, **request_kw)
 
         if resp.status_code < 200 or resp.status_code > 299: