Update typing

Author: reweeden

src/rain_api_core/auth.py

@@ -1,9 +1,10 @@
 import contextlib
 import dataclasses
 import logging
+from collections.abc import Mapping
 from http.cookies import CookieError, SimpleCookie
 from time import time
-from typing import List, Mapping, Optional
+from typing import Optional
 from wsgiref.handlers import format_date_time as format_7231_date
 
 import jwt
@@ -15,12 +16,12 @@
 class UserProfile:
     user_id: str
     token: str
-    groups: List[str]
+    groups: list[str]
     first_name: str
     last_name: str
     email: str
-    iat: int = None
-    exp: int = None
+    iat: Optional[int] = None
+    exp: Optional[int] = None
 
     @classmethod
     def from_jwt_payload(cls, payload):
@@ -115,7 +116,9 @@ def _in_blacklist(self, user_profile: UserProfile):
                 return True
         return False
 
-    def get_profile_from_headers(self, headers) -> Optional[UserProfile]:
+    def get_profile_from_headers(
+        self, headers: Mapping[str, str],
+    ) -> Optional[UserProfile]:
         """Inspects headers for auth cookie and return user_profile if authenticated, None otherwise"""
         auth_cookie = self._get_auth_cookie(headers)
         if not auth_cookie:
@@ -130,7 +133,7 @@ def get_profile_from_headers(self, headers) -> Optional[UserProfile]:
             return None
         return user_profile
 
-    def get_header_to_set_auth_cookie(self, user_profile: Optional[UserProfile], cookie_domain=''):
+    def get_header_to_set_auth_cookie(self, user_profile: Optional[UserProfile], cookie_domain: str = ''):
         """ Gets a header to set auth-cookie
 
         Parameters:

src/rain_api_core/aws_util.py

@@ -155,7 +155,7 @@ def get_yaml_file(bucket: str, key: str) -> dict:
     return get_yaml(bucket, key)
 
 
-def get_role_creds(user_id: str = None, in_region: bool = False):
+def get_role_creds(user_id: Optional[str] = None, in_region: bool = False):
     """
     :param user_id: string with URS username
     :param in_region: boolean If True a download role that works only in region will be returned
@@ -205,7 +205,10 @@ def get_role_creds(user_id: str = None, in_region: bool = False):
     return role_creds_cache[download_role_arn][user_id]["session"], session_offset
 
 
-def get_role_session(creds: dict = None, user_id: str = None) -> boto_Session:
+def get_role_session(
+    creds: Optional[dict] = None,
+    user_id: Optional[str] = None,
+) -> boto_Session:
     global session_cache  # pylint: disable=global-statement
     sts_resp = creds if creds else get_role_creds(user_id)[0]
     log.debug('sts_resp: {0}'.format(sts_resp))

src/rain_api_core/bucket_map.py

@@ -1,15 +1,16 @@
 from collections import defaultdict
+from collections.abc import Generator, Iterable, Sequence
 from dataclasses import dataclass, field
-from typing import Generator, Iterable, Optional, Sequence, Tuple
+from typing import Optional
 
 # By default, buckets are accessible to any logged in users. This is
 # represented by an empty set.
 _DEFAULT_PERMISSION_FACTORY = set
 
 
 def _is_accessible(
-    required_groups: Optional[set],
-    groups: Optional[Iterable[str]]
+    required_groups: Optional[set[str]],
+    groups: Optional[Iterable[str]],
 ) -> bool:
     # Check for public access
     if required_groups is None:
@@ -30,7 +31,7 @@ class BucketMapEntry():
     headers: dict = field(default_factory=dict)
     _access_control: Optional[dict] = None
 
-    def is_accessible(self, groups: Iterable[str] = None) -> bool:
+    def is_accessible(self, groups: Optional[Iterable[str]] = None) -> bool:
         """Check if the object is accessible with the given permissions.
 
         Setting `groups` to an iterable implies that the user has logged in,
@@ -41,7 +42,7 @@ def is_accessible(self, groups: Iterable[str] = None) -> bool:
         required_groups = self.get_required_groups()
         return _is_accessible(required_groups, groups)
 
-    def get_required_groups(self) -> Optional[set]:
+    def get_required_groups(self) -> Optional[set[str]]:
         """Get a set of permissions protecting this object.
 
         It is sufficient to have one of the permissions in the set in order to
@@ -126,7 +127,7 @@ def get_path(self, path: Sequence[str]) -> Optional[BucketMapEntry]:
 
         return None
 
-    def entries(self):
+    def entries(self) -> Generator[BucketMapEntry]:
         for bucket, path_parts, headers in _walk_entries(self._get_map()):
             yield self._make_entry(
                 bucket=bucket,
@@ -135,7 +136,7 @@ def entries(self):
                 headers=headers
             )
 
-    def to_iam_policy(self, groups: Iterable[str] = None) -> dict:
+    def to_iam_policy(self, groups: Optional[Iterable[str]] = None) -> Optional[dict]:
         if not self._iam_compatible:
             _check_iam_compatible(self.access_control)
         generator = IamPolicyGenerator(groups)
@@ -150,8 +151,8 @@ def _make_entry(
         bucket: str,
         bucket_path: str,
         object_key: str,
-        headers: Optional[dict] = None
-    ):
+        headers: Optional[dict] = None,
+    ) -> BucketMapEntry:
         return BucketMapEntry(
             bucket=self.bucket_name_prefix + bucket,
             bucket_path=bucket_path,
@@ -164,7 +165,7 @@ def _make_entry(
         )
 
 
-def _walk_entries(node: dict, path=()) -> Generator[Tuple[str, tuple, Optional[dict]], None, None]:
+def _walk_entries(node: dict, path=()) -> Generator[tuple[str, tuple, Optional[dict]]]:
     """A generator to recursively yield all leaves of a bucket map"""
 
     for key, val in node.items():
@@ -294,7 +295,7 @@ def _access_text(access) -> str:
 
 
 class IamPolicyGenerator:
-    def __init__(self, groups: Iterable[str]):
+    def __init__(self, groups: Optional[Iterable[str]]):
         self.groups = groups
 
     def _is_accessible(self, required_groups: Optional[set]) -> bool:

src/rain_api_core/timer.py

@@ -1,7 +1,8 @@
 import logging
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable, Optional
+from typing import Optional
 
 
 @dataclass(eq=False)
@@ -28,7 +29,7 @@ def __init__(self, timer: Callable[[], float] = time.time):
         self.last_name: Optional[str] = None
         self.total = Interval()
 
-    def mark(self, name: str = None) -> float:
+    def mark(self, name: Optional[str] = None) -> float:
         """Record a new event.
 
         If called without `name`, any previously started event will be marked

src/rain_api_core/urs_util.py

@@ -10,7 +10,7 @@
 log = logging.getLogger(__name__)
 
 
-def get_base_url(ctxt: dict = None) -> str:
+def get_base_url(ctxt: Optional[dict] = None) -> str:
     # Make a redirect url using optional custom domain_name, otherwise use raw domain/stage provided by API Gateway.
     try:
         domain = os.getenv('DOMAIN_NAME') or f"{ctxt['domainName']}/{ctxt['stage']}"
@@ -20,7 +20,7 @@ def get_base_url(ctxt: dict = None) -> str:
         raise
 
 
-def get_redirect_url(ctxt: dict = None) -> str:
+def get_redirect_url(ctxt: Optional[dict] = None) -> str:
     return f'{get_base_url(ctxt)}login'
 
 
@@ -49,7 +49,7 @@ def do_auth(code: str, redirect_url: str, aux_headers: dict = {}) -> dict:
         return {}
 
 
-def get_urs_url(ctxt: dict, to: str = None) -> str:
+def get_urs_url(ctxt: dict, to: Optional[str] = None) -> str:
     base_url = os.getenv('AUTH_BASE_URL', 'https://urs.earthdata.nasa.gov') + '/oauth/authorize'
 
     # From URS Application
@@ -93,7 +93,7 @@ def get_user_profile(urs_user_payload: dict, access_token) -> UserProfile:
 def get_profile(
     user_id: str,
     token: str,
-    temptoken: str = None,
+    temptoken: Optional[str] = None,
     aux_headers: dict = {},
 ) -> Optional[UserProfile]:
     if not user_id or not token: