Revert "chore: some type updates (#473)"

This reverts commit 98a9a16.

Author: giovanni-guidini

shared/helpers/cache.py

@@ -142,7 +142,7 @@ def args_indexes_to_log(self) -> List[int]:
         return self.get("args_indexes_to_log", [])
 
     @property
-    def kwargs_keys_to_log(self) -> List[str]:
+    def kwargs_keys_to_log(self) -> List[Any]:
         """List of args from the function to be logged (if present)"""
         return self.get("kwargs_keys_to_log", [])
 

shared/torngit/base.py

@@ -1,5 +1,6 @@
 import re
 from enum import Enum
+from typing import Dict, List, Optional, Tuple
 
 import httpx
 
@@ -12,6 +13,7 @@
     OnRefreshCallback,
     Token,
 )
+from shared.typings.torngit import TorngitInstanceData
 
 get_start_of_line = re.compile(r"@@ \-(\d+),?(\d*) \+(\d+),?(\d*).*").match
 
@@ -26,67 +28,66 @@ class TokenType(Enum):
     pull = "pull"
 
 
-TokenTypeMapping = dict[TokenType, Token]
+TokenTypeMapping = Dict[TokenType, Token]
 
 
 class TorngitBaseAdapter(object):
     _repo_url: str | None = None
     _aws_key = None
-    _oauth: OauthConsumerToken | None
-    _on_token_refresh: OnRefreshCallback | None
+    _oauth: OauthConsumerToken | None = None
+    _on_token_refresh: OnRefreshCallback = None
     _token: Token | None = None
-    verify_ssl: str | bool
-    _timeout: httpx.Timeout
-    valid_languages = set(language[0] for language in Repository.Languages.choices)
+    verify_ssl = None
 
-    # These are set by the subclasses
-    service_url: str | None = None
-    service: str | None = None
-    urls: dict[str, str] | None = None
+    valid_languages = set(language.value for language in Repository.Languages)
 
     def __init__(
         self,
         oauth_consumer_token: OauthConsumerToken | None = None,
-        timeout: httpx.Timeout | tuple[int, int] | None = None,
+        timeouts=None,
         token: Token | None = None,
         token_type_mapping: TokenTypeMapping | None = None,
-        on_token_refresh: OnRefreshCallback | None = None,
-        verify_ssl: str | bool = True,
+        on_token_refresh: OnRefreshCallback = None,
+        verify_ssl=None,
         **kwargs,
-    ) -> None:
-        if isinstance(timeout, tuple):
-            self._timeout = httpx.Timeout(timeout[1], connect=timeout[0])
-        else:
-            self._timeout = timeout or httpx.Timeout(30, connect=10)
+    ):
+        self._timeouts = timeouts or [10, 30]
         self._token = token
         self._on_token_refresh = on_token_refresh
         self._token_type_mapping = token_type_mapping or {}
         self._oauth = oauth_consumer_token
-        self.verify_ssl = verify_ssl
-        self.data = {
-            "owner": kwargs.get("owner", {}),
-            "repo": kwargs.get("repo", {}),
-            "fallback_installations": kwargs.get("fallback_installations", None),
-            "installation": kwargs.get("installation", None),
-            "additional_data": kwargs.get("additional_data", {}),
+        self.data: TorngitInstanceData = {
+            "owner": {},
+            "repo": {},
+            "fallback_installations": None,
+            "installation": None,
+            "additional_data": {},
         }
+        self.verify_ssl = verify_ssl
+        self.data.update(kwargs)
         # This has the side effect of initializing the torngit_cache
         # (if not yet initialized)
         torngit_cache.initialize()
 
-    def __repr__(self) -> str:
+    def __repr__(self):
         return "<%s slug=%s ownerid=%s repoid=%s>" % (
             self.service,
             self.slug,
             self.data["owner"].get("ownerid"),
             self.data["repo"].get("repoid"),
         )
 
-    def get_client(self, timeout: httpx.Timeout | None = None) -> httpx.AsyncClient:
-        if timeout is None:
-            timeout = self._timeout
+    def get_client(self, timeouts: List[int] = []) -> httpx.AsyncClient:
+        if timeouts:
+            timeout = httpx.Timeout(timeouts[1], connect=timeouts[0])
+        else:
+            timeout = httpx.Timeout(self._timeouts[1], connect=self._timeouts[0])
         return httpx.AsyncClient(
-            verify=self.verify_ssl,
+            verify=(
+                self.verify_ssl
+                if not isinstance(self.verify_ssl, bool)
+                else self.verify_ssl
+            ),
             timeout=timeout,
         )
 
@@ -95,7 +96,7 @@ def get_token_by_type(self, token_type: TokenType):
             return self._token_type_mapping.get(token_type)
         return self.token
 
-    def get_token_by_type_if_none(self, token: str | None, token_type: TokenType):
+    def get_token_by_type_if_none(self, token: Optional[str], token_type: TokenType):
         if token is not None:
             return token
         return self.get_token_by_type(token_type)
@@ -105,15 +106,14 @@ def _oauth_consumer_token(self) -> OauthConsumerToken:
             raise Exception("Oauth consumer token not present")
         return self._oauth
 
-    def _validate_language(self, language: str | None) -> str | None:
+    def _validate_language(self, language: str) -> str | None:
         if language:
             language = language.lower()
-
             if language in self.valid_languages:
                 return language
         return None
 
-    def set_token(self, token: Token) -> None:
+    def set_token(self, token: OauthConsumerToken) -> None:
         self._token = token
 
     @property
@@ -329,7 +329,7 @@ async def edit_webhook(
 
     # OTHERS
 
-    async def get_authenticated(self, token=None) -> tuple[bool, bool]:
+    async def get_authenticated(self, token=None) -> Tuple[bool, bool]:
         """Finds the user permissions about about whether the user on
             `self.data["user"]` can access the repo from `self.data["repo"]`
             Returns a `can_view` and a `can_edit` permission tuple
@@ -357,7 +357,7 @@ async def get_authenticated_user(self, **kwargs):
     async def get_branches(self, token=None):
         raise NotImplementedError()
 
-    async def get_branch(self, branch_name: str, token=None):
+    async def get_branch(self, token=None):
         raise NotImplementedError()
 
     async def get_compare(
@@ -426,7 +426,7 @@ async def list_top_level_files(self, ref, token=None):
     async def get_workflow_run(self, run_id, token=None):
         raise NotImplementedError()
 
-    async def get_best_effort_branches(self, commit_sha: str, token=None) -> list[str]:
+    async def get_best_effort_branches(self, commit_sha: str, token=None) -> List[str]:
         """
         Gets a 'best effort' list of branches this commit is in.
         If a branch is returned, this means this commit is in that branch. If not, it could still be

shared/torngit/cache/__init__.py

@@ -47,7 +47,7 @@ def is_initialized(self) -> bool:
     def is_enabled(self) -> bool:
         return self._enabled
 
-    def get_ttl(self, endpoint: CachedEndpoint) -> int:
+    def get_ttl(self, endpoint: CachedEndpoint) -> dict:
         return self.ttls.get(endpoint, 120)
 
 

shared/typings/oauth_token_types.py

@@ -1,9 +1,8 @@
-from typing import Awaitable, Callable, NotRequired, TypedDict
+from typing import Awaitable, Callable, Optional, TypedDict
 
 
 class Token(TypedDict):
     key: str
-    refresh_token: NotRequired[str]
     # This information is used to identify the token owner in the logs, if present
     username: str | None
     # This represents the entity it belongs to. Entities can have the form of
@@ -14,9 +13,9 @@ class Token(TypedDict):
     entity_name: str | None
 
 
-class OauthConsumerToken(TypedDict):
-    key: str  # client_id
-    secret: str
+class OauthConsumerToken(Token):
+    secret: Optional[str]
+    refresh_token: Optional[str]
 
 
-OnRefreshCallback = Callable[[Token], Awaitable[None]]
+OnRefreshCallback = Optional[Callable[[OauthConsumerToken], Awaitable[None]]]

shared/typings/torngit.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, NotRequired, Optional, TypedDict
+from typing import Dict, List, NotRequired, Optional, TypedDict, Union
 
 from shared.reports.types import UploadType
 
@@ -38,8 +38,8 @@ class AdditionalData(TypedDict):
 
 
 class TorngitInstanceData(TypedDict):
-    owner: OwnerInfo | Dict
-    repo: RepoInfo | Dict
-    fallback_installations: List[GithubInstallationInfo | None] | None
-    installation: GithubInstallationInfo | None
-    additional_data: AdditionalData | None
+    owner: Union[OwnerInfo, Dict]
+    repo: Union[RepoInfo, Dict]
+    fallback_installations: List[Optional[GithubInstallationInfo]] | None
+    installation: Optional[GithubInstallationInfo]
+    additional_data: Optional[AdditionalData]