.. _customising-etl:

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.

.. list-table::
   :header-rows: 1
   :widths: 30 30 40

   * - 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
:mod:`acmad_uploader.ports.pipeline`:

.. code-block:: python

   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
:class:`~acmad_uploader.application.results.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
:class:`~acmad_uploader.ports.source_provider.ClinicalSourceProvider`. Its
``bindings()`` method receives a
:class:`~acmad_uploader.ports.source_provider.SourceProviderContext` and
returns one :class:`~acmad_uploader.ports.source_provider.SliceBinding` for
every package slice.

The required names are available as
:data:`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:

.. code-block:: text

   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.

.. code-block:: python

   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:

.. code-block:: python

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

Field-level issues can derive a more specific location:

.. code-block:: python

   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
:mod:`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
:class:`~acmad_uploader.domain.patient.PatientRecord` values:

.. code-block:: python

   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.

.. code-block:: python

   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
:mod:`acmad_uploader.application.package_upload`:

.. code-block:: python

   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:

.. code-block:: python

   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,
:class:`~acmad_uploader.application.patient.PatientLoader` calls a
:class:`~acmad_uploader.ports.patient.PatientRepository`:

.. code-block:: python

   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``:

.. code-block:: python

   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 :class:`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 :doc:`architecture` for the overall dependency direction and
:doc:`api/index` for the complete Python API reference.