streamcorpus_pipeline/_s3_storage.py
def _decode(self, data):
        '''
        Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.
        '''
        informat = self.config['input_format'].lower()
        if informat == 'spinn3r':
            return _generate_stream_items(data)
        elif informat == 'streamitem':
            ver = self.config['streamcorpus_version']
            if ver not in _message_versions:
                raise ConfigurationError(
                    'Not a valid streamcorpus version: %s '
                    '(choose from: %s)'
                    % (ver, ', '.join(_message_versions.keys())))

            message = _message_versions[ver]
            return streamcorpus.Chunk(data=data, message=message)
        elif informat == 'featurecollection' and FCChunk is not None:
            return FCChunk(data=data)
        else:
            raise ConfigurationError(
                'from_s3_chunks unknown input_format = %r'
                % informat)
def _decode(self, data):
        '''
        Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.
        '''
        informat = self.config['input_format'].lower()
        if informat == 'spinn3r':
            return _generate_stream_items(data)
        elif informat == 'streamitem':
            ver = self.config['streamcorpus_version']
            if ver not in _message_versions:
                raise ConfigurationError(
                    'Not a valid streamcorpus version: %s '
                    '(choose from: %s)'
                    % (ver, ', '.join(_message_versions.keys())))

            message = _message_versions[ver]
            return streamcorpus.Chunk(data=data, message=message)
        elif informat == 'featurecollection' and FCChunk is not None:
            return FCChunk(data=data)
        else:
            raise ConfigurationError(
                'from_s3_chunks unknown input_format = %r'
                % informat)
def _decode ( self , data ) : informat = self . config [ 'input_format' ] . lower ( ) if informat == 'spinn3r' : return _generate_stream_items ( data ) elif informat == 'streamitem' : ver = self . config [ 'streamcorpus_version' ] if ver not in _message_versions : raise ConfigurationError ( 'Not a valid streamcorpus version: %s ' '(choose from: %s)' % ( ver , ', ' . join ( _message_versions . keys ( ) ) ) ) message = _message_versions [ ver ] return streamcorpus . Chunk ( data = data , message = message ) elif informat == 'featurecollection' and FCChunk is not None : return FCChunk ( data = data ) else : raise ConfigurationError ( 'from_s3_chunks unknown input_format = %r' % informat )
Given the raw data from s3, return a generator for the items
        contained in that data. A generator is necessary to support
        chunk files, but non-chunk files can be provided by a generator
        that yields exactly one item.

        Decoding works by case analysis on the config option
        ``input_format``. If an invalid ``input_format`` is given, then
        a ``ConfigurationError`` is raised.
Given the raw data from s3 return a generator for the items contained in that data . A generator is necessary to support chunk files but non - chunk files can be provided by a generator that yields exactly one item .