Add option to filter by regex

capacity-mode-evaluator/README.md

@@ -71,6 +71,7 @@ For more information on when to select on-demand or provisioned capacity modes i
    The following parameters are required to run the script:
 
    - `--dynamodb-tablename`: DynamoDB table name (optional; if not provided, the script will process all tables in the specified region)
+   - `--regex`: Enable regex pattern matching for DynamoDB table names (default: False)
    - `--dynamodb-read-utilization`: DynamoDB read utilization percentage (default: 70)
    - `--dynamodb-write-utilization`: DynamoDB write utilization percentage (default: 70)
    - `--dynamodb-minimum-write-unit`: DynamoDB minimum write unit (default: 1)
@@ -80,6 +81,7 @@ For more information on when to select on-demand or provisioned capacity modes i
    - `--number-of-days-look-back`: Number(1-14) of days to look back for CloudWatch metrics (default: 14)
    - `--max-concurrent-tasks`: Maximum number of tasks to run concurrently (default: 5)
    - `--show-dashboard`: Display results in a GUI for simple visualization (default: True when specified)
+
 
 - with default values:
 

capacity-mode-evaluator/capacity_reco.py

@@ -35,6 +35,7 @@ def get_params(args):
     params['number_of_days_look_back'] = args.number_of_days_look_back
     params['max_concurrent_tasks'] = args.max_concurrent_tasks
     params['show_dashboard'] = args.show_dashboard
+    params['regex'] = args.regex
 
     now = datetime.utcnow()
     midnight = datetime(now.year, now.month, now.day, 0, 0, 0, tzinfo=pytz.UTC)
@@ -117,6 +118,7 @@ def process_dynamodb_table(dynamodb_table_info: pd.DataFrame, params: dict, outp
     parser.add_argument('--max-concurrent-tasks', type=int,
                         default=5, help='Maximum number of tasks to run concurrently')
     parser.add_argument('--show-dashboard', action='store_true', help='Display results in a GUI for simple visualization')
+    parser.add_argument('--regex', action='store_true', help='Enable regex pattern matching for DynamoDB table names')
     args = parser.parse_args()
 
     timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
@@ -127,7 +129,7 @@ def process_dynamodb_table(dynamodb_table_info: pd.DataFrame, params: dict, outp
     logger.info(f"Parameters: {params}")
     DDBinfo = DDBScalingInfo()
     dynamo_tables_result = DDBinfo.get_all_dynamodb_autoscaling_settings_with_indexes(
-        params['dynamodb_tablename'], params['max_concurrent_tasks'])
+        params['dynamodb_tablename'], params['max_concurrent_tasks'], regex=params['regex'])
 
     dynamo_tables_result.to_csv(
         os.path.join(output_path, 'dynamodb_table_info.csv'), index=False)

capacity-mode-evaluator/src/dynamodb.py

@@ -4,6 +4,7 @@
 from tqdm import tqdm
 import boto3
 import logging
+import re
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -94,14 +95,15 @@ def _process_table(self, name):
             logger.error(f"Error processing table {name}: {e}")
             return pd.DataFrame(columns=columns)
 
-    def get_all_dynamodb_autoscaling_settings_with_indexes(self, table_name: str, max_concurrent_tasks: int) -> pd.DataFrame:
+    def get_all_dynamodb_autoscaling_settings_with_indexes(self, table_name: str, max_concurrent_tasks: int, regex: bool = False) -> pd.DataFrame:
 
         dynamodb_client = self.dynamodb_client
 
         # Get a list of all DynamoDB tables
         table_names = []
         last_evaluated_table_name = None
-        if not table_name:
+
+        if not table_name or regex:
             while last_evaluated_table_name != '':
                 params = {}
                 if last_evaluated_table_name:
@@ -110,9 +112,14 @@ def get_all_dynamodb_autoscaling_settings_with_indexes(self, table_name: str, ma
                 table_names += response['TableNames']
                 last_evaluated_table_name = response.get(
                     'LastEvaluatedTableName', '')
-
         else:
             table_names = [table_name]
+
+        if regex and table_name:
+            compiled_pattern = re.compile(table_name)
+            table_names = [name for name in table_names if compiled_pattern.search(name)]
+            logger.info(f"Found {len(table_names)} tables matching pattern '{table_name}'")
+
 
         settings_list = []
         if len(table_names) != 0: