def get_event_source(event_source, lambda_arn, target_function, boto_session, dry=False):
    """

    Given an event_source dictionary item, a session and a lambda_arn,
    hack into Kappa's Gibson, create out an object we can call
    to schedule this event, and return the event source.

    """
    import kappa.function
    import kappa.restapi
    import kappa.event_source.base
    import kappa.event_source.dynamodb_stream
    import kappa.event_source.kinesis
    import kappa.event_source.s3
    import kappa.event_source.sns
    import kappa.event_source.cloudwatch
    import kappa.policy
    import kappa.role
    import kappa.awsclient

    class PseudoContext(object):
        def __init__(self):
            return

    class PseudoFunction(object):
        def __init__(self):
            return

    # Mostly adapted from kappa - will probably be replaced by kappa support
    class SqsEventSource(kappa.event_source.base.EventSource):

        def __init__(self, context, config):
            super(SqsEventSource, self).__init__(context, config)
            self._lambda = kappa.awsclient.create_client(
                'lambda', context.session)

        def _get_uuid(self, function):
            uuid = None
            response = self._lambda.call(
                'list_event_source_mappings',
                FunctionName=function.name,
                EventSourceArn=self.arn)
            LOG.debug(response)
            if len(response['EventSourceMappings']) > 0:
                uuid = response['EventSourceMappings'][0]['UUID']
            return uuid

        def add(self, function):
            try:
                response = self._lambda.call(
                    'create_event_source_mapping',
                    FunctionName=function.name,
                    EventSourceArn=self.arn,
                    BatchSize=self.batch_size,
                    Enabled=self.enabled
                    )
                LOG.debug(response)
            except Exception:
                LOG.exception('Unable to add event source')

        def enable(self, function):
            self._config['enabled'] = True
            try:
                response = self._lambda.call(
                    'update_event_source_mapping',
                    UUID=self._get_uuid(function),
                    Enabled=self.enabled
                    )
                LOG.debug(response)
            except Exception:
                LOG.exception('Unable to enable event source')

        def disable(self, function):
            self._config['enabled'] = False
            try:
                response = self._lambda.call(
                    'update_event_source_mapping',
                    FunctionName=function.name,
                    Enabled=self.enabled
                    )
                LOG.debug(response)
            except Exception:
                LOG.exception('Unable to disable event source')

        def update(self, function):
            response = None
            uuid = self._get_uuid(function)
            if uuid:
                try:
                    response = self._lambda.call(
                        'update_event_source_mapping',
                        BatchSize=self.batch_size,
                        Enabled=self.enabled,
                        FunctionName=function.arn)
                    LOG.debug(response)
                except Exception:
                    LOG.exception('Unable to update event source')

        def remove(self, function):
            response = None
            uuid = self._get_uuid(function)
            if uuid:
                response = self._lambda.call(
                    'delete_event_source_mapping',
                    UUID=uuid)
                LOG.debug(response)
            return response

        def status(self, function):
            response = None
            LOG.debug('getting status for event source %s', self.arn)
            uuid = self._get_uuid(function)
            if uuid:
                try:
                    response = self._lambda.call(
                        'get_event_source_mapping',
                        UUID=self._get_uuid(function))
                    LOG.debug(response)
                except botocore.exceptions.ClientError:
                    LOG.debug('event source %s does not exist', self.arn)
                    response = None
            else:
                LOG.debug('No UUID for event source %s', self.arn)
            return response

    class ExtendedSnsEventSource(kappa.event_source.sns.SNSEventSource):
        @property
        def filters(self):
            return self._config.get('filters')

        def add_filters(self, function):
            try:
                subscription = self.exists(function)
                if subscription:
                    response = self._sns.call(
                        'set_subscription_attributes',
                        SubscriptionArn=subscription['SubscriptionArn'],
                        AttributeName='FilterPolicy',
                        AttributeValue=json.dumps(self.filters)
                    )
                    kappa.event_source.sns.LOG.debug(response)
            except Exception:
                kappa.event_source.sns.LOG.exception('Unable to add filters for SNS topic %s', self.arn)

        def add(self, function):
            super(ExtendedSnsEventSource, self).add(function)
            if self.filters:
                self.add_filters(function)

    event_source_map = {
        'dynamodb': kappa.event_source.dynamodb_stream.DynamoDBStreamEventSource,
        'kinesis': kappa.event_source.kinesis.KinesisEventSource,
        's3': kappa.event_source.s3.S3EventSource,
        'sns': ExtendedSnsEventSource,
        'sqs': SqsEventSource,
        'events': kappa.event_source.cloudwatch.CloudWatchEventSource
    }

    arn = event_source['arn']
    _, _, svc, _ = arn.split(':', 3)

    event_source_func = event_source_map.get(svc, None)
    if not event_source_func:
        raise ValueError('Unknown event source: {0}'.format(arn))

    def autoreturn(self, function_name):
        return function_name

    event_source_func._make_notification_id = autoreturn

    ctx = PseudoContext()
    ctx.session = boto_session

    funk = PseudoFunction()
    funk.name = lambda_arn

    # Kappa 0.6.0 requires this nasty hacking,
    # hopefully we can remove at least some of this soon.
    # Kappa 0.7.0 introduces a whole host over other changes we don't
    # really want, so we're stuck here for a little while.

    # Related:  https://github.com/Miserlou/Zappa/issues/684
    #           https://github.com/Miserlou/Zappa/issues/688
    #           https://github.com/Miserlou/Zappa/commit/3216f7e5149e76921ecdf9451167846b95616313
    if svc == 's3':
        split_arn = lambda_arn.split(':')
        arn_front = ':'.join(split_arn[:-1])
        arn_back = split_arn[-1]
        ctx.environment = arn_back
        funk.arn = arn_front
        funk.name = ':'.join([arn_back, target_function])
    else:
        funk.arn = lambda_arn

    funk._context = ctx

    event_source_obj = event_source_func(ctx, event_source)

    return event_source_obj, ctx, funk
def get_event_source ( event_source , lambda_arn , target_function , boto_session , dry = False ) : import kappa . function import kappa . restapi import kappa . event_source . base import kappa . event_source . dynamodb_stream import kappa . event_source . kinesis import kappa . event_source . s3 import kappa . event_source . sns import kappa . event_source . cloudwatch import kappa . policy import kappa . role import kappa . awsclient class PseudoContext ( object ) : def __init__ ( self ) : return class PseudoFunction ( object ) : def __init__ ( self ) : return class SqsEventSource ( kappa . event_source . base . EventSource ) : def __init__ ( self , context , config ) : super ( SqsEventSource , self ) . __init__ ( context , config ) self . _lambda = kappa . awsclient . create_client ( 'lambda' , context . session ) def _get_uuid ( self , function ) : uuid = None response = self . _lambda . call ( 'list_event_source_mappings' , FunctionName = function . name , EventSourceArn = self . arn ) LOG . debug ( response ) if len ( response [ 'EventSourceMappings' ] ) > 0 : uuid = response [ 'EventSourceMappings' ] [ 0 ] [ 'UUID' ] return uuid def add ( self , function ) : try : response = self . _lambda . call ( 'create_event_source_mapping' , FunctionName = function . name , EventSourceArn = self . arn , BatchSize = self . batch_size , Enabled = self . enabled ) LOG . debug ( response ) except Exception : LOG . exception ( 'Unable to add event source' ) def enable ( self , function ) : self . _config [ 'enabled' ] = True try : response = self . _lambda . call ( 'update_event_source_mapping' , UUID = self . _get_uuid ( function ) , Enabled = self . enabled ) LOG . debug ( response ) except Exception : LOG . exception ( 'Unable to enable event source' ) def disable ( self , function ) : self . _config [ 'enabled' ] = False try : response = self . _lambda . call ( 'update_event_source_mapping' , FunctionName = function . name , Enabled = self . enabled ) LOG . debug ( response ) except Exception : LOG . exception ( 'Unable to disable event source' ) def update ( self , function ) : response = None uuid = self . _get_uuid ( function ) if uuid : try : response = self . _lambda . call ( 'update_event_source_mapping' , BatchSize = self . batch_size , Enabled = self . enabled , FunctionName = function . arn ) LOG . debug ( response ) except Exception : LOG . exception ( 'Unable to update event source' ) def remove ( self , function ) : response = None uuid = self . _get_uuid ( function ) if uuid : response = self . _lambda . call ( 'delete_event_source_mapping' , UUID = uuid ) LOG . debug ( response ) return response def status ( self , function ) : response = None LOG . debug ( 'getting status for event source %s' , self . arn ) uuid = self . _get_uuid ( function ) if uuid : try : response = self . _lambda . call ( 'get_event_source_mapping' , UUID = self . _get_uuid ( function ) ) LOG . debug ( response ) except botocore . exceptions . ClientError : LOG . debug ( 'event source %s does not exist' , self . arn ) response = None else : LOG . debug ( 'No UUID for event source %s' , self . arn ) return response class ExtendedSnsEventSource ( kappa . event_source . sns . SNSEventSource ) : @ property def filters ( self ) : return self . _config . get ( 'filters' ) def add_filters ( self , function ) : try : subscription = self . exists ( function ) if subscription : response = self . _sns . call ( 'set_subscription_attributes' , SubscriptionArn = subscription [ 'SubscriptionArn' ] , AttributeName = 'FilterPolicy' , AttributeValue = json . dumps ( self . filters ) ) kappa . event_source . sns . LOG . debug ( response ) except Exception : kappa . event_source . sns . LOG . exception ( 'Unable to add filters for SNS topic %s' , self . arn ) def add ( self , function ) : super ( ExtendedSnsEventSource , self ) . add ( function ) if self . filters : self . add_filters ( function ) event_source_map = { 'dynamodb' : kappa . event_source . dynamodb_stream . DynamoDBStreamEventSource , 'kinesis' : kappa . event_source . kinesis . KinesisEventSource , 's3' : kappa . event_source . s3 . S3EventSource , 'sns' : ExtendedSnsEventSource , 'sqs' : SqsEventSource , 'events' : kappa . event_source . cloudwatch . CloudWatchEventSource } arn = event_source [ 'arn' ] _ , _ , svc , _ = arn . split ( ':' , 3 ) event_source_func = event_source_map . get ( svc , None ) if not event_source_func : raise ValueError ( 'Unknown event source: {0}' . format ( arn ) ) def autoreturn ( self , function_name ) : return function_name event_source_func . _make_notification_id = autoreturn ctx = PseudoContext ( ) ctx . session = boto_session funk = PseudoFunction ( ) funk . name = lambda_arn if svc == 's3' : split_arn = lambda_arn . split ( ':' ) arn_front = ':' . join ( split_arn [ : - 1 ] ) arn_back = split_arn [ - 1 ] ctx . environment = arn_back funk . arn = arn_front funk . name = ':' . join ( [ arn_back , target_function ] ) else : funk . arn = lambda_arn funk . _context = ctx event_source_obj = event_source_func ( ctx , event_source ) return event_source_obj , ctx , funk
Given an event_source dictionary item, a session and a lambda_arn,
    hack into Kappa's Gibson, create out an object we can call
    to schedule this event, and return the event source.