fix lint errors

Author: Bob Strahan

lib/idp_common_pkg/idp_common/docs_service.py

@@ -9,8 +9,8 @@
 AppSync and DynamoDB implementations while maintaining the same interface.
 """
 
-import os
 import logging
+import os
 from typing import Optional, Union
 
 from idp_common.appsync import DocumentAppSyncService
@@ -30,37 +30,36 @@
 class DocumentServiceFactory:
     """
     Factory class for creating document services based on configuration.
-    
+
     This factory allows switching between AppSync and DynamoDB implementations
     while maintaining the same interface for document operations.
     """
-    
+
     @staticmethod
     def create_service(
-        mode: Optional[str] = None,
-        **kwargs
+        mode: Optional[str] = None, **kwargs
     ) -> Union[DocumentAppSyncService, DocumentDynamoDBService]:
         """
         Create a document service based on the specified mode.
-        
+
         Args:
             mode: Optional mode override. If not provided, uses DOCUMENT_TRACKING_MODE
                   environment variable, defaulting to 'appsync'
             **kwargs: Additional arguments passed to the service constructor
-            
+
         Returns:
             DocumentAppSyncService or DocumentDynamoDBService instance
-            
+
         Raises:
             ValueError: If an unsupported mode is specified
-            
+
         Examples:
             # Use environment variable (default behavior)
             service = DocumentServiceFactory.create_service()
-            
+
             # Override mode explicitly
             service = DocumentServiceFactory.create_service(mode='dynamodb')
-            
+
             # Pass additional arguments
             service = DocumentServiceFactory.create_service(
                 mode='appsync',
@@ -72,16 +71,16 @@ def create_service(
             mode = os.environ.get("DOCUMENT_TRACKING_MODE", DEFAULT_MODE).lower()
         else:
             mode = mode.lower()
-            
+
         # Validate mode
         if mode not in SUPPORTED_MODES:
             raise ValueError(
                 f"Unsupported document tracking mode: '{mode}'. "
                 f"Supported modes are: {', '.join(SUPPORTED_MODES)}"
             )
-        
+
         logger.info(f"Creating document service with mode: {mode}")
-        
+
         # Create the appropriate service
         if mode == APPSYNC_MODE:
             return DocumentAppSyncService(**kwargs)
@@ -90,32 +89,32 @@ def create_service(
         else:
             # This should never happen due to validation above, but included for completeness
             raise ValueError(f"Unsupported mode: {mode}")
-    
+
     @staticmethod
     def get_current_mode() -> str:
         """
         Get the current document tracking mode from environment variable.
-        
+
         Returns:
             Current mode string ('appsync' or 'dynamodb')
         """
         return os.environ.get("DOCUMENT_TRACKING_MODE", DEFAULT_MODE).lower()
-    
+
     @staticmethod
     def is_appsync_mode() -> bool:
         """
         Check if current mode is AppSync.
-        
+
         Returns:
             True if current mode is AppSync, False otherwise
         """
         return DocumentServiceFactory.get_current_mode() == APPSYNC_MODE
-    
+
     @staticmethod
     def is_dynamodb_mode() -> bool:
         """
         Check if current mode is DynamoDB.
-        
+
         Returns:
             True if current mode is DynamoDB, False otherwise
         """
@@ -124,29 +123,28 @@ def is_dynamodb_mode() -> bool:
 
 # Convenience function for creating services
 def create_document_service(
-    mode: Optional[str] = None,
-    **kwargs
+    mode: Optional[str] = None, **kwargs
 ) -> Union[DocumentAppSyncService, DocumentDynamoDBService]:
     """
     Convenience function to create a document service.
-    
+
     This is a shorthand for DocumentServiceFactory.create_service().
-    
+
     Args:
         mode: Optional mode override. If not provided, uses DOCUMENT_TRACKING_MODE
               environment variable, defaulting to 'appsync'
         **kwargs: Additional arguments passed to the service constructor
-        
+
     Returns:
         DocumentAppSyncService or DocumentDynamoDBService instance
-        
+
     Examples:
         # Simple usage
         service = create_document_service()
-        
+
         # With mode override
         service = create_document_service(mode='dynamodb')
-        
+
         # With additional parameters
         service = create_document_service(
             mode='appsync',

lib/idp_common_pkg/idp_common/dynamodb/client.py

@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Optional
 
 import boto3
-from botocore.exceptions import ClientError, BotoCoreError
+from botocore.exceptions import BotoCoreError, ClientError
 
 logger = logging.getLogger(__name__)
 
@@ -164,7 +164,9 @@ def get_item(self, key: Dict[str, Any]) -> Optional[Dict[str, Any]]:
             logger.error(f"BotoCore error during get_item: {str(e)}")
             raise DynamoDBError(f"BotoCore error: {str(e)}")
 
-    def transact_write_items(self, transact_items: List[Dict[str, Any]]) -> Dict[str, Any]:
+    def transact_write_items(
+        self, transact_items: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
         """
         Execute a transaction with multiple write operations.
 
@@ -193,12 +195,16 @@ def transact_write_items(self, transact_items: List[Dict[str, Any]]) -> Dict[str
             response = self.dynamodb.meta.client.transact_write_items(
                 TransactItems=processed_items
             )
-            logger.debug(f"Successfully executed transaction with {len(transact_items)} items")
+            logger.debug(
+                f"Successfully executed transaction with {len(transact_items)} items"
+            )
             return response
         except ClientError as e:
             error_code = e.response["Error"]["Code"]
             error_message = e.response["Error"]["Message"]
-            logger.error(f"DynamoDB transact_write_items failed: {error_code} - {error_message}")
+            logger.error(
+                f"DynamoDB transact_write_items failed: {error_code} - {error_message}"
+            )
             raise DynamoDBError(f"Transaction failed: {error_message}", error_code)
         except BotoCoreError as e:
             logger.error(f"BotoCore error during transact_write_items: {str(e)}")
@@ -247,7 +253,9 @@ def scan(
                 scan_params["ExclusiveStartKey"] = exclusive_start_key
 
             response = self.table.scan(**scan_params)
-            logger.debug(f"Successfully scanned table, returned {len(response.get('Items', []))} items")
+            logger.debug(
+                f"Successfully scanned table, returned {len(response.get('Items', []))} items"
+            )
             return response
         except ClientError as e:
             error_code = e.response["Error"]["Code"]
@@ -300,7 +308,9 @@ def query(
                 query_params["ExclusiveStartKey"] = exclusive_start_key
 
             response = self.table.query(**query_params)
-            logger.debug(f"Successfully queried table, returned {len(response.get('Items', []))} items")
+            logger.debug(
+                f"Successfully queried table, returned {len(response.get('Items', []))} items"
+            )
             return response
         except ClientError as e:
             error_code = e.response["Error"]["Code"]