streamcorpus_pipeline/_pipeline.py
def _run_incremental_transforms(self, si, transforms):
        '''
        Run transforms on stream item.
        Item may be discarded by some transform.
        Writes successful items out to current self.t_chunk
        Returns transformed item or None.
        '''
        ## operate each transform on this one StreamItem
        for transform in transforms:
            try:
                stream_id = si.stream_id
                si_new = transform(si, context=self.context)

                if si_new is None:
                    logger.warn('transform %r deleted %s abs_url=%r',
                                 transform, stream_id, si and si.abs_url)
                    return None
                si = si_new

            except TransformGivingUp:
                ## do nothing
                logger.info('transform %r giving up on %r',
                            transform, si.stream_id)

            except Exception, exc:
                logger.critical(
                    'transform %r failed on %r from i_str=%r abs_url=%r',
                    transform, si and si.stream_id, self.context.get('i_str'),
                    si and si.abs_url, exc_info=True)

        assert si is not None

        ## expect to always have a stream_time
        if not si.stream_time:
            raise InvalidStreamItem('empty stream_time: %s' % si)

        if si.stream_id is None:
            raise InvalidStreamItem('empty stream_id: %r' % si)

        ## put the StreamItem into the output
        if type(si) != streamcorpus.StreamItem_v0_3_0:
            raise InvalidStreamItem('incorrect stream item object %r' %
                                    type(si))
        self.t_chunk.add(si)
        return si
def _run_incremental_transforms(self, si, transforms):
        '''
        Run transforms on stream item.
        Item may be discarded by some transform.
        Writes successful items out to current self.t_chunk
        Returns transformed item or None.
        '''
        ## operate each transform on this one StreamItem
        for transform in transforms:
            try:
                stream_id = si.stream_id
                si_new = transform(si, context=self.context)

                if si_new is None:
                    logger.warn('transform %r deleted %s abs_url=%r',
                                 transform, stream_id, si and si.abs_url)
                    return None
                si = si_new

            except TransformGivingUp:
                ## do nothing
                logger.info('transform %r giving up on %r',
                            transform, si.stream_id)

            except Exception, exc:
                logger.critical(
                    'transform %r failed on %r from i_str=%r abs_url=%r',
                    transform, si and si.stream_id, self.context.get('i_str'),
                    si and si.abs_url, exc_info=True)

        assert si is not None

        ## expect to always have a stream_time
        if not si.stream_time:
            raise InvalidStreamItem('empty stream_time: %s' % si)

        if si.stream_id is None:
            raise InvalidStreamItem('empty stream_id: %r' % si)

        ## put the StreamItem into the output
        if type(si) != streamcorpus.StreamItem_v0_3_0:
            raise InvalidStreamItem('incorrect stream item object %r' %
                                    type(si))
        self.t_chunk.add(si)
        return si
def _run_incremental_transforms ( self , si , transforms ) : ## operate each transform on this one StreamItem for transform in transforms : try : stream_id = si . stream_id si_new = transform ( si , context = self . context ) if si_new is None : logger . warn ( 'transform %r deleted %s abs_url=%r' , transform , stream_id , si and si . abs_url ) return None si = si_new except TransformGivingUp : ## do nothing logger . info ( 'transform %r giving up on %r' , transform , si . stream_id ) except Exception , exc : logger . critical ( 'transform %r failed on %r from i_str=%r abs_url=%r' , transform , si and si . stream_id , self . context . get ( 'i_str' ) , si and si . abs_url , exc_info = True ) assert si is not None ## expect to always have a stream_time if not si . stream_time : raise InvalidStreamItem ( 'empty stream_time: %s' % si ) if si . stream_id is None : raise InvalidStreamItem ( 'empty stream_id: %r' % si ) ## put the StreamItem into the output if type ( si ) != streamcorpus . StreamItem_v0_3_0 : raise InvalidStreamItem ( 'incorrect stream item object %r' % type ( si ) ) self . t_chunk . add ( si ) return si
Run transforms on stream item.
        Item may be discarded by some transform.
        Writes successful items out to current self.t_chunk
        Returns transformed item or None.
Run transforms on stream item . Item may be discarded by some transform . Writes successful items out to current self . t_chunk Returns transformed item or None .