---
title: "vxData jobs ingestion patterns"
description: "Writing ingestion jobs in mono apps/vxdata-jobs - discovery, mapping, stable identifiers, linking, and data gotchas."
image: "https://docs.virdx.dev/img/virdx-social-card.png"
---

> Documentation Index
> Fetch the complete documentation index at: https://docs.virdx.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# vxData jobs ingestion patterns

When writing data ingestion jobs in `mono/apps/vxdata-jobs`.

## Ingestion job structure

- Entry tasks live in `apps/vxdata-jobs/pyproject.toml` under `[project.scripts]`.
- `apps/vxdata-jobs/src/vxdata/jobs/ingestion.py` defines the older `Ingestion` ABC (legacy).
- Current pattern: `source_2026_...` and `f_2026...` jobs. Treat `ingest_*` as legacy unless required.
- Each job is typically a subdirectory under `src/vxdata/jobs/` with `main.py`, optional `utils.py`, and helpers.

## File discovery patterns

Three common approaches:

1. **Hierarchical DICOM trees** (`patient/study[/series]`):
   - Use `Path.iterdir()` or `Path.rglob()` for recursive traversal.
   - Examples: `ingest_prostatex/main.py`, `ingest_prus/main.py`.

2. **Filename globbing** for modality/mask files:
   - Match patterns to find anatomy masks, lesion masks, or specific sequences.
   - Example: `f_20260302_lesion_masks/prostatex.py`.

3. **Tabular source files** (`csv`, `xlsx`, `parquet`) for clinical/pathology data:
   - Parse with pandas or polars; join to existing resources via external UIDs.
   - Examples: `f_20260217_pathology/*.py`, `source_2026_01_28_ingest_files/upload.py`.

**Platform-side discovery** is used heavily: query the vxData API for existing `Patient`, `ImagingStudy`, `DICOMSeries`, `Volume`, `VoxelMap` before uploading to resolve foreign keys and avoid duplicates.

## Mapping and resource tree

The standard resource hierarchy:

```
DataSource → Patient → ImagingStudy → child resources (DICOMSeries, Volume, VoxelMap, Measurement, etc.)
```

### Stable identifiers

Build explicit stable identifiers with small dataset-specific helper functions. Use external UIDs from the source dataset whenever possible (`patient_id`, `study_id`, `series_uid`, etc.).

### Metadata sources

- **DICOM headers**: `utils/ingest_dicoms.py` parses tags; `vicom.load_volumes(...)` handles DICOM/NIfTI conversion.
- **CSV/Excel joins**: match clinical/pathology fields by patient/study identifiers.
- **Derived fields**: some jobs compute study-level metadata from mode/frequency across files (e.g., modality, study date).

### Common payload mappers

- DICOM → `ImagingStudy`, `DICOMSeries`
- Converted volumes → `Volume` / `VolumeConversion`
- Tabular rows → `Measurement`, `PatientStatus`, `PIRADSAssessment`, `PIRADSLesionAssessment`, `PathologyAssessment`
- Masks → `VoxelMap` or segmentation `Volume`

## Linking and foreign keys

### Primary tree link

Use `parent_identifier` to build the resource tree (e.g., `ImagingStudy` parent is `Patient`, `DICOMSeries` parent is `ImagingStudy`).

### Resolution helpers

- `client.get_or_create_patient(...)` and `client.get_or_create_study(...)` handle patient/study lookup or creation.
- For batch jobs, rebuild maps from platform `external_uid` by querying all relevant resources first.

### Secondary links (provenance and cross-references)

- `derived_from`: track provenance (`Volume`, `VoxelMap` derived from other volumes).
- `reference_mri`: link masks to their source MRI volume.
- `mapped_pirads_lesion_ref`, `index_lesion_id`: connect lesion/pathology/radiology assessments.
- `pathology_id`: associate `BiopsyCoreLocation` with pathology reports.

## Data gotchas

### Polars `group_by.agg` first() trap

**`pl.col(X).first()` in `group_by.agg` is NOT "first non-null" and is NOT correlated with `pl.col(Y).first()`.**

Each `first()` is an independent column aggregation. When merging per-module rows where each row has NULLs in other columns (e.g., one row with `pirads=5, isup=None`, another with `pirads=None, isup=4`), plain `first()` will return the first value regardless of NULL, losing data.

**Fix**: use `.drop_nulls().first()` per column when aggregating sparse multi-row data:

```python
.group_by("submission_uid", "lesion_uid")
.agg(
pl.col("pirads").drop_nulls().first(),
pl.col("isup").drop_nulls().first(),
)
```

Alternatively, restructure as separate DataFrames and join on the group key. This bug caused silent loss of ISUP scores in VBC-1250 (2026-04-20); the fix is in `f_20260305_vxannotate_integration/main.py` lines 191–196.

### vxAnnotate placeholder scoring rows

**vxAnnotate writes a placeholder `scoring_annotations` row when a lesion is created**, with every score field blank (`T2Score=""`, `DWIScore=""`, `level=""`, `zones=[]`, or `isupGrade` blank/`"unknown"`). The real score arrives in a second row once the annotator submits.

**Skip placeholder rows at parse time.** Do NOT coerce empty strings to `0` via `isup = isup_grade or 0` — that creates a fake score that collides with the real one and corrupts masks.

**Predicate for placeholder**:

- PIRADS module: `T2Score` blank AND `DWIScore` blank.
- Pathology module: `isupGrade in (None, "", "unknown")`.

See `f_20260305_vxannotate_integration/main.py` lines 133–138 and 153–158 for the implemented skip logic.

### vxAnnotate value-equal duplicate scoring rows

Independent of placeholders, vxAnnotate sometimes emits two scoring rows with **identical** `scoring_data` for the same `(submission, lesion, module)` — edit/resubmit artifacts.

Collapse them with `.unique(maintain_order=True)` on the raw score DataFrame **before** any strict uniqueness assertion. The assertion should only fire on genuinely-different scores (which is the real ambiguity).

## High-signal reference files

- `apps/vxdata-jobs/src/vxdata/jobs/source_2026_01_28_ingest_files/scrape.py` — tabular-driven file ingestion.
- `apps/vxdata-jobs/src/vxdata/jobs/f_20260305_vxannotate_integration/main.py` — vxAnnotate mask export with all the above gotchas handled.
- `apps/vxdata-jobs/src/vxdata/jobs/f_20260217_pathology/basel.py` — pathology table ingestion.
- `apps/vxdata-jobs/src/vxdata/jobs/f_20260302_lesion_masks/prostatex.py` — lesion mask derivation.
- `apps/vxdata-jobs/src/vxdata/jobs/utils/store_resources.py` — batch resource upload helpers.
- `apps/vxdata-jobs/src/vxdata/jobs/utils/ingest_dicoms.py` — DICOM header parsing.

---

**Historical note (2026-05-12)**: A spike evaluated a compact per-job `argo.yaml` contract + renderer against writing full Argo `WorkflowTemplate` / `CronWorkflow` manifests directly. Tradeoff: compact specs stay small but introduce a mini-platform that must stay correct; direct manifests are more explicit but repeat boilerplate. The spike artifacts existed in the archived `data-platform` repo under `apps/worker/argo/` but were not carried forward into `mono`. Current practice is writing Argo manifests directly in `virdx/infra_k8s` under `current/argo-workflows/`.

Source: https://docs.virdx.dev/knowledge/wiki/workstreams/infrastructure/sops/vxdata-jobs-ingestion/index.mdx