linting, sigh

Author: benbart

ctorm/granule-md-db-loader/granule_md_db_loader.py

@@ -11,7 +11,8 @@
 from botocore.config import Config
 
 # Configure logging
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+fmt = "%(asctime)s [%(levelname)s] %(message)s"
+logging.basicConfig(level=logging.INFO, format=fmt)
 logger = logging.getLogger(__name__)
 
 
@@ -34,7 +35,8 @@ def limit(self, pk):
         self.history[pk] = [t for t in self.history[pk] if now - t < 1.0]
 
         if len(self.history[pk]) >= self.max_rate:
-            # Calculate sleep duration to let the oldest request roll off the 1-second window
+            # Calculate sleep duration to let the oldest request roll
+            # off the 1-second window
             sleep_time = 1.0 - (now - self.history[pk][0])
             if sleep_time > 0:
                 time.sleep(sleep_time)
@@ -44,6 +46,53 @@ def limit(self, pk):
         self.history[pk].append(now)
 
 
+def handle_file(
+        local_path,
+        table,
+        rate_limiter: PartitionRateLimiter,
+        records_in_file,
+        total_records,
+):
+    with gzip.open(local_path, "rt", encoding="utf-8") as f:
+        with table.batch_writer() as batch:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+
+                try:
+                    # Convert floats/doubles to Decimal for
+                    # DynamoDB compatibility
+                    item = json.loads(line, parse_float=Decimal)
+                except Exception as e:
+                    logger.error(f"Failed to parse JSON line: {e}")
+                    continue
+
+                # Add/modify fields if needed
+                newdate = datetime.datetime.utcnow().isoformat() + "Z"
+                item["imported_at"] = newdate
+
+                # Ensure partition and sort keys are present
+                pk = item.get("pk")
+                sk = item.get("sk")
+                if not pk or not sk:
+                    logger.warning(
+                        "Skipping record missing pk/sk: %s",
+                        item.get('granule_id'),
+                    )
+
+                    continue
+
+                # Apply dynamic rate limit based on the partition key
+                rate_limiter.limit(pk)
+
+                # Batch insert into DynamoDB
+                batch.put_item(Item=item)
+                records_in_file += 1
+                total_records += 1
+    return records_in_file, total_records
+
+
 def main():
     bucket_name = os.environ.get("CTORM_BUCKET", "ctorm-scratch")
     table_name = os.environ.get("TABLE_NAME")
@@ -55,25 +104,32 @@ def main():
 
     s3 = boto3.client("s3")
 
-    # Configure boto3 with more aggressive retries to gracefully handle scale peaks
+    # Configure boto3 with more aggressive retries to gracefully
+    # handle scale peaks
     retry_config = Config(
         retries={
             "max_attempts": 10,
-            "mode": "standard"
+            "mode": "standard",
         }
     )
     dynamodb = boto3.resource("dynamodb", config=retry_config)
-    table = dynamodb.Table(table_name)
+    dyndb_table = dynamodb.Table(table_name)
 
-    logger.info(f"Starting import from s3://{bucket_name}/{prefix} into DynamoDB table {table_name}")
+    logger.info(
+        "Starting import from s3://%s/%s into DynamoDB table %s",
+        bucket_name,
+        prefix,
+        table_name,
+    )
 
     paginator = s3.get_paginator("list_objects_v2")
     pages = paginator.paginate(Bucket=bucket_name, Prefix=prefix)
 
     total_files = 0
     total_records = 0
 
-    # Initialize partition-level rate limiter set to a safe threshold (850 writes/sec per pk)
+    # Initialize partition-level rate limiter set to a safe
+    # threshold (850 writes/sec per pk)
     rate_limiter = PartitionRateLimiter(max_rate_per_sec=850)
 
     # Walk through the bucket objects
@@ -94,52 +150,42 @@ def main():
             try:
                 s3.download_file(bucket_name, key, local_path)
             except Exception as e:
-                logger.error(f"Failed to download s3://{bucket_name}/{key}: {e}")
+                logger.error(
+                    "Failed to download s3://{%s}/%s: %s",
+                    bucket_name,
+                    key,
+                    e,
+                )
                 continue
 
             # Open, decompress and parse
             records_in_file = 0
             try:
-                with gzip.open(local_path, "rt", encoding="utf-8") as f:
-                    with table.batch_writer() as batch:
-                        for line in f:
-                            line = line.strip()
-                            if not line:
-                                continue
-
-                            try:
-                                # Convert floats/doubles to Decimal for DynamoDB compatibility
-                                item = json.loads(line, parse_float=Decimal)
-                            except Exception as e:
-                                logger.error(f"Failed to parse JSON line: {e}")
-                                continue
-
-                            # Add/modify fields if needed
-                            item["imported_at"] = datetime.datetime.utcnow().isoformat() + "Z"
-
-                            # Ensure partition and sort keys are present
-                            pk = item.get("pk")
-                            sk = item.get("sk")
-                            if not pk or not sk:
-                                logger.warning(f"Skipping record missing pk/sk: {item.get('granule_id')}")
-                                continue
-
-                            # Apply dynamic rate limit based on the partition key
-                            rate_limiter.limit(pk)
-
-                            # Batch insert into DynamoDB
-                            batch.put_item(Item=item)
-                            records_in_file += 1
-                            total_records += 1
-
-                logger.info(f"Successfully imported {records_in_file} records from {key}")
+                records_in_file, total_records = handle_file(
+                    local_path,
+                    dyndb_table,
+                    rate_limiter,
+                    records_in_file,
+                    total_records
+                )
+
+                logger.info(
+                    "Successfully imported %s records from %s",
+                    records_in_file,
+                    key,
+
+                )
             except Exception as e:
-                logger.error(f"Error processing file {key}: {e}")
+                logger.error("Error processing file %s: %s", key, e)
             finally:
                 if os.path.exists(local_path):
                     os.remove(local_path)
 
-    logger.info(f"Import complete. Processed {total_files} files, imported {total_records} total records.")
+    logger.info(
+        "Import complete. Processed %s files, imported %s total records.",
+        total_files,
+        total_records,
+    )
 
 
 if __name__ == "__main__":