Guides
Building an ETL
Extract, transform, and load data between lab instruments and external systems using UniteLabs.
This guide walks through building a production-ready ETL pipeline that monitors a LabChip GXII Touch instrument directory, archives raw files to Object Storage, and writes structured records to the Data Warehouse. The same pattern applies to any File System Connector and any instrument type.
Prerequisites
- A running File System Connector (e.g.,
"File System (LabChip demo)"). datalake-adminandwarehouse-adminsecrets configured in UniteLabs. See Secrets.- Python packages:
prefect>=3.0,unitelabs-sdk,sqlmodel,boto3,pandas,asyncpg.
Architecture
The pipeline uses three sequential stages inside a single Prefect @flow:
- Reconcile — on startup, compare the connector's current file list against Warehouse records to detect files that were written while the pipeline was offline.
- Process backlog — archive and register every reconciled file that has not yet been stored.
- Watch in real time — subscribe to change events and process new files as they arrive.
This three-stage pattern ensures no files are skipped even if the workflow was offline when the instrument wrote results.
Define the data model
Define a SourceFiles model to track every file the pipeline has seen. Keep it in a shared models.py so both the flow and any query scripts can import it:
models.py
from datetime import datetime
from typing import Any
import sqlalchemy as sa
from sqlmodel import Field, SQLModel
class SourceFiles(SQLModel, table=True):
__tablename__ = "source_files"
id: int | None = Field(default=None, primary_key=True)
file_path: str # Original path on the connector
s3_object_path: str # Path inside the S3 bucket
connector_name: str
instrument_type: str
file_metadata: dict[str, Any] | None = Field(
default=None, sa_type=sa.JSON
)
updated_at: datetime
Task: register a source file
Write a single file record to the Warehouse. Using upsert_record with file_path as the conflict key makes the task safe to retry:
from datetime import datetime, timezone
from prefect import task
@task(name="Register Source File")
async def register_source_file_task(
file_path: str,
s3_object_path: str,
connector_name: str,
instrument_type: str,
warehouse,
) -> SourceFiles:
record = SourceFiles(
file_path=file_path,
s3_object_path=s3_object_path,
connector_name=connector_name,
instrument_type=instrument_type,
updated_at=datetime.now(tz=timezone.utc),
)
await warehouse.upsert_record(record)
return record
Task: archive to Object Storage
Upload the raw file from the connector directly to S3. The transfer happens on the connector side — no data passes through the workflow host:
from unitelabs.sdk import AsyncApiClient
@task(name="Archive to Object Storage")
async def archive_file_task(
connector_name: str,
file_path: str,
s3_url: str,
) -> str:
async with AsyncApiClient() as client:
device = await client.get_service_by_name(connector_name)
await device.s3_service.upload_file(path=file_path, s3_url=s3_url)
return s3_url
Task: reconcile missed files
On startup, list all files currently visible to the connector and cross-reference against Warehouse records. Returns only files not yet registered:
@task(name="Reconcile Connector Files")
async def reconcile_connector_files_task(
connector_name: str,
instrument_type: str,
warehouse,
) -> dict:
async with AsyncApiClient() as client:
device = await client.get_service_by_name(connector_name)
whitelist = await device.folder_service.get_folder_whitelist()
all_files: list[str] = []
for folder in whitelist:
files = await device.folder_service.get_folder_contents(path=folder)
all_files.extend(files)
existing = await warehouse.query_records(
SourceFiles,
instrument_type=instrument_type,
)
existing_paths = {r.file_path for r in existing}
reconciled = [f for f in all_files if f not in existing_paths]
return {
"reconciled_files": reconciled,
"total_on_disk": len(all_files),
}
Task: watch for new files
A long-running task that yields each new file event as it arrives from the connector:
@task(name="Watch for New Files")
async def watch_for_new_files_task(connector_name: str):
async with AsyncApiClient() as client:
device = await client.get_service_by_name(connector_name)
async with await device.folder_service.subscribe_changes() as sub:
async for file_path in sub:
yield {"file_path": file_path}
The flow: wiring it together
The @flow function runs all three stages in order. Stages 2 and 3 call the same archive + register tasks so the logic is never duplicated:
from unitelabs.sdk import AsyncApiClient
from prefect import flow, get_run_logger
@flow(log_prints=True, name="LabChip GXII Touch ETL Pipeline")
async def labchip_etl_flow(
connector_name: str = "File System (Labchip demo)",
instrument_type: str = "labchip",
s3_bucket_prefix: str = "data/instruments/labchip",
) -> None:
logger = get_run_logger()
# Load S3 credentials from secrets
async with AsyncApiClient() as client:
secrets = await client.get("/secrets")
datalake = next(s for s in secrets if s["name"] == "datalake-admin")
p = datalake["parameters"]
s3_base = (
f"s3s://{p['minio_root_user']}:{p['minio_root_password']}"
f"@{p['aws_client_parameters']['endpoint_url'].replace('http://', '')}"
f"/{p['aws_client_parameters']['bucket_name']}/{s3_bucket_prefix}"
)
warehouse = get_warehouse_client()
await warehouse.ensure_tables_exist()
# ── Stage 1: Reconcile files missed while offline ──────────────────────
logger.info("Stage 1: Reconciling missed files")
reconciliation = await reconcile_connector_files_task(
connector_name=connector_name,
instrument_type=instrument_type,
warehouse=warehouse,
)
logger.info(
f"Found {len(reconciliation['reconciled_files'])} unregistered files "
f"out of {reconciliation['total_on_disk']} on disk"
)
# ── Stage 2: Process the backlog ───────────────────────────────────────
for file_path in reconciliation["reconciled_files"]:
filename = file_path.split("/")[-1]
s3_url = f"{s3_base}/{filename}"
await archive_file_task(
connector_name=connector_name,
file_path=file_path,
s3_url=s3_url,
)
await register_source_file_task(
file_path=file_path,
s3_object_path=f"{s3_bucket_prefix}/{filename}",
connector_name=connector_name,
instrument_type=instrument_type,
warehouse=warehouse,
)
# ── Stage 3: Real-time monitoring ──────────────────────────────────────
logger.info("Stage 3: Watching for new files in real time")
async for change in watch_for_new_files_task(connector_name):
file_path = change["file_path"]
filename = file_path.split("/")[-1]
s3_url = f"{s3_base}/{filename}"
logger.info(f"New file detected: {file_path}")
await archive_file_task(
connector_name=connector_name,
file_path=file_path,
s3_url=s3_url,
)
await register_source_file_task(
file_path=file_path,
s3_object_path=f"{s3_bucket_prefix}/{filename}",
connector_name=connector_name,
instrument_type=instrument_type,
warehouse=warehouse,
)
Run the flow
uv run python -c "
import asyncio
from my_etl.flows.labchip import labchip_etl_flow
asyncio.run(labchip_etl_flow())
"
Deploy with Prefect
Deploy the flow as a Prefect deployment so it restarts automatically on failure and appears in the Prefect UI for monitoring. See the Deploy a workflow guide for the manifest format and deployment commands.
Verify in the Warehouse
After running the flow, query the source_files table to confirm records were written:
SELECT file_path, instrument_type, updated_at
FROM warehouse.source_files
WHERE instrument_type = 'labchip'
ORDER BY updated_at DESC
LIMIT 20;
See Data Sources for instructions on connecting DBeaver to the warehouse so you can browse tables without writing SQL.
Next steps
- Object Storage — understand how S3 URLs are constructed and how to browse raw files
- File System Connector — full reference for all SDK connector operations
- Data Sources — query the Warehouse from DBeaver or any PostgreSQL client