Spaces:
Sleeping
Sleeping
File size: 4,341 Bytes
76fb273 9be3066 76fb273 fb8c550 76fb273 5d5e978 76fb273 5d5e978 76fb273 5d5e978 76fb273 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | from Bio import Entrez, SeqIO
from io import StringIO
from typing import Optional
from app.config import settings
from app.services.cache import ttl_cache
Entrez.email = "bioflow@example.com"
def _detect_db(accession: str) -> str:
accession = accession.strip().upper()
if accession.startswith(("NP_", "XP_", "YP_", "AP_", "WP_")):
return "protein"
if accession.startswith(("NM_", "XM_", "NR_", "XR_")):
return "nucleotide"
if accession.startswith("NG_"):
return "nucleotide"
if accession.startswith(("NC_", "NT_", "NW_")):
return "nucleotide"
if accession.startswith(("AC_", "AE_")):
return "nucleotide"
return "protein"
def _detect_sequence_type(seq: str) -> str:
clean = seq.upper().replace("-", "").replace(".", "")
if not clean:
return "unknown"
dna_chars = set("ACGTUN")
rna_chars = set("ACGUN")
protein_chars = set("ACDEFGHIKLMNPQRSTVWY")
seq_set = set(clean)
if seq_set.issubset(dna_chars):
if seq_set.intersection({"T", "U"}):
return "dna"
if seq_set.issubset(rna_chars):
return "rna"
if seq_set.issubset(protein_chars):
return "protein"
if seq_set.issubset(dna_chars.union({"N"})):
return "dna"
return "unknown"
class NCBIService:
@ttl_cache(ttl=86400, prefix="ncbi_seq")
async def fetch_by_accession(self, accession: str) -> dict:
accession = accession.strip().upper()
db = _detect_db(accession)
try:
handle = Entrez.efetch(db=db, id=accession, rettype="fasta", retmode="text")
fasta_text = handle.read()
handle.close()
if not fasta_text.strip():
return {"error": f"Accession '{accession}' not found in NCBI"}
record = SeqIO.read(StringIO(fasta_text), "fasta")
seq_str = str(record.seq)
seq_type = _detect_sequence_type(seq_str)
desc = record.description
header_parts = desc.split(" ", 1)
acc_from_header = header_parts[0]
description = header_parts[1] if len(header_parts) > 1 else ""
organism = ""
if "[" in desc and "]" in desc:
organism = desc.split("[")[-1].rstrip("]")
return {
"accession": acc_from_header,
"db_source": "ncbi",
"database": db,
"sequence_type": seq_type,
"sequence": seq_str,
"length": len(seq_str),
"organism": organism,
"description": description,
"from_cache": False,
}
except Exception as e:
return {"error": str(e)}
@ttl_cache(ttl=86400, prefix="ncbi_search")
async def search_by_name(self, term: str, db: str = "protein", max_results: int = 10) -> dict:
try:
handle = Entrez.esearch(db=db, term=term, retmax=max_results)
result = Entrez.read(handle)
handle.close()
ids = result.get("IdList", [])
if not ids:
return {"error": f"No results found for '{term}'", "results": []}
handle = Entrez.esummary(db=db, id=",".join(ids))
summaries = Entrez.read(handle)
handle.close()
results = []
for docsum in summaries:
if hasattr(docsum, "items"):
# Field names differ between databases:
# protein: AccessionVersion, Title, Organism, Length
# nucleotide: Caption (accession), Title, Slen, Organism
# gene: Name, Description, Organism, NomenclatureSymbol
results.append({
"accession": str(docsum.get("Caption", "") or docsum.get("AccessionVersion", "") or ""),
"title": str(docsum.get("Title", "") or docsum.get("Description", "") or docsum.get("Name", "")),
"organism": str(docsum.get("Organism", "") or ""),
"length": int(docsum.get("Slen", 0) or docsum.get("Length", 0) or 0),
})
return {"results": results, "count": len(results), "query": term, "db": db}
except Exception as e:
return {"error": str(e)}
|