WIP commit on significant refactoring of code.

@@ -14,33 +14,8 @@
 import logging
 import logging
 
 
 import click
 import click
-import yaml
 
 
-from kappa import Kappa
-
-FmtString = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-
-
-def set_debug_logger(logger_names=['kappa'], stream=None):
-    """
-    Convenience function to quickly configure full debug output
-    to go to the console.
-    """
-    for logger_name in logger_names:
-        log = logging.getLogger(logger_name)
-        log.setLevel(logging.DEBUG)
-
-    ch = logging.StreamHandler(stream)
-    ch.setLevel(logging.DEBUG)
-
-    # create formatter
-    formatter = logging.Formatter(FmtString)
-
-    # add formatter to ch
-    ch.setFormatter(formatter)
-
-    # add ch to logger
-    log.addHandler(ch)
+from kappa.context import Context
 
 
 
 
 @click.command()
 @click.command()
@@ -62,27 +37,26 @@ def set_debug_logger(logger_names=['kappa'], stream=None):
     type=click.Choice(['deploy', 'test', 'tail', 'add-event-source', 'delete'])
     type=click.Choice(['deploy', 'test', 'tail', 'add-event-source', 'delete'])
 )
 )
 def main(config=None, debug=False, command=None):
 def main(config=None, debug=False, command=None):
-    if debug:
-        set_debug_logger()
-    config = yaml.load(config)
-    kappa = Kappa(config)
+    ctx = Context(config, debug)
     if command == 'deploy':
     if command == 'deploy':
         click.echo('Deploying ...')
         click.echo('Deploying ...')
-        kappa.deploy()
+        ctx.deploy()
         click.echo('...done')
         click.echo('...done')
     elif command == 'test':
     elif command == 'test':
         click.echo('Sending test data ...')
         click.echo('Sending test data ...')
-        kappa.test()
+        ctx.test()
         click.echo('...done')
         click.echo('...done')
     elif command == 'tail':
     elif command == 'tail':
-        kappa.tail()
+        events = ctx.tail()
+        for event in events:
+            print(event['message'])
     elif command == 'delete':
     elif command == 'delete':
         click.echo('Deleting ...')
         click.echo('Deleting ...')
-        kappa.delete()
+        ctx.delete()
         click.echo('...done')
         click.echo('...done')
     elif command == 'add-event-source':
     elif command == 'add-event-source':
         click.echo('Adding event source ...')
         click.echo('Adding event source ...')
-        kappa.add_event_source()
+        ctx.add_event_source()
         click.echo('...done')
         click.echo('...done')
 
 
 
 

@@ -11,233 +11,6 @@
 # ANY KIND, either express or implied. See the License for the specific
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 # language governing permissions and limitations under the License.
 
 
-import logging
 import os
 import os
-import zipfile
-import time
 
 
-import botocore.session
-from botocore.exceptions import ClientError
-
-LOG = logging.getLogger(__name__)
-
-
-class Kappa(object):
-
-    completed_states = ('CREATE_COMPLETE', 'UPDATE_COMPLETE')
-
-    def __init__(self, config):
-        self.config = config
-        self.session = botocore.session.get_session()
-        self.session.profile = config['profile']
-        self.region = config['region']
-
-    def create_update_roles(self, stack_name, roles_path):
-        LOG.debug('create_update_policies: stack_name=%s', stack_name)
-        LOG.debug('create_update_policies: roles_path=%s', roles_path)
-        cfn = self.session.create_client('cloudformation', self.region)
-        # Does stack already exist?
-        try:
-            response = cfn.describe_stacks(StackName=stack_name)
-            LOG.debug('Stack %s already exists', stack_name)
-        except ClientError:
-            LOG.debug('Stack %s does not exist', stack_name)
-            response = None
-        template_body = open(roles_path).read()
-        if response:
-            try:
-                cfn.update_stack(
-                    StackName=stack_name, TemplateBody=template_body,
-                    Capabilities=['CAPABILITY_IAM'])
-            except ClientError, e:
-                LOG.debug(str(e))
-        else:
-            response = cfn.create_stack(
-                StackName=stack_name, TemplateBody=template_body,
-                Capabilities=['CAPABILITY_IAM'])
-        done = False
-        while not done:
-            time.sleep(1)
-            response = cfn.describe_stacks(StackName=stack_name)
-            status = response['Stacks'][0]['StackStatus']
-            LOG.debug('Stack status is: %s', status)
-            if status in self.completed_states:
-                done = True
-
-    def get_role_arn(self, role_name):
-        role_arn = None
-        cfn = self.session.create_client('cloudformation', self.region)
-        try:
-            resources = cfn.list_stack_resources(
-                StackName=self.config['cloudformation']['stack_name'])
-        except Exception:
-            LOG.exception('Unable to find role ARN: %s', role_name)
-        for resource in resources['StackResourceSummaries']:
-            if resource['LogicalResourceId'] == role_name:
-                iam = self.session.create_client('iam')
-                role = iam.get_role(RoleName=resource['PhysicalResourceId'])
-                role_arn = role['Role']['Arn']
-        LOG.debug('role_arn: %s', role_arn)
-        return role_arn
-
-    def delete_roles(self, stack_name):
-        LOG.debug('delete_roles: stack_name=%s', stack_name)
-        cfn = self.session.create_client('cloudformation', self.region)
-        try:
-            cfn.delete_stack(StackName=stack_name)
-        except Exception:
-            LOG.exception('Unable to delete stack: %s', stack_name)
-
-    def _zip_lambda_dir(self, zipfile_name, lambda_dir):
-        LOG.debug('_zip_lambda_dir: lambda_dir=%s', lambda_dir)
-        LOG.debug('zipfile_name=%s', zipfile_name)
-        relroot = os.path.abspath(lambda_dir)
-        with zipfile.ZipFile(zipfile_name, 'w') as zf:
-            for root, dirs, files in os.walk(lambda_dir):
-                zf.write(root, os.path.relpath(root, relroot))
-                for file in files:
-                    filename = os.path.join(root, file)
-                    if os.path.isfile(filename):
-                        arcname = os.path.join(
-                            os.path.relpath(root, relroot), file)
-                        zf.write(filename, arcname)
-
-    def _zip_lambda_file(self, zipfile_name, lambda_file):
-        LOG.debug('_zip_lambda_file: lambda_file=%s', lambda_file)
-        LOG.debug('zipfile_name=%s', zipfile_name)
-        with zipfile.ZipFile(zipfile_name, 'w') as zf:
-            zf.write(lambda_file)
-
-    def zip_lambda_function(self, zipfile_name, lambda_fn):
-        if os.path.isdir(lambda_fn):
-            self._zip_lambda_dir(zipfile_name, lambda_fn)
-        else:
-            self._zip_lambda_file(zipfile_name, lambda_fn)
-
-    def upload_lambda_function(self, zip_file):
-        LOG.debug('uploading %s', zip_file)
-        lambda_svc = self.session.create_client('lambda', self.region)
-        with open(zip_file, 'rb') as fp:
-            exec_role = self.get_role_arn(
-                self.config['cloudformation']['exec_role'])
-            try:
-                response = lambda_svc.upload_function(
-                    FunctionName=self.config['lambda']['name'],
-                    FunctionZip=fp,
-                    Runtime=self.config['lambda']['runtime'],
-                    Role=exec_role,
-                    Handler=self.config['lambda']['handler'],
-                    Mode=self.config['lambda']['mode'],
-                    Description=self.config['lambda']['description'],
-                    Timeout=self.config['lambda']['timeout'],
-                    MemorySize=self.config['lambda']['memory_size'])
-                LOG.debug(response)
-            except Exception:
-                LOG.exception('Unable to upload zip file')
-
-    def delete_lambda_function(self, function_name):
-        LOG.debug('deleting function %s', function_name)
-        lambda_svc = self.session.create_client('lambda', self.region)
-        response = lambda_svc.delete_function(FunctionName=function_name)
-        LOG.debug(response)
-        return response
-
-    def _invoke_asynch(self, data_file):
-        LOG.debug('_invoke_async %s', data_file)
-        with open(data_file) as fp:
-            lambda_svc = self.session.create_client('lambda', self.region)
-            response = lambda_svc.invoke_async(
-                FunctionName=self.config['lambda']['name'],
-                InvokeArgs=fp)
-            LOG.debug(response)
-
-    def _tail(self, function_name):
-        LOG.debug('tailing function: %s', function_name)
-        log_svc = self.session.create_client('logs', self.region)
-        # kinda kludgy but can't find any way to get log group name
-        log_group_name = '/aws/lambda/%s' % function_name
-        latest_stream = None
-        response = log_svc.describe_log_streams(logGroupName=log_group_name)
-        # The streams are not ordered by time, hence this ugliness
-        for stream in response['logStreams']:
-            if not latest_stream:
-                latest_stream = stream
-            elif stream['lastEventTimestamp'] > latest_stream['lastEventTimestamp']:
-                latest_stream = stream
-        response = log_svc.get_log_events(
-            logGroupName=log_group_name,
-            logStreamName=latest_stream['logStreamName'])
-        for log_event in response['events']:
-            print(log_event['message'])
-
-    def _get_function_arn(self):
-        name = self.config['lambda']['name']
-        arn = None
-        lambda_svc = self.session.create_client('lambda', self.region)
-        try:
-            response = lambda_svc.get_function_configuration(
-                FunctionName=name)
-            LOG.debug(response)
-            arn = response['FunctionARN']
-        except Exception:
-            LOG.debug('Unable to find ARN for function: %s' % name)
-        return arn
-
-    def _add_kinesis_event_source(self, event_source_arn):
-        lambda_svc = self.session.create_client('lambda', self.region)
-        try:
-            invoke_role = self.get_role_arn(
-                self.config['cloudformation']['invoke_role'])
-            response = lambda_svc.add_event_source(
-                FunctionName=self.config['lambda']['name'],
-                Role=invoke_role,
-                EventSource=event_source_arn,
-                BatchSize=self.config['lambda'].get('batch_size', 100))
-            LOG.debug(response)
-        except Exception:
-            LOG.exception('Unable to add event source')
-
-    def _add_s3_event_source(self, event_source_arn):
-        s3_svc = self.session.create_client('s3', self.region)
-        bucket_name = event_source_arn.split(':')[-1]
-        invoke_role = self.get_role_arn(
-            self.config['cloudformation']['invoke_role'])
-        notification_spec = {
-            'CloudFunctionConfiguration': {
-                'Id': 'Kappa-%s-notification' % self.config['lambda']['name'],
-                'Events': [e for e in self.config['lambda']['s3_events']],
-                'CloudFunction': self._get_function_arn(),
-                'InvocationRole': invoke_role}}
-        response = s3_svc.put_bucket_notification(
-            Bucket=bucket_name,
-            NotificationConfiguration=notification_spec)
-        LOG.debug(response)
-
-    def add_event_source(self):
-        event_source_arn = self.config['lambda']['event_source']
-        _, _, svc, _ = event_source_arn.split(':', 3)
-        if svc == 'kinesis':
-            self._add_kinesis_event_source(event_source_arn)
-        elif svc == 's3':
-            self._add_s3_event_source(event_source_arn)
-        else:
-            raise ValueError('Unsupported event source: %s' % event_source_arn)
-
-    def deploy(self):
-        self.create_update_roles(
-            self.config['cloudformation']['stack_name'],
-            self.config['cloudformation']['template'])
-        self.zip_lambda_function(
-            self.config['lambda']['zipfile_name'],
-            self.config['lambda']['path'])
-        self.upload_lambda_function(self.config['lambda']['zipfile_name'])
-
-    def test(self):
-        self._invoke_asynch(self.config['lambda']['test_data'])
-
-    def tail(self):
-        self._tail(self.config['lambda']['name'])
-
-    def delete(self):
-        self.delete_roles(self.config['cloudformation']['stack_name'])
-        self.delete_lambda_function(self.config['lambda']['name'])
+__version__ = open(os.path.join(os.path.dirname(__file__), '_version')).read()

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import botocore.session
+
+
+class __AWS(object):
+
+    def __init__(self, profile=None, region=None):
+        self._client_cache = {}
+        self._session = botocore.session.get_session()
+        self._session.profile = profile
+        self._region = region
+
+    def create_client(self, client_name):
+        if client_name not in self._client_cache:
+            self._client_cache[client_name] = self._session.create_client(
+                client_name, self._region)
+        return self._client_cache[client_name]
+
+
+__Singleton_AWS = None
+
+
+def get_aws(context):
+    global __Singleton_AWS
+    if __Singleton_AWS is None:
+        __Singleton_AWS = __AWS(context.profile, context.region)
+    return __Singleton_AWS

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import logging
+import yaml
+
+import kappa.function
+import kappa.event_source
+import kappa.stack
+
+LOG = logging.getLogger(__name__)
+
+DebugFmtString = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+InfoFmtString = '\t%(message)s'
+
+
+class Context(object):
+
+    def __init__(self, config_file, debug=False):
+        if debug:
+            self.set_logger('kappa', logging.DEBUG)
+        else:
+            self.set_logger('kappa', logging.INFO)
+        self.config = yaml.load(config_file)
+        self._stack = kappa.stack.Stack(
+            self, self.config['cloudformation'])
+        self.function = kappa.function.Function(
+            self, self.config['lambda'])
+        self.event_sources = []
+        self._create_event_sources()
+
+    @property
+    def profile(self):
+        return self.config.get('profile', None)
+
+    @property
+    def region(self):
+        return self.config.get('region', None)
+
+    @property
+    def cfn_config(self):
+        return self.config.get('cloudformation', None)
+
+    @property
+    def lambda_config(self):
+        return self.config.get('lambda', None)
+
+    @property
+    def exec_role_arn(self):
+        return self._stack.invoke_role_arn
+
+    @property
+    def invoke_role_arn(self):
+        return self._stack.invoke_role_arn
+
+    def debug(self):
+        self.set_logger('kappa', logging.DEBUG)
+
+    def set_logger(self, logger_name, level=logging.INFO):
+        """
+        Convenience function to quickly configure full debug output
+        to go to the console.
+        """
+        log = logging.getLogger(logger_name)
+        log.setLevel(level)
+
+        ch = logging.StreamHandler(None)
+        ch.setLevel(level)
+
+        # create formatter
+        if level == logging.INFO:
+            formatter = logging.Formatter(InfoFmtString)
+        else:
+            formatter = logging.Formatter(DebugFmtString)
+
+        # add formatter to ch
+        ch.setFormatter(formatter)
+
+        # add ch to logger
+        log.addHandler(ch)
+
+    def _create_event_sources(self):
+        for event_source_cfg in self.config['lambda']['event_sources']:
+            _, _, svc, _ = event_source_cfg['arn'].split(':', 3)
+        if svc == 'kinesis':
+            self.event_sources.append(kappa.event_source.KinesisEventSource(
+                self, event_source_cfg))
+        elif svc == 's3':
+            self.event_sources.append(kappa.event_source.S3EventSource(
+                self, event_source_cfg))
+        else:
+            msg = 'Unsupported event source: %s' % event_source_cfg['arn']
+            raise ValueError(msg)
+
+    def add_event_sources(self):
+        for event_source in self.event_sources:
+            event_source.add(self.function)
+
+    def deploy(self):
+        if self._stack.exists():
+            self._stack.update()
+        else:
+            self._stack.create()
+        self.function.upload()
+
+    def test(self):
+        self.function.test()
+
+    def tail(self):
+        return self.function.tail()
+
+    def delete(self):
+        self._stack.delete()
+        self.function.delete()

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import logging
+
+import kappa.aws
+
+LOG = logging.getLogger(__name__)
+
+
+class EventSource(object):
+
+    def __init__(self, context, config):
+        self._context = context
+        self._config = config
+
+    @property
+    def arn(self):
+        return self._config['arn']
+
+    @property
+    def batch_size(self):
+        return self._config.get('batch_size', 100)
+
+
+class KinesisEventSource(EventSource):
+
+    def __init__(self, context, config):
+        super(KinesisEventSource, self).__init__(context, config)
+        aws = kappa.aws.get_aws(context)
+        self._lambda = aws.create_client('lambda')
+
+    def _get_uuid(self, function):
+        uuid = None
+        response = self._lambda.list_event_sources(
+            FunctionName=function.name,
+            EventSourceArn=self.arn)
+        LOG.debug(response)
+        if len(response['EventSources']) > 0:
+            uuid = response['EventSources'][0]['UUID']
+        return uuid
+
+    def add(self, function):
+        try:
+            response = self._lambda.add_event_source(
+                FunctionName=function.name,
+                Role=self.invoke_role_arn,
+                EventSource=self.arn,
+                BatchSize=self.batch_size)
+            LOG.debug(response)
+        except Exception:
+            LOG.exception('Unable to add Kinesis event source')
+
+    def remove(self, function):
+        response = None
+        uuid = self._get_uuid(function)
+        if uuid:
+            response = self._lambda.remove_event_source(
+                UUID=uuid)
+            LOG.debug(response)
+        return response
+
+
+class S3EventSource(EventSource):
+
+    def __init__(self, context, config):
+        super(S3EventSource, self).__init__(context, config)
+        aws = kappa.aws.get_aws(context)
+        self._s3 = aws.create_client('s3')
+
+    def _make_notification_id(self, function_name):
+        return 'Kappa-%s-notification' % function_name
+
+    def _get_bucket_name(self):
+        return self.arn.split(':')[-1]
+
+    def add(self, function):
+        notification_spec = {
+            'CloudFunctionConfiguration': {
+                'Id': self._make_notification_id(function.name),
+                'Events': [e for e in self.config['events']],
+                'CloudFunction': function.arn(),
+                'InvocationRole': self.invoke_role_arn}}
+        try:
+            response = self._s3.put_bucket_notification(
+                Bucket=self._get_bucket_name(),
+                NotificationConfiguration=notification_spec)
+            LOG.debug(response)
+        except Exception:
+            LOG.exception('Unable to add S3 event source')
+
+    def remove(self, function):
+        response = self._s3.get_bucket_notification(
+            Bucket=self._get_bucket_name())
+        LOG.debug(response)

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import logging
+import os
+import zipfile
+
+import kappa.aws
+import kappa.log
+
+LOG = logging.getLogger(__name__)
+
+
+class Function(object):
+
+    def __init__(self, context, config):
+        self._context = context
+        self._config = config
+        aws = kappa.aws.get_aws(context)
+        self._lambda_svc = aws.create_client('lambda')
+        self._arn = None
+        self._log = None
+
+    @property
+    def name(self):
+        return self._config['name']
+
+    @property
+    def runtime(self):
+        return self._config['runtime']
+
+    @property
+    def handler(self):
+        return self._config['handler']
+
+    @property
+    def mode(self):
+        return self._config['mode']
+
+    @property
+    def description(self):
+        return self._config['description']
+
+    @property
+    def timeout(self):
+        return self._config['timeout']
+
+    @property
+    def memory_size(self):
+        return self._config['memory_size']
+
+    @property
+    def zipfile_name(self):
+        return self._config['zipfile_name']
+
+    @property
+    def path(self):
+        return self._config['path']
+
+    @property
+    def test_data(self):
+        return self._config['test_data']
+
+    @property
+    def arn(self):
+        if self._arn is None:
+            try:
+                response = self._lambda_svc.get_function_configuration(
+                    FunctionName=self.name)
+                LOG.debug(response)
+                self._arn = response['FunctionARN']
+            except Exception:
+                LOG.debug('Unable to find ARN for function: %s' % self.name)
+        return self._arn
+
+    @property
+    def log(self):
+        if self._log is None:
+            log_group_name = '/aws/lambda/%s' % self.name
+            self._log = kappa.log.Log(self._context, log_group_name)
+        return self._log
+
+    def tail(self):
+        LOG.debug('tailing function: %s', self.name)
+        return self.log.tail()
+
+    def _zip_lambda_dir(self, zipfile_name, lambda_dir):
+        LOG.debug('_zip_lambda_dir: lambda_dir=%s', lambda_dir)
+        LOG.debug('zipfile_name=%s', zipfile_name)
+        relroot = os.path.abspath(lambda_dir)
+        with zipfile.ZipFile(zipfile_name, 'w') as zf:
+            for root, dirs, files in os.walk(lambda_dir):
+                zf.write(root, os.path.relpath(root, relroot))
+                for file in files:
+                    filename = os.path.join(root, file)
+                    if os.path.isfile(filename):
+                        arcname = os.path.join(
+                            os.path.relpath(root, relroot), file)
+                        zf.write(filename, arcname)
+
+    def _zip_lambda_file(self, zipfile_name, lambda_file):
+        LOG.debug('_zip_lambda_file: lambda_file=%s', lambda_file)
+        LOG.debug('zipfile_name=%s', zipfile_name)
+        with zipfile.ZipFile(zipfile_name, 'w') as zf:
+            zf.write(lambda_file)
+
+    def zip_lambda_function(self, zipfile_name, lambda_fn):
+        if os.path.isdir(lambda_fn):
+            self._zip_lambda_dir(zipfile_name, lambda_fn)
+        else:
+            self._zip_lambda_file(zipfile_name, lambda_fn)
+
+    def upload(self):
+        LOG.debug('uploading %s', self.zipfile_name)
+        self.zip_lambda_function(self.zipfile_name, self.path)
+        with open(self.zipfile_name, 'rb') as fp:
+            exec_role = self._context.exec_role_arn
+            try:
+                response = self._lambda_svc.upload_function(
+                    FunctionName=self.name,
+                    FunctionZip=fp,
+                    Runtime=self.runtime,
+                    Role=exec_role,
+                    Handler=self.handler,
+                    Mode=self.mode,
+                    Description=self.description,
+                    Timeout=self.timeout,
+                    MemorySize=self.memory_size)
+                LOG.debug(response)
+            except Exception:
+                LOG.exception('Unable to upload zip file')
+
+    def delete(self):
+        LOG.debug('deleting function %s', self.name)
+        response = self._lambda_svc.delete_function(FunctionName=self.name)
+        LOG.debug(response)
+        return response
+
+    def invoke_asynch(self, data_file):
+        LOG.debug('_invoke_async %s', data_file)
+        with open(data_file) as fp:
+            response = self._lambda_svc.invoke_async(
+                FunctionName=self.name,
+                InvokeArgs=fp)
+            LOG.debug(response)
+
+    def test(self):
+        self.invoke_asynch(self.test_data)

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import logging
+
+LOG = logging.getLogger(__name__)
+
+import kappa.aws
+
+
+class Log(object):
+
+    def __init__(self, context, log_group_name):
+        self._context = context
+        self.log_group_name = log_group_name
+        aws = kappa.aws.get_aws(self._context)
+        self._log_svc = aws.create_client('logs')
+
+    def streams(self):
+        LOG.debug('getting streams for log group: %s', self.log_group_name)
+        response = self._log_svc.describe_log_streams(
+            logGroupName=self.log_group_name)
+        LOG.debug(response)
+        return response['logStreams']
+
+    def tail(self):
+        LOG.debug('tailing log group: %s', self.log_group_name)
+        latest_stream = None
+        streams = self.streams()
+        for stream in streams:
+            if not latest_stream:
+                latest_stream = stream
+            elif stream['lastEventTimestamp'] > latest_stream['lastEventTimestamp']:
+                latest_stream = stream
+        response = self._log_svc.get_log_events(
+            logGroupName=self.log_group_name,
+            logStreamName=latest_stream['logStreamName'])
+        LOG.debug(response)
+        return response['events']

+# Copyright (c) 2014 Mitch Garnaat http://garnaat.org/
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+
+import logging
+import time
+
+import kappa.aws
+
+LOG = logging.getLogger(__name__)
+
+
+class Stack(object):
+
+    completed_states = ('CREATE_COMPLETE', 'UPDATE_COMPLETE')
+
+    def __init__(self, context, config):
+        self._context = context
+        self._config = config
+        aws = kappa.aws.get_aws(self._context)
+        self._cfn = aws.create_client('cloudformation')
+        self._iam = aws.create_client('iam')
+
+    @property
+    def name(self):
+        return self._config['stack_name']
+
+    @property
+    def template_path(self):
+        return self._config['template']
+
+    @property
+    def exec_role(self):
+        return self._config['exec_role']
+
+    @property
+    def exec_role_arn(self):
+        return self._get_role_arn(self.exec_role)
+
+    @property
+    def invoke_role(self):
+        return self._config['invoke_role']
+
+    @property
+    def invoke_role_arn(self):
+        return self._get_role_arn(self.invoke_role)
+
+    def _get_role_arn(self, role_name):
+        role_arn = None
+        try:
+            resources = self._cfn.list_stack_resources(
+                StackName=self.name)
+        except Exception:
+            LOG.exception('Unable to find role ARN: %s', role_name)
+        for resource in resources['StackResourceSummaries']:
+            if resource['LogicalResourceId'] == role_name:
+                role = self._iam.get_role(
+                    RoleName=resource['PhysicalResourceId'])
+                LOG.debug(role)
+                role_arn = role['Role']['Arn']
+        LOG.debug('role_arn: %s', role_arn)
+        return role_arn
+
+    def exists(self):
+        """
+        Does Cloudformation Stack already exist?
+        """
+        try:
+            response = self._cfn.describe_stacks(StackName=self.name)
+            LOG.debug('Stack %s exists', self.name)
+        except Exception:
+            LOG.debug('Stack %s does not exist', self.name)
+            response = None
+        return response
+
+    def wait(self):
+        done = False
+        while not done:
+            time.sleep(1)
+            response = self._cfn.describe_stacks(StackName=self.name)
+            status = response['Stacks'][0]['StackStatus']
+            LOG.debug('Stack status is: %s', status)
+            if status in self.completed_states:
+                done = True
+
+    def create(self):
+        LOG.debug('create_stack: stack_name=%s', self.name)
+        template_body = open(self.template_path).read()
+        try:
+            self._cfn.create_stack(
+                StackName=self.name, TemplateBody=template_body,
+                Capabilities=['CAPABILITY_IAM'])
+        except Exception:
+            LOG.exception('Unable to create stack')
+        self.wait()
+
+    def update(self):
+        LOG.debug('create_stack: stack_name=%s', self.name)
+        template_body = open(self.template_path).read()
+        try:
+            self._cfn.update_stack(
+                StackName=self.name, TemplateBody=template_body,
+                Capabilities=['CAPABILITY_IAM'])
+        except Exception, e:
+            if 'ValidationError' in str(e):
+                LOG.info('No Updates Required')
+            else:
+                LOG.exception('Unable to update stack')
+        self.wait()
+
+    def delete(self):
+        LOG.debug('delete_stack: stack_name=%s', self.name)
+        try:
+            self._cfn.delete_stack(StackName=self.name)
+        except Exception:
+            LOG.exception('Unable to delete stack: %s', self.name)

-botocore==0.75.0
+botocore==0.80.0
 click==3.3
 click==3.3
 PyYAML>=3.11
 PyYAML>=3.11
 nose==1.3.1
 nose==1.3.1

@@ -16,6 +16,8 @@ lambda:
   memory_size: 128
   memory_size: 128
   timeout: 3
   timeout: 3
   mode: event
   mode: event
-  event_source: arn:aws:kinesis:us-east-1:084307701560:stream/lambdastream
+  event_sources:
+    -
+      arn: arn:aws:kinesis:us-east-1:084307701560:stream/lambdastream
   test_data: input.json
   test_data: input.json
   
   
\ No newline at end of file

@@ -17,6 +17,8 @@ lambda:
   timeout: 3
   timeout: 3
   mode: event
   mode: event
   test_data: input.json
   test_data: input.json
-  event_source: arn:aws:s3:::sourcebucket
-  s3_events:
-    - s3:ObjectCreated:*
\ No newline at end of file
+  event_sources:
+    -
+      arn: arn:aws:s3:::test-1245812163
+      events:
+        - s3:ObjectCreated:*

@@ -5,7 +5,7 @@ from setuptools import setup, find_packages
 import os
 import os
 
 
 requires = [
 requires = [
-    'botocore==0.75.0',
+    'botocore==0.80.0',
     'click==3.3',
     'click==3.3',
     'PyYAML>=3.11'
     'PyYAML>=3.11'
 ]
 ]