Add support for retrieving dynamodb resources

plumbum.py

@@ -170,6 +170,12 @@ def list_kinesis_applications(region, filter_by_kwargs):
         kinesis_streams[stream_name] = shard_ids
     return kinesis_streams
 
+def list_dynamodb(region, filter_by_kwargs):
+    """List all DynamoDB tables."""
+    conn = boto.dynamodb.connect_to_region(region)
+    tables = conn.list_tables()
+    return lookup(tables, filter_by=filter_by_kwargs)
+
 list_resources = {
     'ec2': list_ec2,
     'elb': list_elb,
@@ -178,6 +184,7 @@ def list_kinesis_applications(region, filter_by_kwargs):
     'asg': list_autoscaling_group,
     'sqs': list_sqs,
     'kinesisapp': list_kinesis_applications,
+    'dynamodb': list_dynamodb
 }
 
 

sample_templates/dynamodb.yml.j2

@@ -0,0 +1,22 @@
+{%- set metrics = {'ConsumedReadCapacityUnits': {'stat': ['Minimum', 'Maximum', 'Sum', 'Average'], 'unit': 'Count'},
+                   'ConsumedWriteCapacityUnits': {'stat': ['Minimum', 'Maximum', 'Sum', 'Average'], 'unit': 'Count'},
+                   } -%}
+
+# If connecting to a different region other than default, set region
+Auth:
+  region: "{{ region }}"
+Metrics:
+{%- for table_name in resources %}
+  {%- for metric in metrics %}
+- Namespace: "AWS/DynamoDB"
+  MetricName: "{{ metric }}"
+  Statistics: "{{ metrics[metric]['stat'] }}"
+  Unit: "{{ metrics[metric]['unit'] }}"
+  Dimensions:
+    TableName: "{{ table_name }}"
+  Options:
+    Formatter: 'cloudwatch.%(Namespace)s.{{ stream_name}}.%(MetricName)s.%(statistic)s.%(Unit)s'
+    Period: 1
+  {% endfor %}
+{% endfor %}
+

test_plumbum.py

@@ -71,6 +71,12 @@ def test_filters_and_extras_with_region_specified(self):
         self.assertEqual(len(filter_by), 1)
         self.assertEqual(len(extras), 2)
 
+    def test_dynamodb_namespace(self):
+        argv = ['foo.j2', 'dynamodb', 'us-east-1']
+        templ, ns, region, filter_by, extras = plumbum.interpret_options(argv)
+        self.assertEqual(templ, 'foo.j2')
+        self.assertEqual(ns, 'dynamodb')
+        self.assertEqual(region, 'us-east-1')
 
 if __name__ == '__main__':
     unittest.main()