Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

38 changes: 31 additions & 7 deletions protea/core/operations/_protein_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,22 @@ def add(self, other: StoreCounts) -> None:


def store_records(
session: Session, records: list[UniProtProteinRecord], emit: EmitFn
session: Session,
records: list[UniProtProteinRecord],
emit: EmitFn,
*,
asserts_reviewed: bool = True,
) -> StoreCounts:
"""Upsert ``records`` as ``sequence`` + ``protein`` rows.

``asserts_reviewed=False`` says the source cannot answer the Swiss-Prot
question, so ``reviewed`` is left NULL instead of being filled from the
record. UniParc is such a source: it keeps sequences, not entry status, and
an accession it answers for is one UniProtKB no longer serves, so there is
no status to read. Writing ``False`` there would be indistinguishable from a
TrEMBL entry UniProt actually called unreviewed, and NULL already means the
thing that is true: nobody asked, or nobody could answer.

Does NOT commit: the caller owns the transaction boundary, because the two
callers want different ones -- ``insert_proteins`` commits per page so an
interrupted ingest leaves a partial-but-consistent dataset, and
Expand Down Expand Up @@ -129,6 +141,8 @@ def _upsert_proteins(
existing_prot: dict[str, Protein],
existing_seq_ids: dict[str, int],
emit: EmitFn,
*,
asserts_reviewed: bool = True,
) -> tuple[int, int]:
"""Insert new ``Protein`` rows and conservatively patch the existing ones."""
proteins_inserted = 0
Expand All @@ -137,10 +151,12 @@ def _upsert_proteins(
for r in records:
seq_id = existing_seq_ids[r.sequence_hash]
if r.accession in existing_prot:
if apply_record_updates(existing_prot[r.accession], r, seq_id):
if apply_record_updates(
existing_prot[r.accession], r, seq_id, asserts_reviewed=asserts_reviewed
):
proteins_updated += 1
else:
to_add.append(build_protein(r, seq_id))
to_add.append(build_protein(r, seq_id, asserts_reviewed=asserts_reviewed))
proteins_inserted += 1
if to_add:
emit("db.insert_proteins_start", None, {"rows": len(to_add)}, "info")
Expand All @@ -150,7 +166,13 @@ def _upsert_proteins(
return proteins_inserted, proteins_updated


def apply_record_updates(p: Protein, r: UniProtProteinRecord, seq_id: int | None) -> bool:
def apply_record_updates(
p: Protein,
r: UniProtProteinRecord,
seq_id: int | None,
*,
asserts_reviewed: bool = True,
) -> bool:
"""Patch ``p`` from ``r`` without clobbering; return ``True`` if anything changed.

The null-filling is what lets the universe be built in two steps:
Expand All @@ -174,7 +196,7 @@ def apply_record_updates(p: Protein, r: UniProtProteinRecord, seq_id: int | None
if getattr(p, "isoform_index", None) != r.isoform_index:
p.isoform_index = r.isoform_index
changed = True
if getattr(p, "reviewed", None) is None:
if asserts_reviewed and getattr(p, "reviewed", None) is None:
p.reviewed = r.reviewed
changed = True
if getattr(p, "taxonomy_id", None) in (None, "") and r.taxonomy_id:
Expand All @@ -192,14 +214,16 @@ def apply_record_updates(p: Protein, r: UniProtProteinRecord, seq_id: int | None
return changed


def build_protein(r: UniProtProteinRecord, seq_id: int | None) -> Protein:
def build_protein(
r: UniProtProteinRecord, seq_id: int | None, *, asserts_reviewed: bool = True
) -> Protein:
"""A new ``Protein`` row from a UniProt record."""
return Protein(
accession=r.accession,
canonical_accession=r.canonical_accession,
is_canonical=r.is_canonical,
isoform_index=r.isoform_index,
reviewed=r.reviewed,
reviewed=r.reviewed if asserts_reviewed else None,
entry_name=r.entry_name,
organism=r.organism,
taxonomy_id=r.taxonomy_id,
Expand Down
18 changes: 14 additions & 4 deletions protea/core/operations/_universe_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,9 +85,7 @@
#: whether an embedding neighbourhood captures what curated alignment captures,
#: without touching the scored truth. Measured on GOA 156: 62,363 proteins enter
#: by these and by nothing else.
CURATED_INFERENCE_CODES = frozenset(
{"ISS", "ISO", "ISA", "ISM", "IGC", "RCA", "NAS", "IKR", "IRD"}
)
CURATED_INFERENCE_CODES = frozenset({"ISS", "ISO", "ISA", "ISM", "IGC", "RCA", "NAS", "IKR", "IRD"})

#: Excluded, each for its own reason, and the reasons are not interchangeable.
#:
Expand Down Expand Up @@ -344,6 +342,11 @@ class _FetchOutcome:
dates_written: int = 0
alias: dict[str, str] = field(default_factory=dict)
unresolved: list[str] = field(default_factory=list)
#: Accessions UniProtKB no longer serves and UniParc did. Kept apart from
#: ``fetched`` because they are a different claim: ``fetched`` means UniProt
#: answered for a live entry, this means the entry is gone and the sequence
#: was recovered from the archive.
recovered_from_uniparc: list[str] = field(default_factory=list)
demerges: dict[str, list[str]] = field(default_factory=dict)
artefactos: dict[str, Any] = field(default_factory=dict)

Expand Down Expand Up @@ -442,6 +445,12 @@ def resolution_report(
``None`` in a dry run rather than ``0``, because "zero recoverable" and "not
attempted" are not the same thing: an earlier report read 0% recoverable merges
because ten batches had answered 400 and the failures were tallied as zeroes.

``dates_pending`` broke that rule and read ``0`` in a dry run while 644.659
rows were waiting, because the pass that counts them is the one a dry run
skips. A dry run is exactly where somebody decides whether the real one is
worth starting, so the one field that said "nothing to do" was the one that
could not know.
"""
return {
"candidates": outcome.candidates,
Expand All @@ -450,9 +459,10 @@ def resolution_report(
"proteins_inserted": outcome.inserted,
"sequences_inserted": outcome.sequences,
"resolved_as_merge": None if dry_run else len(outcome.alias),
"recovered_from_uniparc": None if dry_run else len(outcome.recovered_from_uniparc),
"not_retrievable": None if dry_run else len(outcome.unresolved),
"demerged": None if dry_run else len(outcome.demerges),
"dates_pending": outcome.dates_pending,
"dates_pending": None if dry_run else outcome.dates_pending,
"dates_backfilled": None if dry_run else outcome.dates_written,
"artefactos": outcome.artefactos,
"dry_run": dry_run,
Expand Down
142 changes: 137 additions & 5 deletions protea/core/operations/resolve_protein_sequences.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@
import time
from typing import Annotated, Any

from protea_contracts import UniProtProteinRecord
from protea_contracts import UniProtProteinRecord, compute_sequence_hash
from pydantic import Field
from sqlalchemy import select
from sqlalchemy.orm import Session
Expand Down Expand Up @@ -194,6 +194,7 @@ def execute(
if not p.dry_run and candidates:
self._fetch_sequences(session, candidates, p, emit, outcome)
self._resolve_merges(session, candidates, p, emit, outcome)
self._resolve_retired(session, candidates, p, emit, outcome)
if not p.dry_run:
self._backfill_dates(session, p, emit, outcome)
outcome.artefactos = self._store_artifacts(payload.get("_job_id"), outcome)
Expand Down Expand Up @@ -328,9 +329,7 @@ def _merge_batch(
accession that appears in SEVERAL entries is a demerge, not a merge, and
that cannot be decided until the whole response has been read.
"""
body = self._uniprot.search_secondary_accessions(
batch, emit=emit, knobs=self._knobs(p)
)
body = self._uniprot.search_secondary_accessions(batch, emit=emit, knobs=self._knobs(p))
candidates, entry_by_acc = candidates_from(body, set(batch))
records = classify(candidates, outcome.alias, outcome.demerges)
counts = StoreCounts()
Expand Down Expand Up @@ -358,6 +357,139 @@ def _merge_batch(
)
return counts

# ---- pass three: what UniProtKB no longer serves, from UniParc ----

def _resolve_retired(
self,
session: Session,
candidates: list[str],
p: ResolveProteinSequencesPayload,
emit: EmitFn,
outcome: _FetchOutcome,
) -> None:
"""Rescue from UniParc the accessions UniProtKB has stopped serving.

The two passes before this one ask UniProtKB, and UniProtKB answers
about entries it still has. An accession it deleted is absent from the
batch endpoint while ``X-Total-Results`` counts it, and its own URL
answers 200 with an empty body -- so after the merge pass what is left
is not "unknown", it is "gone from UniProtKB". UniParc keeps the
sequence of every entry UniProtKB has ever held, deleted ones included.
Measured 2026-10-11 over a random sample of 600 of the 28.868 a full
pass could not place: 600 resolved.

WHY IT IS WORTH A PASS. The ones missing are not missing at random. By
tier over the 74 releases, 9,2% of the proteins that EVER carried
experimental evidence have no sequence, and 7,8% of the curated
inference tier, against 0,07% of the rest. In an early cut that is 7,5%
of the donor bank and in the last one it is 0%, so the hole is a
visibility gradient that looks exactly like the bank growing -- which is
one of the two things a longitudinal dataset exists to measure.
"""
from protea_sources.uniprot import MAX_UNIPARC_OR_CONDITIONS

pending = self._still_without_sequence(session, candidates)
if not pending:
return
counts = StoreCounts()
recuperadas: set[str] = set()
for batch in chunks(pending, MAX_UNIPARC_OR_CONDITIONS):
counts.add(self._uniparc_batch(session, batch, p, emit, recuperadas))
outcome.recovered_from_uniparc = sorted(recuperadas)
outcome.unresolved = sorted(set(pending) - recuperadas)
outcome.updated += counts.proteins_updated
outcome.inserted += counts.proteins_inserted
outcome.sequences += counts.sequences_inserted
emit(
"resolve_protein_sequences.uniparc",
None,
{
"pending": len(pending),
"recovered": len(recuperadas),
"unresolved": len(outcome.unresolved),
},
"info",
)

@staticmethod
def _newest_per_accession(tsv: str, pedidas: set[str]) -> dict[str, Any]:
"""The newest UniParc row for each accession that was asked for.

Both filters are here because both are decisions. An accession answers
one row per sequence version it ever had, and the one with the latest
``last_seen`` is the sequence it held when it disappeared. A row answers
for every accession sharing that exact sequence, and only the ones in
this batch are ours to write: the rest may be perfectly alive entries
belonging to somebody else's question.

The version suffix is stripped for MATCHING only (``A0A014NDJ0.1`` is
the accession ``A0A014NDJ0``), and the row keeps it, so nothing is lost.
"""
from protea_sources.uniprot import parse_uniparc_tsv

mejor: dict[str, Any] = {}
for fila in parse_uniparc_tsv(tsv):
for cruda in fila.accessions:
acc = cruda.split(".")[0]
if acc not in pedidas:
continue
previa = mejor.get(acc)
if previa is None or fila.last_seen > previa.last_seen:
mejor[acc] = fila
return mejor

def _uniparc_batch(
self,
session: Session,
batch: list[str],
p: ResolveProteinSequencesPayload,
emit: EmitFn,
recuperadas: set[str],
) -> StoreCounts:
"""One UniParc query, and the two choices it forces.

WHICH ROW. An accession can answer several, one per sequence version
UniParc has ever seen: 670 rows came back for a sample of 600. This
takes the one with the latest ``last_seen``, which is the sequence the
accession held when it disappeared. The alternative -- the version
current at each cut -- would be a better answer and a different corpus:
every other protein here carries the sequence UniProt serves TODAY, so
picking the newest keeps the whole store on one convention instead of
making these 28.868 the only ones with a second rule. The dates are
recorded so a later pass can revisit that without asking UniParc again.

WHICH ACCESSION. Identical sequences share a UPI, so one row answers for
entries nobody asked about. Only the ones in this batch are written: the
rest are somebody else's accessions and may be perfectly alive.

``reviewed`` is left NULL on purpose, see ``store_records``.
"""
tsv = self._uniprot.search_uniparc_tsv(
batch,
fields="upi,accession,first_seen,last_seen,length,sequence",
emit=emit,
knobs=self._knobs(p),
)
mejor = self._newest_per_accession(tsv, set(batch))
if not mejor:
return StoreCounts()
records = [
UniProtProteinRecord(
accession=acc,
canonical_accession=acc,
is_canonical=True,
reviewed=False,
sequence=fila.sequence,
length=len(fila.sequence),
sequence_hash=compute_sequence_hash(fila.sequence),
)
for acc, fila in sorted(mejor.items())
]
counts = store_records(session, records, emit, asserts_reviewed=False)
recuperadas.update(mejor)
session.commit()
return counts

def _still_without_sequence(self, session: Session, candidates: list[str]) -> list[str]:
pending: list[str] = []
for chunk in chunks(candidates, _DB_CHUNK):
Expand All @@ -371,7 +503,7 @@ def _still_without_sequence(self, session: Session, candidates: list[str]) -> li
)
return sorted(pending)

# ---- pass three: the dates of everything that already had a sequence ----
# ---- pass four: the dates of everything that already had a sequence ----

def _backfill_dates(
self,
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ priority = "explicit"
# protea.core.contracts.operation); the remaining four are optional
# plugin extras installed via --with plugins.
protea-method = { git = "https://github.com/frapercan/protea-method.git", rev = "a120f613b9d2f1646ab5a566f97a8eadc1310a23" }
protea-sources = { git = "https://github.com/frapercan/protea-sources.git", rev = "946227ca3b8840eb0ea7223c8fc52b746027f9fe" }
protea-sources = { git = "https://github.com/frapercan/protea-sources.git", rev = "fc43ab942ebec811afae25eda3be18b3f37d0e46" }
protea-runners = { git = "https://github.com/frapercan/protea-runners.git", rev = "ccd06fd80c12b5195339478dae7255d1ea58f4a6" }
protea-backends = { git = "https://github.com/frapercan/protea-backends.git", rev = "9b76719a14d729390be7d0faf957fae06a31ced5" }

Expand Down
Loading
Loading