chore: some type updates (#473)

Some relevant decisions:
* `timeout` passed to `TorngitBaseAdapter.__init__` is transformed into `httpx.Timeout` directly,
(so that `self._timeout` can be directly an `httpx.Timeout` object instead a bunch of different options
* `verify_ssl` receives the default value of `True`. That shouldn't break anyone except BB server.
(but no one is using it, so as I said it shouldn't break anyone)
* Better differentiation of aouth_consumer_token and token. They are actually different things.

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[Any]:
+    def kwargs_keys_to_log(self) -> List[str]:
         """List of args from the function to be logged (if present)"""
         return self.get("kwargs_keys_to_log", [])
 

shared/torngit/base.py

@@ -1,6 +1,5 @@
 import re
 from enum import Enum
-from typing import Dict, List, Optional, Tuple
 
 import httpx
 
@@ -13,7 +12,6 @@
     OnRefreshCallback,
     Token,
 )
-from shared.typings.torngit import TorngitInstanceData
 
 get_start_of_line = re.compile(r"@@ \-(\d+),?(\d*) \+(\d+),?(\d*).*").match
 
@@ -28,66 +26,67 @@ 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 = None
-    _on_token_refresh: OnRefreshCallback = None
+    _oauth: OauthConsumerToken | None
+    _on_token_refresh: OnRefreshCallback | None
     _token: Token | None = None
-    verify_ssl = None
+    verify_ssl: str | bool
+    _timeout: httpx.Timeout
+    valid_languages = set(language[0] for language in Repository.Languages.choices)
 
-    valid_languages = set(language.value for language in Repository.Languages)
+    # These are set by the subclasses
+    service_url: str | None = None
+    service: str | None = None
+    urls: dict[str, str] | None = None
 
     def __init__(
         self,
         oauth_consumer_token: OauthConsumerToken | None = None,
-        timeouts=None,
+        timeout: httpx.Timeout | tuple[int, int] | None = None,
         token: Token | None = None,
         token_type_mapping: TokenTypeMapping | None = None,
-        on_token_refresh: OnRefreshCallback = None,
-        verify_ssl=None,
+        on_token_refresh: OnRefreshCallback | None = None,
+        verify_ssl: str | bool = True,
         **kwargs,
-    ):
-        self._timeouts = timeouts or [10, 30]
+    ) -> 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._token = token
         self._on_token_refresh = on_token_refresh
         self._token_type_mapping = token_type_mapping or {}
         self._oauth = oauth_consumer_token
-        self.data: TorngitInstanceData = {
-            "owner": {},
-            "repo": {},
-            "fallback_installations": None,
-            "installation": None,
-            "additional_data": {},
-        }
         self.verify_ssl = verify_ssl
-        self.data.update(kwargs)
+        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", {}),
+        }
         # This has the side effect of initializing the torngit_cache
         # (if not yet initialized)
         torngit_cache.initialize()
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         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, 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])
+    def get_client(self, timeout: httpx.Timeout | None = None) -> httpx.AsyncClient:
+        if timeout is None:
+            timeout = self._timeout
         return httpx.AsyncClient(
-            verify=(
-                self.verify_ssl
-                if not isinstance(self.verify_ssl, bool)
-                else self.verify_ssl
-            ),
+            verify=self.verify_ssl,
             timeout=timeout,
         )
 
@@ -96,7 +95,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: Optional[str], token_type: TokenType):
+    def get_token_by_type_if_none(self, token: str | None, token_type: TokenType):
         if token is not None:
             return token
         return self.get_token_by_type(token_type)
@@ -106,14 +105,15 @@ def _oauth_consumer_token(self) -> OauthConsumerToken:
             raise Exception("Oauth consumer token not present")
         return self._oauth
 
-    def _validate_language(self, language: str) -> str | None:
+    def _validate_language(self, language: str | None) -> str | None:
         if language:
             language = language.lower()
+
             if language in self.valid_languages:
                 return language
         return None
 
-    def set_token(self, token: OauthConsumerToken) -> None:
+    def set_token(self, token: Token) -> 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, token=None):
+    async def get_branch(self, branch_name: str, 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) -> dict:
+    def get_ttl(self, endpoint: CachedEndpoint) -> int:
         return self.ttls.get(endpoint, 120)
 
 

shared/typings/oauth_token_types.py

@@ -1,8 +1,9 @@
-from typing import Awaitable, Callable, Optional, TypedDict
+from typing import Awaitable, Callable, NotRequired, 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
@@ -13,9 +14,9 @@ class Token(TypedDict):
     entity_name: str | None
 
 
-class OauthConsumerToken(Token):
-    secret: Optional[str]
-    refresh_token: Optional[str]
+class OauthConsumerToken(TypedDict):
+    key: str  # client_id
+    secret: str
 
 
-OnRefreshCallback = Optional[Callable[[OauthConsumerToken], Awaitable[None]]]
+OnRefreshCallback = Callable[[Token], Awaitable[None]]

shared/typings/torngit.py

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