refactor(boxsdkgen): annotate optional types and add py.typed for type checking (box/box-codegen#996)

.codegen.json

@@ -1 +1 @@
-{ "engineHash": "fd56d95", "specHash": "c5a35a6", "version": "4.17.0" }
+{ "engineHash": "d55532c", "specHash": "c5a35a6", "version": "4.17.0" }

box_sdk_gen/box/ccg_auth.py

@@ -35,7 +35,7 @@ def __init__(
         *,
         enterprise_id: Optional[str] = None,
         user_id: Optional[str] = None,
-        token_storage: TokenStorage = None
+        token_storage: Optional[TokenStorage] = None
     ):
         """
         :param client_id: Box API key used for identifying the application the user is authenticating with
@@ -47,7 +47,7 @@ def __init__(
         :param user_id: The user id to authenticate. This value is not required. But if it is provided, then the user will be auto-authenticated at the time of the first API call., defaults to None
         :type user_id: Optional[str], optional
         :param token_storage: Object responsible for storing token. If no custom implementation provided,the token will be stored in memory., defaults to None
-        :type token_storage: TokenStorage, optional
+        :type token_storage: Optional[TokenStorage], optional
         """
         if token_storage is None:
             token_storage = InMemoryTokenStorage()
@@ -122,7 +122,7 @@ def retrieve_authorization_header(
         return ''.join(['Bearer ', token.access_token])
 
     def with_user_subject(
-        self, user_id: str, *, token_storage: TokenStorage = None
+        self, user_id: str, *, token_storage: Optional[TokenStorage] = None
     ) -> 'BoxCCGAuth':
         """
         Create a new BoxCCGAuth instance that uses the provided user ID as the subject ID.
@@ -138,7 +138,7 @@ def with_user_subject(
         :param user_id: The id of the user to authenticate
         :type user_id: str
         :param token_storage: Object responsible for storing token in newly created BoxCCGAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
-        :type token_storage: TokenStorage, optional
+        :type token_storage: Optional[TokenStorage], optional
         """
         if token_storage is None:
             token_storage = InMemoryTokenStorage()
@@ -152,14 +152,14 @@ def with_user_subject(
         return BoxCCGAuth(config=new_config)
 
     def with_enterprise_subject(
-        self, enterprise_id: str, *, token_storage: TokenStorage = None
+        self, enterprise_id: str, *, token_storage: Optional[TokenStorage] = None
     ) -> 'BoxCCGAuth':
         """
         Create a new BoxCCGAuth instance that uses the provided enterprise ID as the subject ID.
         :param enterprise_id: The id of the enterprise to authenticate
         :type enterprise_id: str
         :param token_storage: Object responsible for storing token in newly created BoxCCGAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
-        :type token_storage: TokenStorage, optional
+        :type token_storage: Optional[TokenStorage], optional
         """
         if token_storage is None:
             token_storage = InMemoryTokenStorage()

box_sdk_gen/box/developer_token_auth.py

@@ -34,10 +34,12 @@ def __init__(
 
 
 class BoxDeveloperTokenAuth(Authentication):
-    def __init__(self, token: str, *, config: DeveloperTokenConfig = None, **kwargs):
+    def __init__(
+        self, token: str, *, config: Optional[DeveloperTokenConfig] = None, **kwargs
+    ):
         """
         :param config: Configuration object of DeveloperTokenAuth., defaults to None
-        :type config: DeveloperTokenConfig, optional
+        :type config: Optional[DeveloperTokenConfig], optional
         """
         super().__init__(**kwargs)
         self.token = token

box_sdk_gen/box/errors.py

@@ -77,7 +77,7 @@ def __init__(
         self,
         status_code: int,
         headers: Dict[str, str],
-        body: Dict = None,
+        body: Optional[Dict] = None,
         raw_body: Optional[str] = None,
         code: Optional[str] = None,
         context_info: Optional[Dict[str, Any]] = None,
@@ -122,9 +122,9 @@ def __init__(
         response_info: ResponseInfo,
         message: str,
         timestamp: Optional[datetime] = None,
-        error: Optional[str] = None,
+        error: Optional[Exception] = None,
         *,
-        data_sanitizer: DataSanitizer = None,
+        data_sanitizer: Optional[DataSanitizer] = None,
         **kwargs,
     ):
         super().__init__(message=message, timestamp=timestamp, error=error, **kwargs)

box_sdk_gen/box/event_stream.py

@@ -149,7 +149,8 @@ def _do_long_poll(self) -> str:
         try:
             self._long_polling_retries += 1
 
-            long_poll_url = self._long_polling_info.url
+            # _get_long_poll_info always runs before the first poll
+            long_poll_url: str = self._long_polling_info.url  # type: ignore[union-attr, assignment]
             separator = '&' if '?' in long_poll_url else '?'
             long_poll_with_stream_position = (
                 f"{long_poll_url}{separator}stream_position={self._stream_position}"

box_sdk_gen/box/jwt_auth.py

@@ -165,8 +165,8 @@ def __init__(
         enterprise_id: Optional[str] = None,
         user_id: Optional[str] = None,
         algorithm: Optional[JwtAlgorithm] = JwtAlgorithm.RS256,
-        token_storage: TokenStorage = None,
-        private_key_decryptor: PrivateKeyDecryptor = None
+        token_storage: Optional[TokenStorage] = None,
+        private_key_decryptor: Optional[PrivateKeyDecryptor] = None
     ):
         """
         :param client_id: App client ID
@@ -356,7 +356,7 @@ def retrieve_authorization_header(
         return ''.join(['Bearer ', token.access_token])
 
     def with_user_subject(
-        self, user_id: str, *, token_storage: TokenStorage = None
+        self, user_id: str, *, token_storage: Optional[TokenStorage] = None
     ) -> 'BoxJWTAuth':
         """
         Create a new BoxJWTAuth instance that uses the provided user ID as the subject of the JWT assertion.
@@ -372,7 +372,7 @@ def with_user_subject(
         :param user_id: The id of the user to authenticate
         :type user_id: str
         :param token_storage: Object responsible for storing token in newly created BoxJWTAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
-        :type token_storage: TokenStorage, optional
+        :type token_storage: Optional[TokenStorage], optional
         """
         if token_storage is None:
             token_storage = InMemoryTokenStorage()
@@ -390,14 +390,14 @@ def with_user_subject(
         return new_auth
 
     def with_enterprise_subject(
-        self, enterprise_id: str, *, token_storage: TokenStorage = None
+        self, enterprise_id: str, *, token_storage: Optional[TokenStorage] = None
     ) -> 'BoxJWTAuth':
         """
         Create a new BoxJWTAuth instance that uses the provided enterprise ID as the subject of the JWT assertion.
         :param enterprise_id: The id of the enterprise to authenticate
         :type enterprise_id: str
         :param token_storage: Object responsible for storing token in newly created BoxJWTAuth. If no custom implementation provided, the token will be stored in memory., defaults to None
-        :type token_storage: TokenStorage, optional
+        :type token_storage: Optional[TokenStorage], optional
         """
         if token_storage is None:
             token_storage = InMemoryTokenStorage()

box_sdk_gen/box/oauth.py

@@ -37,7 +37,11 @@
 
 class OAuthConfig:
     def __init__(
-        self, client_id: str, client_secret: str, *, token_storage: TokenStorage = None
+        self,
+        client_id: str,
+        client_secret: str,
+        *,
+        token_storage: Optional[TokenStorage] = None
     ):
         if token_storage is None:
             token_storage = InMemoryTokenStorage()
@@ -85,7 +89,9 @@ def __init__(self, config: OAuthConfig, **kwargs):
         self.config = config
         self.token_storage = self.config.token_storage
 
-    def get_authorize_url(self, *, options: GetAuthorizeUrlOptions = None) -> str:
+    def get_authorize_url(
+        self, *, options: Optional[GetAuthorizeUrlOptions] = None
+    ) -> str:
         """
         Get the authorization URL for the app user.
         """

box_sdk_gen/client.py

@@ -1,5 +1,7 @@
 from typing import Dict
 
+from typing import Optional
+
 from box_sdk_gen.managers.authorization import AuthorizationManager
 
 from box_sdk_gen.managers.files import FilesManager
@@ -216,7 +218,9 @@
 
 
 class BoxClient:
-    def __init__(self, auth: Authentication, *, network_session: NetworkSession = None):
+    def __init__(
+        self, auth: Authentication, *, network_session: Optional[NetworkSession] = None
+    ):
         if network_session is None:
             network_session = NetworkSession(base_urls=BaseUrls())
         self.auth = auth
@@ -531,12 +535,12 @@ def with_suppressed_notifications(self) -> 'BoxClient':
         )
 
     def with_extra_headers(
-        self, *, extra_headers: Dict[str, str] = None
+        self, *, extra_headers: Optional[Dict[str, str]] = None
     ) -> 'BoxClient':
         """
         Create a new client with a custom set of headers that will be included in every API call
         :param extra_headers: Custom set of headers that will be included in every API call, defaults to None
-        :type extra_headers: Dict[str, str], optional
+        :type extra_headers: Optional[Dict[str, str]], optional
         """
         if extra_headers is None:
             extra_headers = {}

box_sdk_gen/internal/base_object.py

@@ -1,13 +1,18 @@
 from datetime import datetime, date
 from enum import EnumMeta, Enum
+import typing as _typing
 from typing import get_args, get_origin, Union, Optional
 from .null_value import NullValue
 
 
 class BaseObject:
-    _discriminator = (None, {})
-    _json_to_fields_mapping = {}
-    _fields_to_json_mapping = {}
+    # typing is aliased because this module is star-exported and Collection is also a schema name
+    _discriminator = (
+        None,
+        {},
+    )  # type: _typing.Tuple[Optional[str], _typing.Collection[str]]
+    _json_to_fields_mapping = {}  # type: _typing.Dict[str, str]
+    _fields_to_json_mapping = {}  # type: _typing.Dict[str, str]
 
     def __init__(self, **kwargs):
         self._raw_data: dict = {}
@@ -40,6 +45,7 @@ def to_dict(self) -> dict:
                 continue
             if v is None:
                 continue
+            value: _typing.Any
             if isinstance(v, NullValue):
                 value = None
             elif type(v) is list:

box_sdk_gen/internal/utils.py

@@ -10,7 +10,17 @@
 from random import uniform
 from enum import Enum
 from io import SEEK_CUR, SEEK_END, SEEK_SET, BufferedIOBase, BytesIO
-from typing import Any, Callable, Dict, Iterable, Optional, TypeVar, BinaryIO
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    Optional,
+    TypeVar,
+    BinaryIO,
+    overload,
+)
 
 from abc import abstractmethod
 from typing import Any
@@ -103,7 +113,7 @@ def seek(self, position, whence=SEEK_SET):
         return self._position
 
 
-def get_env_var(name: str) -> str:
+def get_env_var(name: str) -> Optional[str]:
     return os.getenv(name)
 
 
@@ -170,25 +180,38 @@ def prepare_params(map: Dict[str, Optional[str]]) -> Dict[str, str]:
     return {k: v for k, v in map.items() if v is not None}
 
 
-def to_string(value: Any) -> Optional[str]:
-    if value is None:
-        return None
-    if isinstance(value, datetime.datetime):
-        return date_time_to_string(value)
-    if isinstance(value, datetime.date):
-        return date_to_string(value)
-    if (
-        isinstance(value, BaseObject)
-        or isinstance(value, list)
-        and len(value) >= 1
-        and isinstance(value[0], BaseObject)
-    ):
-        return ''.join(sd_to_json(serialize(value)).split())
-    if isinstance(value, list):
-        return ','.join(map(to_string, value))
-    if isinstance(value, Enum):
-        return value.value
-    return str(value)
+# Overloads exist for type checkers only, so the runtime function stays unchanged
+if TYPE_CHECKING:
+
+    @overload
+    def to_string(value: None) -> None: ...
+
+    @overload
+    def to_string(value: Any) -> str: ...
+
+    def to_string(value: Any) -> Optional[str]: ...
+
+else:
+
+    def to_string(value: Any) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, datetime.datetime):
+            return date_time_to_string(value)
+        if isinstance(value, datetime.date):
+            return date_to_string(value)
+        if (
+            isinstance(value, BaseObject)
+            or isinstance(value, list)
+            and len(value) >= 1
+            and isinstance(value[0], BaseObject)
+        ):
+            return ''.join(sd_to_json(serialize(value)).split())
+        if isinstance(value, list):
+            return ','.join(map(to_string, value))
+        if isinstance(value, Enum):
+            return value.value
+        return str(value)
 
 
 class HashName(str, Enum):
@@ -210,7 +233,7 @@ def digest_hash(self, encoding):
         return base64.b64encode(self.hash.digest()).decode("utf-8")
 
 
-def hex_to_base_64(data: hex):
+def hex_to_base_64(data: str) -> str:
     return base64.b64encode(bytes.fromhex(data)).decode()
 
 
@@ -321,7 +344,7 @@ class JwtSignOptions(BaseObject):
     def __init__(
         self,
         algorithm: JwtAlgorithm,
-        headers: Dict[str, str] = None,
+        headers: Optional[Dict[str, str]] = None,
         audience: Optional[str] = None,
         issuer: Optional[str] = None,
         subject: Optional[str] = None,

box_sdk_gen/managers/ai.py

@@ -239,7 +239,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()

box_sdk_gen/managers/ai_studio.py

@@ -68,7 +68,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()

box_sdk_gen/managers/app_item_associations.py

@@ -38,7 +38,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()

box_sdk_gen/managers/archives.py

@@ -46,7 +46,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()

box_sdk_gen/managers/authorization.py

@@ -93,7 +93,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()

box_sdk_gen/managers/automate_workflows.py

@@ -52,7 +52,7 @@ def __init__(
         self,
         *,
         auth: Optional[Authentication] = None,
-        network_session: NetworkSession = None
+        network_session: Optional[NetworkSession] = None
     ):
         if network_session is None:
             network_session = NetworkSession()