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
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.