Building a Biomedical Data Lake with FastAPI, MinIO, and PostgreSQL

raw/        → Original collected data (immutable, 90-day object-lock)
processed/  → Filtered, normalized, or aligned data
curated/    → Curated and annotated data ready for consumption
archive/    → Historical snapshots for audit (180-day lifecycle)

# app/models/dataset.py
class Dataset(Base):
    __tablename__ = "datasets"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str]
    source: Mapped[SourceType]        # geo, ncbi_gene, pubmed, uniprot, upload
    external_id: Mapped[str]          # GSE2034, P04637, 32581362...
    layer: Mapped[DatasetLayer]       # raw, processed, curated, archive
    minio_bucket: Mapped[str]
    minio_prefix: Mapped[str]
    file_count: Mapped[int]
    total_size: Mapped[int | None]
    metadata_: Mapped[dict] = mapped_column(JSONB)  # {"source": "geo", "date": "2026-06-17", "version": "v1"}
    tags: Mapped[list[str]] = mapped_column(ARRAY(Text))
    created_at: Mapped[datetime]
    updated_at: Mapped[datetime]

# app/schemas/catalog.py
class DatasetCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=256)
    source: SourceType
    external_id: str = Field(..., min_length=1, max_length=128)
    layer: DatasetLayer = DatasetLayer.raw
    minio_bucket: str
    minio_prefix: str
    metadata_: dict[str, object] | None = None
    tags: list[str] | None = None

    @model_validator(mode="after")
    def _ensure_metadata(self) -> DatasetCreate:
        if self.metadata_ is None:
            self.metadata_ = {}
        if "source" not in self.metadata_:
            self.metadata_["source"] = str(self.source)
        if "date" not in self.metadata_:
            self.metadata_["date"] = datetime.now(UTC).date().isoformat()
        if "version" not in self.metadata_:
            self.metadata_["version"] = "v1"
        return self

# app/services/layer_service.py
_LAYER_ORDER = [
    DatasetLayer.raw,
    DatasetLayer.processed,
    DatasetLayer.curated,
    DatasetLayer.archive,
]

async def promote_dataset(session: AsyncSession, data: PromoteRequest) -> dict[str, object]:
    dataset = await session.get(Dataset, data.dataset_id)
    if not dataset:
        raise NotFoundError("Dataset not found")

    current_idx = _LAYER_ORDER.index(dataset.layer)
    target_idx = _LAYER_ORDER.index(data.target_layer)

    if target_idx != current_idx + 1:
        raise ServiceError(
            f"Cannot promote from {dataset.layer} to {data.target_layer}. "
            f"Only adjacent promotion is allowed."
        )

    # Copy objects in MinIO (never move)
    mc = MinioClient()
    objects = await mc.list_objects(dataset.minio_bucket, prefix=dataset.minio_prefix)
    for obj in objects:
        await mc.copy_object(
            source_bucket=dataset.minio_bucket,
            source_object=obj["object_name"],
            dest_bucket=data.target_layer,
            dest_object=f"{dataset.minio_prefix}/{obj['object_name'].split('/')[-1]}",
        )

    old_layer = dataset.layer
    dataset.layer = data.target_layer
    dataset.minio_bucket = data.target_layer

    provenance = Provenance(
        dataset_id=dataset.id,
        action=ProvenanceAction.promotion,
        layer_from=old_layer,
        layer_to=data.target_layer,
    )
    session.add(provenance)
    await session.commit()

    return {"status": "completed", "source_layer": old_layer, "target_layer": data.target_layer}

# app/models/provenance.py
class Provenance(Base):
    __tablename__ = "provenance"

    id: Mapped[uuid.UUID]
    dataset_id: Mapped[uuid.UUID]          # FK → datasets
    source_dataset_id: Mapped[uuid.UUID | None]  # FK → datasets (optional)
    action: Mapped[ProvenanceAction]       # collection, promotion, pipeline, manual
    layer_from: Mapped[DatasetLayer | None]
    layer_to: Mapped[DatasetLayer | None]
    parameters: Mapped[dict | None] = mapped_column(JSONB)
    created_at: Mapped[datetime]

curl -X POST http://localhost:8002/catalog \
  -H "Content-Type: application/json" \
  -d '{
    "name": "GSE2034 - Breast Cancer Metastasis",
    "source": "geo",
    "external_id": "GSE2034",
    "layer": "raw",
    "minio_bucket": "raw",
    "minio_prefix": "geo/GSE2034"
  }'

curl -X POST http://localhost:8002/layers/promote \
  -H "Content-Type: application/json" \
  -d '{
    "dataset_id": "<uuid>",
    "target_layer": "processed",
    "notes": "After FastQC + MultiQC"
  }'

curl http://localhost:8002/provenance/<uuid>/graph

curl -X POST http://localhost:8002/backup/trigger

# app/tasks/backup_scheduler.py
async def run(self) -> None:
    while True:
        now = datetime.now(UTC)
        if await self._needs_backup(now):
            await self._run_backup()
        await asyncio.sleep(3600)

# Unit tests
uv run pytest tests/ -v

# Integration
uv run pytest tests/test_integration/ -v

02-data-lake/
├── app/           → FastAPI backend
├── frontend/      → TypeScript SPA
├── tests/         → 29 tests
└── docker-compose.yml