refactor: fixup lint/types

scripts/aws/aws_credential_source.py

@@ -1,8 +1,8 @@
-
 from dataclasses import dataclass
 from typing import Optional
 
 
+# pylint: disable=too-many-instance-attributes
 @dataclass
 class CredentialSource:
     bucket: str
@@ -32,4 +32,4 @@ class CredentialSource:
     flags: Optional[str] = None
     """
     flags that the role can use either "r" for read-only or "rw" for read-write
-    """
+    """

scripts/aws/aws_helper.py

@@ -1,6 +1,6 @@
 import json
 from os import environ
-from typing import Dict, List, NamedTuple, Optional
+from typing import Any, Dict, List, NamedTuple, Optional
 from urllib.parse import urlparse
 
 import boto3
@@ -19,10 +19,9 @@
 
 bucket_roles: List[CredentialSource] = []
 
-bucket_credentials = {}
 client_sts = session.client("sts")
 
-bucket_config_path ="s3://linz-bucket-config/config-v2.json"
+bucket_config_path = "s3://linz-bucket-config/config-v2.json"
 
 
 # Load bucket to roleArn mapping for LINZ internal buckets from SSM
@@ -41,12 +40,13 @@ def _init_roles() -> None:
     get_log().debug("bucket_config_loaded", config=bucket_config_path, prefix_count=len(bucket_roles))
 
 
-def _get_client_creator(session):
-    def client_creator(service_name, **kwargs):
-        return session.client(service_name, **kwargs)
+def _get_client_creator(local_session: boto3.Session) -> Any:
+    def client_creator(service_name: str, **kwargs: Any) -> Any:
+        return local_session.client(service_name, **kwargs)
 
     return client_creator
 
+
 def get_session(prefix: str) -> boto3.Session:
     cfg = _get_credential_config(prefix)
     if cfg is None:
@@ -56,18 +56,20 @@ def get_session(prefix: str) -> boto3.Session:
     if current_session is not None:
         return current_session
 
-    extra_args = {"DurationSeconds": cfg.roleSessionDuration}
+    extra_args: Dict[str, Any] = {"DurationSeconds": cfg.roleSessionDuration}
 
     if cfg.externalId:
         extra_args["ExternalId"] = cfg.externalId
-        
+
     fetcher = AssumeRoleCredentialFetcher(
         client_creator=_get_client_creator(session),
         source_credentials=session.get_credentials(),
         role_arn=cfg.roleArn,
-        extra_args=extra_args
+        extra_args=extra_args,
     )
     botocore_session = botocore.session.Session()
+
+    # pylint:disable=protected-access
     botocore_session._credentials = DeferredRefreshableCredentials(
         method="assume-role", refresh_using=fetcher.fetch_credentials
     )

scripts/files/fs_s3.py

@@ -1,4 +1,3 @@
-
 import boto3
 import botocore
 from linz_logger import get_log

scripts/gdal/gdal_helper.py

@@ -4,8 +4,7 @@
 
 from linz_logger import get_log
 
-from scripts.aws.aws_helper import is_s3
-from scripts.files.fs_s3 import get_session
+from scripts.aws.aws_helper import get_session, is_s3
 from scripts.logging.time_helper import time_in_ms
 
 
@@ -61,7 +60,7 @@ def run_gdal(
         if is_s3(input_file):
             # Set the credentials for GDAL to be able to read the input file
             session = get_session(input_file)
-            credentials= session.get_credentials()
+            credentials = session.get_credentials()
             gdal_env["AWS_ACCESS_KEY_ID"] = credentials.access_key
             gdal_env["AWS_SECRET_ACCESS_KEY"] = credentials.secret_key
             gdal_env["AWS_SESSION_TOKEN"] = credentials.token