alab_management.sample_view.analysis_view module

Analysis results and analysis claims on samples: the one implementation behind the Data API.

This is the in-process half of decision D7/D9 in alab_one’s docs/data-architecture.md: an offline analysis service consumes artifacts and writes its results back through AlabOS onto the sample, and the HTTP routes in alab_management/dashboard/routes/data_api.py (/api/data/analysis/...) are only an adapter over this class, so an in-process AlabOS caller and a remote client go through the same rules.

What “one write path” does and does not mean here, spelled out because the distinction is exactly what a reviewer should check:

• This class never builds a Mongo client of its own. Every write to the live database uses the collection handle of the SampleView the caller hands in.

• The unconditional result write (no owner) goes through SampleView.update_sample_metadata.

• The claim-conditional result write, and every claim operation, issue their own update_one on that same collection handle, because update_sample_metadata is unconditional and cannot express “only if I still hold the claim”. They stamp last_updated the same way it does.

• _mirror_to_completed writes the completed database through CompletedSampleView, which is a second collection the caller did not hand in. _update is the only other place this class reaches past the given SampleView: once a sample’s live copy has been pruned after archiving, its archived copy is the only one left, so claims and results are applied there instead.

Three operations, matching what an analysis worker needs each cycle:

• pending_samples which samples have input data for a technique but no result matching my config

• claim / release so two workers never analyze the same sample (a crashed worker’s claim expires on its own)

• write_result store metadata.analysis.<kind> additively, optionally only while still holding the claim

Both documents live under metadata, side by side and never nested in each other:

metadata.analysis.<kind>         = {...the result...}
metadata.analysis_claims.<kind>  = {claimed_by, claimed_at, expires_at, ttl_s}

class AnalysisView(sample_view=None)

Bases: object

Read and write the analysis blocks of samples through an existing SampleView.

claim(sample_id, kind, owner, ttl_s=7200)

Take (or renew) this kind’s claim on one sample; None when another live worker holds it.

Renewing is the same call from the same owner, so a heartbeat is just a repeat: there is no separate renew operation to get wrong, and a claim nobody renews expires at expires_at. Raises SampleNotFoundError when there is no such sample, which is a different thing from losing the race.

Return type:

dict[str, Any] | None

static claim_field(kind)

Field path of one kind’s claim document.

Return type:

str

get_claim(sample_id, kind)

The claim document currently stored for this kind, if any.

Return type:

dict[str, Any] | None

pending_samples(kind, data_source, params_hash, limit=50, owner=None, include_errors=False, include_claimed=False, created_since=None)

Samples that carry data_source’s input data but no result whose params_hash is the caller’s.

params_hash is the caller’s own identifier for its analysis configuration; AlabOS only compares it for equality with what is stored, and never interprets it. A stored block without params_hash (a hand-written result) counts as fresh and is left alone. Errors are skipped unless include_errors, because when to retry an error is the caller’s policy, not AlabOS’s. With created_since, only samples created at or after it are listed (by the creation time in the sample id; a naive time is UTC), so a service that starts on a running lab can leave the samples made before it alone.

Return type:

list[dict[str, Any]]

release(sample_id, kind, owner)

Drop this kind’s claim if owner still holds it. False when it was already taken over.

Return type:

bool

static result_field(kind)

Field path of one kind’s stored result.

Return type:

str

write_result(sample_id, kind, result, owner=None, release_claim=True, mirror_completed=True)

Store metadata.analysis.<kind> on one sample.

Returns {"written": ..., "mirrored": ...}. written is False only when owner was given and that owner no longer holds the claim; mirrored says whether the same key was also set on this sample’s document in the completed database. SampleNotFoundError is raised when there is no such sample, on both paths.

The write is additive: it replaces that one key and touches no other metadata. Without owner it goes through SampleView.update_sample_metadata, which turns {"analysis.<kind>": ...} into a $set of the field path metadata.analysis.<kind>.

With owner the update is conditional, and it is worth being precise about what it checks: it matches on the claim’s current holder, not on expires_at. So a worker that overran its lease still stores its result when nobody took the sample over – deliberate, because a finished analysis should not be discarded just because a lease lapsed – while a worker whose claim was taken over, or whose successor has already released it, matches no claim and gets written: False. The case this cannot defend against is a writer that passes no owner at all, since such a write opts out of claim arbitration by definition.

Return type:

dict[str, Any]

exception SampleNotFoundError

Bases: ValueError

No sample with the given id exists in the live database.

A subclass of ValueError on purpose, so in-process code written against SampleView.update_sample_metadata – which raises a plain ValueError for the same situation – keeps working, while the HTTP adapter can still answer 404 instead of lumping it in with a validation error.

check_kind(kind)

A kind becomes part of a Mongo field path, so only a plain identifier is accepted.

Return type:

str

check_result_document(result)

The stored result must be a non-empty JSON object whose keys are safe as Mongo field names.

Return type:

dict[str, Any]

data_source_query(data_source)

The query that says “this sample carries the input data for that source”.

Built here from the source’s declared field (and optional BSON type), so a config file cannot inject a filter: the worst a bad entry can do is name a field that matches nothing.

Return type:

dict[str, Any]

input_data_sources()

The configured sources, keyed by name, read fresh so a config edit needs no restart.

Returns {} when the section is absent, which is the default: a deployment that has not declared where its patterns live gets no pending query rather than a guess.

Return type:

dict[str, dict[str, Any]]