Customising the ETL pipeline

ACMAD Upload Helper separates source extraction, transformation into canonical clinical records, and destination loading. An organisation can replace one stage for a single record type, or provide a complete alternative source such as a hospital database, without making that source imitate the Excel package.

This guide describes the supported Python extension points. The application does not load arbitrary class names from acmad.yaml and does not currently discover source providers as runtime plugins. A custom provider is trusted Python code, normally supplied in a separate institution-specific package and composed explicitly by that package’s command-line application.

Choose the appropriate extension point

Use the narrowest extension point that owns the required change.

Requirement	Extension point	Usually retained
Read a different format for one slice	Custom Extractor and, when its intermediate model differs, a custom Transformer	Canonical domain record, loader, repository, and API adapter
Read every slice from a hospital database	ClinicalSourceProvider containing one SliceBinding per slice	Package dependency order, progress reporting, result aggregation, loaders, authentication, and API adapters
Change source-to-ACMAD vocabulary mappings	Custom Transformer	Extraction and loading
Send canonical records to another destination	Custom Loader or an implementation of the relevant repository port	Extractors, transformers, and canonical records
Run a one-off or single-slice integration	Construct EtlPipeline directly	Only the components explicitly supplied by the caller

For a complete alternative input, prefer a source provider over a set of independent configuration overrides. An extractor and transformer agree on an intermediate type, so replacing only one of them often creates accidental coupling to VerticalForm or other workbook concepts.

Core contracts

The generic stage protocols are defined in acmad_uploader.ports.pipeline:

class Extractor[SourceT, ExtractedT](Protocol):
    def extract(self, source: SourceT) -> StageResult[ExtractedT]: ...

class Transformer[ExtractedT, CanonicalT](Protocol):
    def transform(self, extracted: ExtractedT) -> StageResult[CanonicalT]: ...

class Loader[CanonicalT, LoadedT](Protocol):
    def load(self, canonical: CanonicalT) -> StageResult[LoadedT]: ...

These are structural protocols. A custom class does not need to inherit from them; a correctly typed method with the matching signature is sufficient. Expected data-quality and integration failures should be returned as StageResult issues. Exceptions are reserved for unexpected programming or infrastructure failures at a boundary where no useful structured result can be produced.

A complete source implements ClinicalSourceProvider. Its bindings() method receives a SourceProviderContext and returns one SliceBinding for every package slice.

The required names are available as acmad_uploader.application.package_upload.PACKAGE_SLICE_NAMES. The package runner rejects missing, duplicate, and unrecognised names, then puts valid bindings back into the core dependency order. A provider cannot cause a dependent slice to upload before its prerequisite merely by changing the order in which it returns bindings.

Design an institution-specific package

Keep hospital schema details and database-driver dependencies outside the core uploader package. A typical project might use this layout:

hospital-acmad-adapter/
├── pyproject.toml
├── src/hospital_acmad/
│   ├── cli.py
│   ├── provider.py
│   ├── database.py
│   ├── patient.py
│   ├── diagnosis.py
│   └── ...
└── tests/
    ├── test_patient.py
    ├── test_provider.py
    └── test_excel_equivalence.py

The adapter package should depend on a compatible version of acmad-upload-helper and on its chosen database driver. It should own:

• read-only queries or calls to versioned export views;

• database row models;

• mappings from those rows to ACMAD canonical records;

• institution-specific configuration and secret retrieval; and

• its own executable entry point, deployment, and integration tests.

Prefer versioned, read-only views such as acmad_export_v1.patients over queries against internal operational tables. Views provide a stable contract when the source system changes and give database administrators a clear place to enforce cohort and de-identification policy.

Implement a database extractor

An extractor describes what the source contained. It should preserve raw values that the transformer needs, attach useful provenance, and avoid applying API payload rules.

The following simplified patient example uses an application-defined query port. The concrete database module can implement this port with psycopg, SQLAlchemy, ODBC, or the institution’s preferred driver without exposing that choice to the transformer.

from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

from acmad_uploader.application.results import StageResult
from acmad_uploader.domain.issues import IssueSeverity, PipelineIssue
from acmad_uploader.domain.provenance import SourceLocation


@dataclass(frozen=True, slots=True)
class HospitalPatientRow:
    site_patient_id: str | None
    site_abbreviation: str | None
    biological_sex: str | None
    ethnicity: str | None
    patient_hash: str | None
    source: SourceLocation


class PatientQuery(Protocol):
    def fetch_patients(self) -> tuple[HospitalPatientRow, ...]:
        """Read one consistent patient snapshot."""


class HospitalDatabaseError(Exception):
    """Expected failure while reading the hospital export schema."""


class PatientDatabaseExtractor:
    def extract(
        self,
        source: PatientQuery,
    ) -> StageResult[tuple[HospitalPatientRow, ...]]:
        try:
            rows = source.fetch_patients()
        except HospitalDatabaseError as exc:
            return StageResult.failure(
                PipelineIssue(
                    code="hospital_patient.query_failed",
                    severity=IssueSeverity.ERROR,
                    message=f"The patient export view could not be read: {exc}",
                    location=SourceLocation.database(
                        "acmad_export_v1.patients"
                    ),
                )
            )
        return StageResult.success(rows)

Construct each row’s location as specifically as is safe and useful:

location = SourceLocation.database(
    "acmad_export_v1.patients",
    record="source_patient_key=4815",
)

Field-level issues can derive a more specific location:

sex_location = SourceLocation.database(
    "acmad_export_v1.patients",
    record="source_patient_key=4815",
    field="biological_sex",
)

Do not include database passwords, connection strings, direct identifiers, or other sensitive operational details in provenance or issue messages.

Implement a source-specific transformer

The transformer maps the extracted row model into records from acmad_uploader.domain. It owns source vocabulary conversion, required field checks, numeric and date coercion, joins, and source-specific consistency rules.

For example, a patient transformer ultimately produces PatientRecord values:

from acmad_uploader.application.results import StageResult
from acmad_uploader.domain.issues import (
    IssueSeverity,
    PipelineIssue,
    has_errors,
)
from acmad_uploader.domain.patient import (
    BiologicalSex,
    Ethnicity,
    PatientRecord,
)
from acmad_uploader.domain.provenance import SourceLocation
from acmad_uploader.domain.site_override import SiteOverride


_SEX_VALUES = {
    "female": BiologicalSex.FEMALE,
    "male": BiologicalSex.MALE,
    "unknown": BiologicalSex.NOT_SPECIFIED,
}

_ETHNICITY_VALUES = {
    "": Ethnicity.NOT_RECORDED,
    "european": Ethnicity.EUROPEAN,
    "australian": Ethnicity.AUSTRALIAN,
    "maori": Ethnicity.MAORI,
}


class PatientDatabaseTransformer:
    def __init__(self, site_override: SiteOverride | None = None) -> None:
        self._site_override = site_override

    def transform(
        self,
        extracted: tuple[HospitalPatientRow, ...],
    ) -> StageResult[tuple[PatientRecord, ...]]:
        records: list[PatientRecord] = []
        issues: list[PipelineIssue] = []

        for row in extracted:
            sex_text = (row.biological_sex or "").strip().casefold()
            sex = _SEX_VALUES.get(sex_text)
            ethnicity_text = (row.ethnicity or "").strip().casefold()
            ethnicity = _ETHNICITY_VALUES.get(ethnicity_text)
            site = (
                self._site_override.acronym
                if self._site_override is not None
                else (row.site_abbreviation or "").strip()
            )
            if not row.site_patient_id:
                issues.append(
                    PipelineIssue(
                        code="hospital_patient.identifier_required",
                        severity=IssueSeverity.ERROR,
                        message="The patient identifier is required.",
                        location=row.source,
                    )
                )
            if not site:
                issues.append(
                    PipelineIssue(
                        code="hospital_patient.site_required",
                        severity=IssueSeverity.ERROR,
                        message="The site abbreviation is required.",
                        location=SourceLocation.database(
                            "acmad_export_v1.patients",
                            record=row.source.record,
                            field="site_abbreviation",
                        ),
                    )
                )
            if sex is None:
                issues.append(
                    PipelineIssue(
                        code="hospital_patient.sex_invalid",
                        severity=IssueSeverity.ERROR,
                        message=(
                            "The source biological-sex value is not recognised."
                        ),
                        location=SourceLocation.database(
                            "acmad_export_v1.patients",
                            record=row.source.record,
                            field="biological_sex",
                        ),
                    )
                )
            if ethnicity is None:
                issues.append(
                    PipelineIssue(
                        code="hospital_patient.ethnicity_invalid",
                        severity=IssueSeverity.ERROR,
                        message="The source ethnicity value is not recognised.",
                        location=SourceLocation.database(
                            "acmad_export_v1.patients",
                            record=row.source.record,
                            field="ethnicity",
                        ),
                    )
                )
            if (
                not row.site_patient_id
                or not site
                or sex is None
                or ethnicity is None
            ):
                continue

            records.append(
                PatientRecord(
                    site_patient_id=row.site_patient_id,
                    site_abbreviation=site,
                    biological_sex=sex,
                    ethnicity=ethnicity,
                    patient_hash=row.patient_hash,
                    source=row.source,
                )
            )

        issue_tuple = tuple(issues)
        if has_errors(issue_tuple):
            return StageResult(value=None, issues=issue_tuple)
        return StageResult.success(tuple(records), issues=issue_tuple)

This example is intentionally small. A production transformer must enforce the same canonical meaning as the built-in workbook transformer, including duplicate natural keys, required relationships, permitted vocabularies, and cross-record consistency. At present some of those rules live in the workbook-specific transformers, so use their tests and behaviour as the reference when implementing an equivalent database mapping.

Never silently replace an unknown source value with a convenient default. Return a structured error unless the data dictionary explicitly defines that default. This is particularly important for clinical classifications and controlled vocabularies.

Assemble the complete provider

Bind each source-specific extractor and transformer as one unit. The same query service may be used as the source for several slices, and a provider may pass the site override from its context to every transformer that requires it.

from acmad_uploader.application.package_upload import PACKAGE_SLICE_NAMES
from acmad_uploader.ports.source_provider import (
    AnySliceBinding,
    SliceBinding,
    SourceProviderContext,
)


class HospitalDatabaseProvider:
    def __init__(self, queries: HospitalQueries) -> None:
        self._queries = queries

    def bindings(
        self,
        context: SourceProviderContext,
    ) -> tuple[AnySliceBinding, ...]:
        bindings = (
            SliceBinding(
                name="patient",
                source=self._queries,
                extractor=PatientDatabaseExtractor(),
                transformer=PatientDatabaseTransformer(
                    site_override=context.site_override,
                ),
            ),
            self._diagnosis_binding(context),
            self._gait_assessment_binding(context),
            self._function_assessment_binding(context),
            self._mocap_data_binding(context),
            self._biomechanics_artifact_binding(context),
            self._physical_examination_binding(context),
            self._proms_binding(context),
            self._surgery_binding(context),
        )
        assert len(bindings) == len(PACKAGE_SLICE_NAMES)
        assert {binding.name for binding in bindings} == set(
            PACKAGE_SLICE_NAMES
        )
        return bindings

The omitted helper methods follow the same pattern as the patient binding. The assertion is optional because the runner validates the set independently, but it gives adapter developers immediate feedback during construction.

Every slice needs a binding, even when the source currently contains no rows for that record type. Decide that policy explicitly:

• return a successful empty batch when the source contract genuinely supports the slice and an empty result is valid;

• return a structured error when an expected export view or dataset is unavailable; and

• do not silently use an empty extractor merely to satisfy the provider contract when doing so could omit clinical records.

Validate and upload through the provider

Use the provider-specific entry points from acmad_uploader.application.package_upload:

from acmad_uploader.application.package_upload import (
    PackageRunOptions,
    run_provider_upload,
    run_provider_validation,
)
from acmad_uploader.domain.site_override import SiteOverride
from acmad_uploader.ports.source_provider import SourceProviderContext

provider = HospitalDatabaseProvider(queries)
context = SourceProviderContext(
    site_override=SiteOverride(acronym="QCH", source="hospital profile")
)

validation = run_provider_validation(provider, context=context)
if not validation.succeeded:
    for issue in validation.issues:
        print(issue.display())
    raise SystemExit(2)

upload = run_provider_upload(
    provider,
    PackageRunOptions(
        api_url="https://acmad.example/api/v1/",
        realm_url="https://id.acmad.example/realms/prod",
        client_id="hospital-acmad-adapter",
    ),
    context=context,
)

Validation does not authenticate or contact the ACMAD service. Upload validates the provider contract before authentication, primes authentication once, and then attaches the built-in ACMAD loader for each slice.

Always present PackageRunResult.issues to the operator and return a non-zero process status after any failed step. The progress callbacks accepted by both functions can update a command-line display or graphical interface.

Run one slice directly

During adapter development it is often useful to run a single slice without implementing the complete provider first:

from acmad_uploader.application.pipeline import EtlPipeline

pipeline = EtlPipeline(
    extractor=PatientDatabaseExtractor(),
    transformer=PatientDatabaseTransformer(),
)
result = pipeline.run(queries)

if result.succeeded:
    patients = result.transformation.value if result.transformation else ()
    print(f"Validated {len(patients or ())} patients")

To load that slice, supply the existing record-specific loader and repository adapter. Direct composition is also the appropriate approach when replacing a destination; the provider-wide upload helper intentionally always uses the built-in ACMAD destination loaders.

Customise destination loading

Record-specific loaders depend on repository protocols rather than directly on HTTP. Implement the relevant protocol when the loading policy remains the same but persistence moves elsewhere. For example, PatientLoader calls a PatientRepository:

from acmad_uploader.domain.patient import PatientRecord
from acmad_uploader.ports.patient import (
    PatientLoadAction,
    PatientRepositoryError,
    PatientUpsertResult,
)


class WarehousePatientRepository:
    def __init__(self, warehouse: ClinicalWarehouse) -> None:
        self._warehouse = warehouse

    def upsert(self, patient: PatientRecord) -> PatientUpsertResult:
        try:
            created = self._warehouse.upsert_patient(patient)
        except WarehouseError as exc:
            raise PatientRepositoryError(str(exc)) from exc

        return PatientUpsertResult(
            site_patient_id=patient.site_patient_id,
            patient_url=f"warehouse://patients/{patient.site_patient_id}",
            action=(
                PatientLoadAction.CREATED
                if created
                else PatientLoadAction.UPDATED
            ),
        )

Supply that repository to the existing loader and run the pipeline with load=True:

from acmad_uploader.application.patient import PatientLoader

pipeline = EtlPipeline(
    extractor=PatientDatabaseExtractor(),
    transformer=PatientDatabaseTransformer(),
    loader=PatientLoader(WarehousePatientRepository(warehouse)),
)
result = pipeline.run(queries, load=True)

Implement a custom Loader instead when the application-level loading policy itself differs—for example, when a destination requires one atomic bulk operation rather than independent upserts. Preserve structured issues and source provenance so an operator can still identify rejected records.

For a complete alternative destination, write a small institution-specific package orchestrator that attaches its own loaders to each provider binding. Do not modify run_provider_upload to switch destination behaviour based on source-provider type; source and destination choices should remain independent.

Transactions, repeatability, and incremental reads

A provider-wide run invokes slices sequentially. If several extractors open independent database connections, they may observe different source states. The institution-specific adapter should therefore define its consistency strategy. Common approaches include:

• materialising versioned export tables for one approved cohort;

• sharing a read-only repeatable-read snapshot where the driver supports it;

• recording a source-system cut-off timestamp and applying it to every query; or

• exporting an immutable batch identifier used by all slices.

Start with complete, repeatable batches. The destination loaders upsert by natural key, so retrying the same validated batch is safe. Introduce an updated_at watermark only after defining:

• the timezone and inclusive/exclusive boundary rules;

• late-arriving or corrected records;

• how relationships are included when only one side changes;

• how failed batches affect the saved watermark; and

• deletion or withdrawal semantics. An upsert does not imply deletion.

Biomechanics artefacts

The current biomechanics artefact domain record and REST adapter upload from a local pathlib.Path. A custom provider whose database stores BLOBs or object-store references must make each artefact available as a local file for the duration of the upload. Use a restricted temporary directory, stream rather than buffering large files in memory, verify the size and SHA-256 hash, and delete temporary files after the run.

If motion-capture files remain on a managed filesystem, the database provider may be hybrid: database extractors can supply clinical slices while the artefact binding resolves approved filesystem paths. Do not accept unvalidated paths from clinical table values.

Configuration and secrets

Keep destination options and source-database secrets separate:

• PackageRunOptions and optional AppConfig values configure the ACMAD API and identity provider;

• the provider package configures its own database host, export schema, cohort, and snapshot policy; and

• passwords, client secrets, and full connection strings come from an operating-system credential store, environment injection, or an institution-managed secret service.

Do not place database credentials in acmad.yaml. Do not support YAML values such as module:ClassName that import arbitrary Python objects. If a future executable needs to select among several installed providers, expose a small allow-listed registry of stable provider identifiers and validate a provider API version.

Use a database account with the minimum required read-only permissions. It should normally be restricted to the versioned export views and should not be able to update operational clinical tables.

Testing a custom provider

Test each boundary independently and then test the complete provider.

Extractor tests

• Read representative rows, NULL values, Unicode text, dates, and large batches.

• Verify query failures become structured issues without exposing credentials.

• Confirm read-only and snapshot behaviour against the supported database.

• Verify provenance names the correct relation and stable source key.

Transformer tests

• Cover every source vocabulary value and every invalid-value path.

• Test required fields, duplicate natural keys, joins, and site consistency.

• Assert the exact canonical domain records produced.

• Use Australian English spelling in issue text, comments, and docstrings.

Provider contract tests

• Call run_provider_validation and assert that all names in PACKAGE_SLICE_NAMES are reported in core dependency order.

• Verify missing, duplicate, and unrecognised bindings fail before authentication.

• Confirm the same source selection and site context reach every applicable binding.

Equivalence tests

Where the same de-identified example can be represented in Excel and the hospital export schema, validate both and compare their canonical records. These tests are the strongest protection against mappings drifting between source formats. Compare issues as well as successful records, especially for controlled vocabularies and natural-key joins.

Do not use production clinical data as a test fixture. Construct synthetic or formally approved de-identified cases that exercise the same schema rules.

Deployment considerations

The default acmad-upload command and desktop application use the built-in Excel/filesystem provider. A hospital adapter should initially publish its own command, for example hospital-acmad-upload, which constructs the database provider explicitly and calls the provider runners.

Dynamic Python packages cannot simply be added beside a compiled standalone desktop application. If a custom provider must be available through the GUI, produce an institution-specific build that includes the provider, database driver, native driver libraries, and approved configuration interface. Test that build on the exact operating-system and database-driver combination used by the institution.

Implementation checklist

Before deploying a custom ETL integration, confirm that:

• every package slice has an intentional binding;

• source views and adapter mappings are versioned;

• all canonical mappings and validation failures are tested;

• a consistent snapshot or cut-off policy covers the whole run;

• credentials are external to package configuration and logs;

• provenance is useful without revealing sensitive infrastructure details;

• uploads have been rehearsed against a non-production ACMAD environment;

• repeat runs produce the expected unchanged or skipped outcomes;

• incremental and deletion behaviour, if any, is documented; and

• operators receive structured issues and a non-zero status on failure.

See Architecture and Python use for the overall dependency direction and Python API reference for the complete Python API reference.