sample_metadata/parser/generic_parser.py
def parse_manifest(  # pylint: disable=too-many-branches
        self, file_pointer, delimiter=',', confirm=False, dry_run=False
    ) -> Union[Dict[str, str], Tuple[List, Dict, Dict, Dict, Dict]]:
        """
        Parse manifest from iterable (file pointer / String.IO)

        Returns a dict mapping external sample ID to CPG sample ID
        """
        # a sample has many rows
        sample_map = defaultdict(list)

        reader = csv.DictReader(file_pointer, delimiter=delimiter)
        for row in reader:
            sample_id = self.get_sample_id(row)
            sample_map[sample_id].append(row)

        if len(sample_map) == 0:
            raise ValueError(f'The manifest file contains no records')

        # now we can start adding!!
        sapi = SampleApi()
        seqapi = SequenceApi()
        analysisapi = AnalysisApi()

        # determine if any samples exist
        existing_external_id_to_cpgid = sapi.get_sample_id_map_by_external(
            self.sample_metadata_project, list(sample_map.keys()), allow_missing=True
        )
        existing_cpgid_to_seq_id = {}

        if len(existing_external_id_to_cpgid) > 0:
            existing_cpgid_to_seq_id = seqapi.get_sequence_ids_from_sample_ids(
                request_body=list(existing_external_id_to_cpgid.values()),
            )

        samples_to_add: List[NewSample] = []
        # by external_sample_id
        sequences_to_add: Dict[str, NewSequence] = defaultdict(list)
        analyses_to_add: Dict[str, List[AnalysisModel]] = defaultdict(list)

        samples_to_update: Dict[str, SampleUpdateModel] = {}
        sequences_to_update: Dict[int, SequenceUpdateModel] = {}

        for external_sample_id in sample_map:
            if self.verbose:
                logger.info(f'Preparing {external_sample_id}')

            rows: Union[Dict[str, str], List[Dict[str, str]]] = sample_map[external_sample_id]

            if isinstance(rows, list) and len(rows) == 1:
                rows = rows[0]
            # now we have sample / sequencing meta across 4 different rows, so collapse them
            collapsed_sequencing_meta = self.get_sequence_meta(external_sample_id, rows)
            collapsed_sample_meta = self.get_sample_meta(external_sample_id, rows)
            collapsed_qc = self.get_qc_meta(external_sample_id, rows)
            collapsed_analyses = self.get_analyses(
                external_sample_id,
                rows,
                cpg_id=existing_external_id_to_cpgid.get(external_sample_id)
            )

            sample_type = self.get_sample_type(external_sample_id, rows)
            sequence_status = self.get_sequence_status(external_sample_id, rows)

            if external_sample_id not in existing_external_id_to_cpgid:
                samples_to_add.append(
                    NewSample(
                        external_id=external_sample_id,
                        type=SampleType(sample_type),
                        meta=collapsed_sample_meta,
                    )
                )

            else:
                # it already exists
                cpgid = existing_external_id_to_cpgid[external_sample_id]
                samples_to_update[cpgid] = SampleUpdateModel(
                    meta=collapsed_sample_meta,
                )
                # ignore QC results if sample already exists

            if collapsed_analyses:
                analyses_to_add[external_sample_id].extend(collapsed_analyses)

            if collapsed_qc:
                analyses_to_add[external_sample_id].append(
                    AnalysisModel(
                        sample_ids=['<none>'],
                        type=AnalysisType('qc'),
                        status=AnalysisStatus('completed'),
                        meta=collapsed_qc,
                    )
                )

            # should we add or update sequencing
            if collapsed_sequencing_meta:
                if (
                    external_sample_id in existing_external_id_to_cpgid
                    and existing_external_id_to_cpgid[external_sample_id]
                    in existing_cpgid_to_seq_id
                ):
                    # update, we have a cpg sample ID AND a sequencing ID
                    cpgid = existing_external_id_to_cpgid[external_sample_id]
                    seq_id = existing_cpgid_to_seq_id[cpgid]
                    sequences_to_update[seq_id] = SequenceUpdateModel(
                        meta=collapsed_sequencing_meta,
                        status=SequenceStatus(sequence_status)
                    )
                else:
                    # the internal sample ID gets mapped back later
                    sequences_to_add[external_sample_id] = NewSequence(
                        sample_id='<None>',  # keep the type initialisation happy
                        meta=collapsed_sequencing_meta,
                        type=SequenceType('wgs'),
                        status=SequenceStatus(sequence_status),
                    )

        message = f"""\
Processing samples: {', '.join(sample_map.keys())}

Adding {len(samples_to_add)} samples
Adding {len(sequences_to_add)} sequences
Adding {len(analyses_to_add)} analysis results

Updating {len(samples_to_update)} sample
Updating {len(sequences_to_update)} sequences"""

        if dry_run:
            logger.info('Dry run, so returning without inserting / updating metadata')
            return (
                samples_to_add,
                sequences_to_add,
                samples_to_update,
                sequences_to_update,
                analyses_to_add,
            )

        if confirm:
            resp = str(input(message + '\n\nConfirm (y): '))
            if resp.lower() != 'y':
                raise SystemExit()
        else:
            logger.info(message)

        logger.info(f'Adding {len(samples_to_add)} samples to SM-DB database')
        added_ext_sample_to_internal_id = {}
        for new_sample in samples_to_add:
            sample_id = sapi.create_new_sample(
                project=self.sample_metadata_project, new_sample=new_sample
            )
            added_ext_sample_to_internal_id[new_sample.external_id] = sample_id
            existing_external_id_to_cpgid[new_sample.external_id] = sample_id

        logger.info(f'Adding {len(sequences_to_add)} sequence entries')
        for sample_id, seq in sequences_to_add.items():
            seq.sample_id = existing_external_id_to_cpgid[sample_id]
            seqapi.create_new_sequence(new_sequence=seq)

        logger.info(f'Adding analysis entries for {len(analyses_to_add)} samples')
        for sample_id, analyses in analyses_to_add.items():
            for analysis in analyses:
                analysis.sample_ids = [existing_external_id_to_cpgid[sample_id]]
                analysisapi.create_new_analysis(
                    project=self.sample_metadata_project, analysis_model=analysis
                )

        logger.info(f'Updating {len(samples_to_update)} samples')
        for internal_sample_id, sample_update in samples_to_update.items():
            sapi.update_sample(
                id_=internal_sample_id,
                sample_update_model=sample_update,
            )

        logger.info(f'Updating {len(sequences_to_update)} sequences')
        for i, (seq_id, seq_update) in enumerate(sequences_to_update.items()):
            try:
                seqapi.update_sequence(
                    sequence_id=seq_id,
                    sequence_update_model=seq_update,
                )
                logger.info(f'#{i+1}: updating sequence')
            except ApiException:
                traceback.print_exc()
                logger.error(f'#{i+1}: error updating sequence {seq_id}: {seq_update}')

        return added_ext_sample_to_internal_id

        Parse manifest from iterable (file pointer / String.IO)

        Returns a dict mapping external sample ID to CPG sample ID
        
Parse manifest from iterable (file pointer / String.IO)
Returns a dict mapping external sample ID to CPG sample ID
Parse manifest from iterable ( file pointer / String . IO ) Returns a dict mapping external sample ID to CPG sample ID
def parse_manifest(  
        self, file_pointer, delimiter=',', confirm=False, dry_run=False
    ) -> Union[Dict[str, str], Tuple[List, Dict, Dict, Dict, Dict]]:
        sample_map = defaultdict(list)
        reader = csv.DictReader(file_pointer, delimiter=delimiter)
        for row in reader:
            sample_id = self.get_sample_id(row)
            sample_map[sample_id].append(row)
        if len(sample_map) == 0:
            raise ValueError(f'The manifest file contains no records')
        sapi = SampleApi()
        seqapi = SequenceApi()
        analysisapi = AnalysisApi()
        existing_external_id_to_cpgid = sapi.get_sample_id_map_by_external(
            self.sample_metadata_project, list(sample_map.keys()), allow_missing=True
        )
        existing_cpgid_to_seq_id = {}
        if len(existing_external_id_to_cpgid) > 0:
            existing_cpgid_to_seq_id = seqapi.get_sequence_ids_from_sample_ids(
                request_body=list(existing_external_id_to_cpgid.values()),
            )
        samples_to_add: List[NewSample] = []
        sequences_to_add: Dict[str, NewSequence] = defaultdict(list)
        analyses_to_add: Dict[str, List[AnalysisModel]] = defaultdict(list)
        samples_to_update: Dict[str, SampleUpdateModel] = {}
        sequences_to_update: Dict[int, SequenceUpdateModel] = {}
        for external_sample_id in sample_map:
            if self.verbose:
                logger.info(f'Preparing {external_sample_id}')
            rows: Union[Dict[str, str], List[Dict[str, str]]] = sample_map[external_sample_id]
            if isinstance(rows, list) and len(rows) == 1:
                rows = rows[0]
            collapsed_sequencing_meta = self.get_sequence_meta(external_sample_id, rows)
            collapsed_sample_meta = self.get_sample_meta(external_sample_id, rows)
            collapsed_qc = self.get_qc_meta(external_sample_id, rows)
            collapsed_analyses = self.get_analyses(
                external_sample_id,
                rows,
                cpg_id=existing_external_id_to_cpgid.get(external_sample_id)
            )
            sample_type = self.get_sample_type(external_sample_id, rows)
            sequence_status = self.get_sequence_status(external_sample_id, rows)
            if external_sample_id not in existing_external_id_to_cpgid:
                samples_to_add.append(
                    NewSample(
                        external_id=external_sample_id,
                        type=SampleType(sample_type),
                        meta=collapsed_sample_meta,
                    )
                )
            else:
                cpgid = existing_external_id_to_cpgid[external_sample_id]
                samples_to_update[cpgid] = SampleUpdateModel(
                    meta=collapsed_sample_meta,
                )
            if collapsed_analyses:
                analyses_to_add[external_sample_id].extend(collapsed_analyses)
            if collapsed_qc:
                analyses_to_add[external_sample_id].append(
                    AnalysisModel(
                        sample_ids=['<none>'],
                        type=AnalysisType('qc'),
                        status=AnalysisStatus('completed'),
                        meta=collapsed_qc,
                    )
                )
            if collapsed_sequencing_meta:
                if (
                    external_sample_id in existing_external_id_to_cpgid
                    and existing_external_id_to_cpgid[external_sample_id]
                    in existing_cpgid_to_seq_id
                ):
                    cpgid = existing_external_id_to_cpgid[external_sample_id]
                    seq_id = existing_cpgid_to_seq_id[cpgid]
                    sequences_to_update[seq_id] = SequenceUpdateModel(
                        meta=collapsed_sequencing_meta,
                        status=SequenceStatus(sequence_status)
                    )
                else:
                    sequences_to_add[external_sample_id] = NewSequence(
                        sample_id='<None>',  
                        meta=collapsed_sequencing_meta,
                        type=SequenceType('wgs'),
                        status=SequenceStatus(sequence_status),
                    )
        message = f"""\
Processing samples: {', '.join(sample_map.keys())}
Adding {len(samples_to_add)} samples
Adding {len(sequences_to_add)} sequences
Adding {len(analyses_to_add)} analysis results
Updating {len(samples_to_update)} sample
Updating {len(sequences_to_update)} sequences"""
        if dry_run:
            logger.info('Dry run, so returning without inserting / updating metadata')
            return (
                samples_to_add,
                sequences_to_add,
                samples_to_update,
                sequences_to_update,
                analyses_to_add,
            )
        if confirm:
            resp = str(input(message + '\n\nConfirm (y): '))
            if resp.lower() != 'y':
                raise SystemExit()
        else:
            logger.info(message)
        logger.info(f'Adding {len(samples_to_add)} samples to SM-DB database')
        added_ext_sample_to_internal_id = {}
        for new_sample in samples_to_add:
            sample_id = sapi.create_new_sample(
                project=self.sample_metadata_project, new_sample=new_sample
            )
            added_ext_sample_to_internal_id[new_sample.external_id] = sample_id
            existing_external_id_to_cpgid[new_sample.external_id] = sample_id
        logger.info(f'Adding {len(sequences_to_add)} sequence entries')
        for sample_id, seq in sequences_to_add.items():
            seq.sample_id = existing_external_id_to_cpgid[sample_id]
            seqapi.create_new_sequence(new_sequence=seq)
        logger.info(f'Adding analysis entries for {len(analyses_to_add)} samples')
        for sample_id, analyses in analyses_to_add.items():
            for analysis in analyses:
                analysis.sample_ids = [existing_external_id_to_cpgid[sample_id]]
                analysisapi.create_new_analysis(
                    project=self.sample_metadata_project, analysis_model=analysis
                )
        logger.info(f'Updating {len(samples_to_update)} samples')
        for internal_sample_id, sample_update in samples_to_update.items():
            sapi.update_sample(
                id_=internal_sample_id,
                sample_update_model=sample_update,
            )
        logger.info(f'Updating {len(sequences_to_update)} sequences')
        for i, (seq_id, seq_update) in enumerate(sequences_to_update.items()):
            try:
                seqapi.update_sequence(
                    sequence_id=seq_id,
                    sequence_update_model=seq_update,
                )
                logger.info(f'#{i+1}: updating sequence')
            except ApiException:
                traceback.print_exc()
                logger.error(f'#{i+1}: error updating sequence {seq_id}: {seq_update}')
        return added_ext_sample_to_internal_id
def parse_manifest ( self , file_pointer , delimiter = ',' , confirm = False , dry_run = False ) -> Union [ Dict [ str , str ] , Tuple [ List , Dict , Dict , Dict , Dict ] ] : sample_map = defaultdict ( list ) reader = csv . DictReader ( file_pointer , delimiter = delimiter ) for row in reader : sample_id = self . get_sample_id ( row ) sample_map [ sample_id ] . append ( row ) if len ( sample_map ) == 0 : raise ValueError ( f'The manifest file contains no records' ) sapi = SampleApi ( ) seqapi = SequenceApi ( ) analysisapi = AnalysisApi ( ) existing_external_id_to_cpgid = sapi . get_sample_id_map_by_external ( self . sample_metadata_project , list ( sample_map . keys ( ) ) , allow_missing = True ) existing_cpgid_to_seq_id = { } if len ( existing_external_id_to_cpgid ) > 0 : existing_cpgid_to_seq_id = seqapi . get_sequence_ids_from_sample_ids ( request_body = list ( existing_external_id_to_cpgid . values ( ) ) , ) samples_to_add : List [ NewSample ] = [ ] sequences_to_add : Dict [ str , NewSequence ] = defaultdict ( list ) analyses_to_add : Dict [ str , List [ AnalysisModel ] ] = defaultdict ( list ) samples_to_update : Dict [ str , SampleUpdateModel ] = { } sequences_to_update : Dict [ int , SequenceUpdateModel ] = { } for external_sample_id in sample_map : if self . verbose : logger . info ( f'Preparing {external_sample_id}' ) rows : Union [ Dict [ str , str ] , List [ Dict [ str , str ] ] ] = sample_map [ external_sample_id ] if isinstance ( rows , list ) and len ( rows ) == 1 : rows = rows [ 0 ] collapsed_sequencing_meta = self . get_sequence_meta ( external_sample_id , rows ) collapsed_sample_meta = self . get_sample_meta ( external_sample_id , rows ) collapsed_qc = self . get_qc_meta ( external_sample_id , rows ) collapsed_analyses = self . get_analyses ( external_sample_id , rows , cpg_id = existing_external_id_to_cpgid . get ( external_sample_id ) ) sample_type = self . get_sample_type ( external_sample_id , rows ) sequence_status = self . get_sequence_status ( external_sample_id , rows ) if external_sample_id not in existing_external_id_to_cpgid : samples_to_add . append ( NewSample ( external_id = external_sample_id , type = SampleType ( sample_type ) , meta = collapsed_sample_meta , ) ) else : cpgid = existing_external_id_to_cpgid [ external_sample_id ] samples_to_update [ cpgid ] = SampleUpdateModel ( meta = collapsed_sample_meta , ) if collapsed_analyses : analyses_to_add [ external_sample_id ] . extend ( collapsed_analyses ) if collapsed_qc : analyses_to_add [ external_sample_id ] . append ( AnalysisModel ( sample_ids = [ '<none>' ] , type = AnalysisType ( 'qc' ) , status = AnalysisStatus ( 'completed' ) , meta = collapsed_qc , ) ) if collapsed_sequencing_meta : if ( external_sample_id in existing_external_id_to_cpgid and existing_external_id_to_cpgid [ external_sample_id ] in existing_cpgid_to_seq_id ) : cpgid = existing_external_id_to_cpgid [ external_sample_id ] seq_id = existing_cpgid_to_seq_id [ cpgid ] sequences_to_update [ seq_id ] = SequenceUpdateModel ( meta = collapsed_sequencing_meta , status = SequenceStatus ( sequence_status ) ) else : sequences_to_add [ external_sample_id ] = NewSequence ( sample_id = '<None>' , meta = collapsed_sequencing_meta , type = SequenceType ( 'wgs' ) , status = SequenceStatus ( sequence_status ) , ) message = f"""\
Processing samples: {', '.join(sample_map.keys())}

Adding {len(samples_to_add)} samples
Adding {len(sequences_to_add)} sequences
Adding {len(analyses_to_add)} analysis results

Updating {len(samples_to_update)} sample
Updating {len(sequences_to_update)} sequences""" if dry_run : logger . info ( 'Dry run, so returning without inserting / updating metadata' ) return ( samples_to_add , sequences_to_add , samples_to_update , sequences_to_update , analyses_to_add , ) if confirm : resp = str ( input ( message + '\n\nConfirm (y): ' ) ) if resp . lower ( ) != 'y' : raise SystemExit ( ) else : logger . info ( message ) logger . info ( f'Adding {len(samples_to_add)} samples to SM-DB database' ) added_ext_sample_to_internal_id = { } for new_sample in samples_to_add : sample_id = sapi . create_new_sample ( project = self . sample_metadata_project , new_sample = new_sample ) added_ext_sample_to_internal_id [ new_sample . external_id ] = sample_id existing_external_id_to_cpgid [ new_sample . external_id ] = sample_id logger . info ( f'Adding {len(sequences_to_add)} sequence entries' ) for sample_id , seq in sequences_to_add . items ( ) : seq . sample_id = existing_external_id_to_cpgid [ sample_id ] seqapi . create_new_sequence ( new_sequence = seq ) logger . info ( f'Adding analysis entries for {len(analyses_to_add)} samples' ) for sample_id , analyses in analyses_to_add . items ( ) : for analysis in analyses : analysis . sample_ids = [ existing_external_id_to_cpgid [ sample_id ] ] analysisapi . create_new_analysis ( project = self . sample_metadata_project , analysis_model = analysis ) logger . info ( f'Updating {len(samples_to_update)} samples' ) for internal_sample_id , sample_update in samples_to_update . items ( ) : sapi . update_sample ( id_ = internal_sample_id , sample_update_model = sample_update , ) logger . info ( f'Updating {len(sequences_to_update)} sequences' ) for i , ( seq_id , seq_update ) in enumerate ( sequences_to_update . items ( ) ) : try : seqapi . update_sequence ( sequence_id = seq_id , sequence_update_model = seq_update , ) logger . info ( f'#{i+1}: updating sequence' ) except ApiException : traceback . print_exc ( ) logger . error ( f'#{i+1}: error updating sequence {seq_id}: {seq_update}' ) return added_ext_sample_to_internal_id
Parse manifest from iterable (file pointer / String.IO)
Returns a dict mapping external sample ID to CPG sample ID
Parse manifest from iterable ( file pointer / String . IO ) Returns a dict mapping external sample ID to CPG sample ID
# pylint: disable=too-many-branches """
        Parse manifest from iterable (file pointer / String.IO)

        Returns a dict mapping external sample ID to CPG sample ID
        """ # a sample has many rows # now we can start adding!! # determine if any samples exist # by external_sample_id # now we have sample / sequencing meta across 4 different rows, so collapse them # it already exists # ignore QC results if sample already exists # should we add or update sequencing # update, we have a cpg sample ID AND a sequencing ID # the internal sample ID gets mapped back later # keep the type initialisation happy
self
file_pointer
delimiter
confirm
dry_run