fulltext/agent/consumer.py
def process_record(self, record: dict) -> None:
        """
        Call for each record that is passed to process_records.

        Parameters
        ----------
        data : bytes
        partition_key : bytes
        sequence_number : int
        sub_sequence_number : int

        Raises
        ------
        IndexingFailed
            Indexing of the document failed in a way that indicates recovery
            is unlikely for subsequent papers, or too many individual
            documents failed.

        """
        time.sleep(self.sleep)
        logger.debug(f'Processing record %s', record["SequenceNumber"])
        try:
            deserialized = json.loads(record['Data'].decode('utf-8'))
        except json.decoder.JSONDecodeError as e:
            logger.error("Error while deserializing data %s", e)
            logger.error("Data payload: %s", record['Data'])
            raise BadMessage('Could not deserialize payload')

        arxiv_id: str = deserialized.get('document_id')
        logger.info(f'Processing notification for %s', arxiv_id)
        extract.create_task(arxiv_id, 'arxiv')

        Call for each record that is passed to process_records.

        Parameters
        ----------
        data : bytes
        partition_key : bytes
        sequence_number : int
        sub_sequence_number : int

        Raises
        ------
        IndexingFailed
            Indexing of the document failed in a way that indicates recovery
            is unlikely for subsequent papers, or too many individual
            documents failed.

        
Call for each record that is passed to process_records.
Parameters

data : bytes
partition_key : bytes
sequence_number : int
sub_sequence_number : int

Raises

IndexingFailed
Indexing of the document failed in a way that indicates recovery
is unlikely for subsequent papers, or too many individual
documents failed.
Call for each record that is passed to process_records . Parameters data : bytes partition_key : bytes sequence_number : int sub_sequence_number : int Raises IndexingFailed Indexing of the document failed in a way that indicates recovery is unlikely for subsequent papers or too many individual documents failed .
def process_record(self, record: dict) -> None:
        time.sleep(self.sleep)
        logger.debug(f'Processing record %s', record["SequenceNumber"])
        try:
            deserialized = json.loads(record['Data'].decode('utf-8'))
        except json.decoder.JSONDecodeError as e:
            logger.error("Error while deserializing data %s", e)
            logger.error("Data payload: %s", record['Data'])
            raise BadMessage('Could not deserialize payload')
        arxiv_id: str = deserialized.get('document_id')
        logger.info(f'Processing notification for %s', arxiv_id)
        extract.create_task(arxiv_id, 'arxiv')
def process_record ( self , record : dict ) -> None : time . sleep ( self . sleep ) logger . debug ( f'Processing record %s' , record [ "SequenceNumber" ] ) try : deserialized = json . loads ( record [ 'Data' ] . decode ( 'utf-8' ) ) except json . decoder . JSONDecodeError as e : logger . error ( "Error while deserializing data %s" , e ) logger . error ( "Data payload: %s" , record [ 'Data' ] ) raise BadMessage ( 'Could not deserialize payload' ) arxiv_id : str = deserialized . get ( 'document_id' ) logger . info ( f'Processing notification for %s' , arxiv_id ) extract . create_task ( arxiv_id , 'arxiv' )
Call for each record that is passed to process_records.
Call for each record that is passed to process_records .
"""
        Call for each record that is passed to process_records.

        Parameters
        ----------
        data : bytes
        partition_key : bytes
        sequence_number : int
        sub_sequence_number : int

        Raises
        ------
        IndexingFailed
            Indexing of the document failed in a way that indicates recovery
            is unlikely for subsequent papers, or too many individual
            documents failed.

        """
self
record
dict