style: reduce return statements and branches in searcher methods

- Refactor search_by_sequence in ncbi_searcher.py to reduce return statements from 7 to 1
- Refactor search_by_sequence in rnacentral_searcher.py to reduce return statements from 8 to 1 and branches from 16 to 12
- Extract helper methods to improve code readability and maintainability
- Fix pylint errors R0911 (too-many-return-statements) and R0912 (too-many-branches)

Author: CHERRY-ui8

graphgen/models/searcher/db/ncbi_searcher.py

@@ -569,6 +569,45 @@ def _local_blast(self, seq: str, threshold: float) -> Optional[str]:
             logger.error("Local blastn failed: %s", exc)
             return None
 
+    def _extract_and_normalize_sequence(self, sequence: str) -> Optional[str]:
+        """Extract and normalize DNA sequence from input."""
+        if sequence.startswith(">"):
+            seq_lines = sequence.strip().split("\n")
+            seq = "".join(seq_lines[1:])
+        else:
+            seq = sequence.strip().replace(" ", "").replace("\n", "")
+        return seq if seq and re.fullmatch(r"[ATCGN\s]+", seq, re.I) else None
+
+    def _process_network_blast_result(self, blast_record, seq: str, threshold: float) -> Optional[dict]:
+        """Process network BLAST result and return dictionary or 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
+        accession_match = re.search(r"ref\|([^|]+)", hit_id)
+        if accession_match:
+            accession = accession_match.group(1).split(".")[0]
+            return self.get_by_accession(accession)
+
+        # If unable to extract accession, return basic information
+        return {
+            "molecule_type": "DNA",
+            "database": "NCBI",
+            "id": hit_id,
+            "title": best_alignment.title,
+            "sequence_length": len(seq),
+            "e_value": best_hsp.expect,
+            "identity": best_hsp.identities / best_hsp.align_length if best_hsp.align_length > 0 else 0,
+            "url": f"https://www.ncbi.nlm.nih.gov/nuccore/{hit_id}",
+        }
+
     def search_by_sequence(self, sequence: str, threshold: float = 0.01) -> Optional[dict]:
         """
         Search NCBI with a DNA sequence using BLAST.
@@ -577,77 +616,40 @@ def search_by_sequence(self, sequence: str, threshold: float = 0.01) -> Optional
         :param threshold: E-value threshold for BLAST search.
         :return: A dictionary containing the best hit information or None if not found.
         """
+        result = None
         try:
-            # Extract sequence (if in FASTA format)
-            if sequence.startswith(">"):
-                seq_lines = sequence.strip().split("\n")
-                seq = "".join(seq_lines[1:])
-            else:
-                seq = sequence.strip().replace(" ", "").replace("\n", "")
-
-            # Validate sequence
-            if not seq or not re.fullmatch(r"[ATCGN\s]+", seq, re.I):
-                if not seq:
-                    logger.error("Empty DNA sequence provided.")
-                else:
-                    logger.error("Invalid DNA sequence provided.")
+            seq = self._extract_and_normalize_sequence(sequence)
+            if not seq:
+                logger.error("Empty or invalid DNA sequence provided.")
                 return None
 
             # Try local BLAST first if enabled
-            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)
-
-            # Fall back to network BLAST
-            logger.debug("Falling back to NCBIWWW.qblast.")
-            logger.debug("Performing BLAST search for DNA sequence...")
-            time.sleep(0.35)
-
-            result_handle = NCBIWWW.qblast(
-                program="blastn",
-                database="nr",
-                sequence=seq,
-                hitlist_size=1,
-                expect=threshold,
-            )
-            blast_record = NCBIXML.read(result_handle)
-
-            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
-
-            # Extract accession number
-            # Format may be: gi|123456|ref|NM_000546.5|
-            accession_match = re.search(r"ref\|([^|]+)", hit_id)
-            if accession_match:
-                accession = accession_match.group(1).split(".")[0]
-                return self.get_by_accession(accession)
-            # If unable to extract accession, return basic information
-            return {
-                "molecule_type": "DNA",
-                "database": "NCBI",
-                "id": hit_id,
-                "title": best_alignment.title,
-                "sequence_length": len(seq),
-                "e_value": best_hsp.expect,
-                "identity": best_hsp.identities / best_hsp.align_length if best_hsp.align_length > 0 else 0,
-                "url": f"https://www.ncbi.nlm.nih.gov/nuccore/{hit_id}",
-            }
+                    result = self.get_by_accession(accession)
+
+            # Fall back to network BLAST if local BLAST didn't find result
+            if not result:
+                logger.debug("Falling back to NCBIWWW.qblast.")
+                logger.debug("Performing BLAST search for DNA sequence...")
+                time.sleep(0.35)
+
+                result_handle = NCBIWWW.qblast(
+                    program="blastn",
+                    database="nr",
+                    sequence=seq,
+                    hitlist_size=1,
+                    expect=threshold,
+                )
+                blast_record = NCBIXML.read(result_handle)
+                result = self._process_network_blast_result(blast_record, seq, threshold)
         except RequestException:
             raise
         except Exception as e:  # pylint: disable=broad-except
             logger.error("BLAST search failed: %s", e)
-            return None
+        return result
 
     @retry(
         stop=stop_after_attempt(5),

graphgen/models/searcher/db/rnacentral_searcher.py

@@ -339,6 +339,49 @@ def _local_blast(self, seq: str, threshold: float) -> Optional[str]:
             logger.error("Local blastn failed: %s", exc)
             return None
 
+    @staticmethod
+    def _extract_and_normalize_sequence(sequence: str) -> Optional[str]:
+        """Extract and normalize RNA sequence from input."""
+        if sequence.startswith(">"):
+            seq_lines = sequence.strip().split("\n")
+            seq = "".join(seq_lines[1:])
+        else:
+            seq = sequence.strip().replace(" ", "").replace("\n", "")
+        return seq if seq and re.fullmatch(r"[AUCGN\s]+", seq, re.I) else None
+
+    def _find_best_match_from_results(self, results: List[Dict], seq: str) -> Optional[Dict]:
+        """Find best match from search results, preferring exact match."""
+        exact_match = None
+        for result_item in results:
+            result_seq = result_item.get("sequence", "")
+            if result_seq == seq:
+                exact_match = result_item
+                break
+        return exact_match if exact_match else (results[0] if results else None)
+
+    async def _process_api_search_results(
+        self, results: List[Dict], seq: str
+    ) -> Optional[dict]:
+        """Process API search results and return dictionary or None."""
+        if not results:
+            logger.info("No results found for sequence.")
+            return None
+
+        target_result = self._find_best_match_from_results(results, seq)
+        if not target_result:
+            return None
+
+        rna_id = target_result.get("rnacentral_id")
+        if not rna_id:
+            return None
+
+        # Try to get complete information
+        result = await self.get_by_rna_id(rna_id)
+        if not result:
+            logger.debug("get_by_rna_id() failed for %s, using search result data", rna_id)
+            result = self._rna_data_to_dict(rna_id, target_result)
+        return result
+
     async def search_by_sequence(self, sequence: str, threshold: float = 0.01) -> Optional[dict]:
         """
         Search RNAcentral with an RNA sequence.
@@ -348,86 +391,47 @@ async def search_by_sequence(self, sequence: str, threshold: float = 0.01) -> Op
         :param threshold: E-value threshold for BLAST search.
         :return: A dictionary containing complete RNA information or None if not found.
         """
+        result = None
         try:
-            # Extract sequence (if in FASTA format)
-            if sequence.startswith(">"):
-                seq_lines = sequence.strip().split("\n")
-                seq = "".join(seq_lines[1:])
-            else:
-                seq = sequence.strip().replace(" ", "").replace("\n", "")
-
-            # Validate if it's an RNA sequence (contains U instead of T)
-            if not re.fullmatch(r"[AUCGN\s]+", seq, re.I):
-                logger.error("Invalid RNA sequence provided.")
-                return None
-
+            seq = self._extract_and_normalize_sequence(sequence)
             if not seq:
-                logger.error("Empty RNA sequence provided.")
+                logger.error("Empty or invalid RNA sequence provided.")
                 return None
 
             # Try local BLAST first if enabled
             if self.use_local_blast:
                 accession = self._local_blast(seq, threshold)
                 if accession:
                     logger.debug("Local BLAST found accession: %s", accession)
-                    # Try to get RNA ID from accession (may need conversion)
-                    # For now, try using accession as RNA ID or search by it
                     result = await self.get_by_rna_id(accession)
-                    if result:
-                        return result
-                    # If not found by ID, try keyword search
-                    result = await self.get_best_hit(accession)
-                    if result:
-                        return result
-
-            # Fall back to RNAcentral API
-            logger.debug("Falling back to RNAcentral API.")
-            async with aiohttp.ClientSession() as session:
-                search_url = f"{self.base_url}/rna"
-                params = {"sequence": seq, "format": "json"}
-                async with session.get(
-                    search_url,
-                    params=params,
-                    headers=self.headers,
-                    timeout=aiohttp.ClientTimeout(total=60),  # Sequence search may take longer
-                ) as resp:
-                    if resp.status == 200:
-                        search_results = await resp.json()
-                        results = search_results.get("results", [])
-                        if results:
-                            # Step 1: Find best match (prefer exact match)
-                            exact_match = None
-                            for result in results:
-                                result_seq = result.get("sequence", "")
-                                if result_seq == seq:
-                                    exact_match = result
-                                    break
-
-                            # Use exact match if found, otherwise use first result
-                            target_result = exact_match if exact_match else results[0]
-                            rna_id = target_result.get("rnacentral_id")
-
-                            if rna_id:
-                                # Step 2: Unified call to get_by_rna_id() for complete information
-                                result = await self.get_by_rna_id(rna_id)
-
-                                # Step 3: If get_by_rna_id() failed, use search result data as fallback
-                                if not result:
-                                    logger.debug("get_by_rna_id() failed for %s, using search result data", rna_id)
-                                    result = self._rna_data_to_dict(rna_id, target_result)
-
-                                return result
-                        logger.info("No results found for sequence.")
-                        return None
-                    error_text = await resp.text()
-                    logger.error("HTTP %d error for sequence search: %s", resp.status, error_text[:200])
-                    raise Exception(f"HTTP {resp.status}: {error_text}")
+                    if not result:
+                        result = await self.get_best_hit(accession)
+
+            # Fall back to RNAcentral API if local BLAST didn't find result
+            if not result:
+                logger.debug("Falling back to RNAcentral API.")
+                async with aiohttp.ClientSession() as session:
+                    search_url = f"{self.base_url}/rna"
+                    params = {"sequence": seq, "format": "json"}
+                    async with session.get(
+                        search_url,
+                        params=params,
+                        headers=self.headers,
+                        timeout=aiohttp.ClientTimeout(total=60),  # Sequence search may take longer
+                    ) as resp:
+                        if resp.status == 200:
+                            search_results = await resp.json()
+                            results = search_results.get("results", [])
+                            result = await self._process_api_search_results(results, seq)
+                        else:
+                            error_text = await resp.text()
+                            logger.error("HTTP %d error for sequence search: %s", resp.status, error_text[:200])
+                            raise Exception(f"HTTP {resp.status}: {error_text}")
         except aiohttp.ClientError as e:
             logger.error("Network error searching for sequence: %s", e)
-            return None
         except Exception as e:  # pylint: disable=broad-except
             logger.error("Sequence search failed: %s", e)
-            return None
+        return result
 
     @retry(
         stop=stop_after_attempt(3),