Add support for prefix/suffix filters on S3

kappa/event_source/s3.py

@@ -37,10 +37,20 @@ def add(self, function):
                 {
                     'Id': self._make_notification_id(function.name),
                     'Events': [e for e in self._config['events']],
-                    'LambdaFunctionArn': function.arn,
+                    'LambdaFunctionArn': '%s:%s' % (function.arn, function._context.environment),
                 }
             ]
         }
+
+        # Add S3 key filters
+        if 'key_filters' in self._config:
+            filters_spec = { 'Key' : { 'FilterRules' : [] } }
+            for filter in self._config['key_filters']:
+                if 'type' in filter and 'value' in filter and filter['type'] in ('prefix', 'suffix'):
+                    rule = { 'Name' : filter['type'], 'Value' : filter['value'] }
+                    filters_spec['Key']['FilterRules'].append(rule)
+            notification_spec['LambdaFunctionConfigurations'][0]['Filter'] = filters_spec
+
         try:
             response = self._s3.call(
                 'put_bucket_notification_configuration',

samples/s3/config.yml

@@ -34,3 +34,8 @@ lambda:
       arn: arn:aws:s3:::test-1245812163
       events:
         - s3:ObjectCreated:*
+      key_filters:
+        - type: prefix
+          value: someprefix/
+        - type: suffix
+          value: .zip

setup.py

@@ -24,7 +24,7 @@ def run_setup():
         author='Mitch Garnaat',
         author_email='mitch@garnaat.com',
         license='Apache License 2.0',
-        packages=['kappa', 'kappa.scripts'],
+        packages=['kappa', 'kappa.scripts', 'kappa.event_source'],
         package_data={'kappa': ['_version']},
         package_dir={'kappa': 'kappa'},
         entry_points="""