fix: disable API fallback when local BLAST is enabled

Author: CHERRY-ui8

graphgen/models/searcher/db/ncbi_searcher.py

@@ -393,11 +393,18 @@ def _process_network_blast_result(blast_record, seq: str, threshold: float) -> O
                 return None
 
             # Try local BLAST first if enabled
-            if self.use_local_blast and (accession := self._local_blast(seq, threshold)):
-                logger.debug("Local BLAST found accession: %s", accession)
-                return self.get_by_accession(accession)
+            if self.use_local_blast:
+                accession = self._local_blast(seq, threshold)
+                if accession:
+                    logger.debug("Local BLAST found accession: %s", accession)
+                    return self.get_by_accession(accession)
+                logger.info(
+                    "Local BLAST found no match for sequence. "
+                    "API fallback disabled when using local database."
+                )
+                return None
 
-            # Fall back to network BLAST
+            # Fall back to network BLAST only if local BLAST is not enabled
             logger.debug("Falling back to NCBIWWW.qblast")
 
             with NCBIWWW.qblast("blastn", "nr", seq, hitlist_size=1, expect=threshold) as result_handle:

graphgen/models/searcher/db/rnacentral_searcher.py

@@ -255,8 +255,13 @@ def _extract_sequence(sequence: str) -> Optional[str]:
                 if accession:
                     logger.debug("Local BLAST found accession: %s", accession)
                     return self.get_by_rna_id(accession)
+                logger.info(
+                    "Local BLAST found no match for sequence. "
+                    "API fallback disabled when using local database."
+                )
+                return None
 
-            # Fall back to RNAcentral API if local BLAST didn't find result
+            # Fall back to RNAcentral API only if local BLAST is not enabled
             logger.debug("Falling back to RNAcentral API.")
 
             md5_hash = self._calculate_md5(seq)
@@ -272,11 +277,13 @@ def _extract_sequence(sequence: str) -> Optional[str]:
             if not results:
                 logger.info("No exact match found in RNAcentral for sequence")
                 return None
+
             rna_id = results[0].get("rnacentral_id")
-            if not rna_id:
-                logger.error("No RNAcentral ID found in search results.")
-                return None
-            return self.get_by_rna_id(rna_id)
+            if rna_id:
+                return self.get_by_rna_id(rna_id)
+
+            logger.error("No RNAcentral ID found in search results.")
+            return None
         except Exception as e:
             logger.error("Sequence search failed: %s", e)
             return None

graphgen/models/searcher/db/uniprot_searcher.py

@@ -124,47 +124,52 @@ def get_by_fasta(self, fasta_sequence: str, threshold: float) -> Optional[Dict]:
             logger.error("Empty FASTA sequence provided.")
             return None
 
-        accession = None
         if self.use_local_blast:
             accession = self._local_blast(seq, threshold)
             if accession:
                 logger.debug("Local BLAST found accession: %s", accession)
+                return self.get_by_accession(accession)
+            logger.info(
+                "Local BLAST found no match for sequence. "
+                "API fallback disabled when using local database."
+            )
+            return None
 
-        if not accession:
-            logger.debug("Falling back to NCBIWWW.qblast.")
+        # Fall back to network BLAST only if local BLAST is not enabled
+        logger.debug("Falling back to NCBIWWW.qblast.")
 
-            # UniProtKB/Swiss-Prot BLAST API
-            try:
-                logger.debug(
-                    "Performing BLAST searcher for the given sequence: %s", seq
-                )
-                result_handle = NCBIWWW.qblast(
-                    program="blastp",
-                    database="swissprot",
-                    sequence=seq,
-                    hitlist_size=1,
-                    expect=threshold,
-                )
-                blast_record = NCBIXML.read(result_handle)
-            except RequestException:
-                raise
-            except Exception as e:  # pylint: disable=broad-except
-                logger.error("BLAST searcher failed: %s", e)
-                return None
+        # UniProtKB/Swiss-Prot BLAST API
+        try:
+            logger.debug(
+                "Performing BLAST searcher for the given sequence: %s", seq
+            )
+            result_handle = NCBIWWW.qblast(
+                program="blastp",
+                database="swissprot",
+                sequence=seq,
+                hitlist_size=1,
+                expect=threshold,
+            )
+            blast_record = NCBIXML.read(result_handle)
+        except RequestException:
+            raise
+        except Exception as e:  # pylint: disable=broad-except
+            logger.error("BLAST searcher failed: %s", e)
+            return None
 
-            if not blast_record.alignments:
-                logger.info("No BLAST hits found for the given sequence.")
-                return None
+        if not blast_record.alignments:
+            logger.info("No BLAST hits found for the given sequence.")
+            return None
 
-            best_alignment = blast_record.alignments[0]
-            best_hsp = best_alignment.hsps[0]
-            if best_hsp.expect > threshold:
-                logger.info("No BLAST hits below the threshold E-value.")
-                return None
-            hit_id = best_alignment.hit_id
+        best_alignment = blast_record.alignments[0]
+        best_hsp = best_alignment.hsps[0]
+        if best_hsp.expect > threshold:
+            logger.info("No BLAST hits below the threshold E-value.")
+            return None
 
-            # like sp|P01308.1|INS_HUMAN
-            accession = hit_id.split("|")[1].split(".")[0] if "|" in hit_id else hit_id
+        # like sp|P01308.1|INS_HUMAN
+        hit_id = best_alignment.hit_id
+        accession = hit_id.split("|")[1].split(".")[0] if "|" in hit_id else hit_id
         return self.get_by_accession(accession)
 
     def _local_blast(self, seq: str, threshold: float) -> Optional[str]: