Application services

Results shared by ETL stages and orchestration.

class acmad_uploader.application.results.StageResult(value, issues=())

Bases: Generic

Value and issues produced by one ETL stage.

value: T | None

issues: tuple[PipelineIssue, ...]

property succeeded: bool: Return whether the stage produced a value without errors.

classmethod success(value, *, issues=())

Create a successful result, optionally containing warnings.

Return type:: StageResult[TypeVar(T)]

classmethod failure(*issues)

Create a failed result.

Return type:: StageResult[TypeVar(T)]

class acmad_uploader.application.results.PipelineResult(extraction, transformation=None, loading=None)

Bases: Generic

Results of the stages reached during one pipeline execution.

extraction: StageResult

transformation: StageResult | None

loading: StageResult | None

property issues: tuple[PipelineIssue, ...]: Return issues from all stages in execution order.

property succeeded: bool: Return whether every executed stage succeeded.

Pipeline orchestration independent of Excel and the ACMAD REST API.

class acmad_uploader.application.pipeline.EtlPipeline(extractor, transformer, loader=None)

Bases: Generic

Coordinate extraction, transformation and optional loading.

The pipeline stops after the first failed stage. Loading is opt-in so a dry run is the safe default for clinical data.

extractor: Extractor

transformer: Transformer

loader: Loader | None

run(source, *, load=False)

Run reachable stages and return all stage results.

Return type:: PipelineResult[TypeVar(ExtractedT), TypeVar(CanonicalT), TypeVar(LoadedT)]

Package-wide validation and upload orchestration.

class acmad_uploader.application.package_upload.PackageRunOptions(site=None, api_url=None, realm_url=None, client_id=None, persist_tokens=True, open_browser=True, verify_tls=True, login_url_callback=None)

Bases: object

Options for package-wide validation and upload.

site: str | None

api_url: str | None

realm_url: str | None

client_id: str | None

persist_tokens: bool

open_browser: bool

verify_tls: bool

login_url_callback: Callable[[str], None] | None

class acmad_uploader.application.package_upload.PackageStepResult(name, succeeded, issues=(), record_count=None, result=None)

Bases: object

Outcome for one package-wide ETL step.

name: str

succeeded: bool

issues: tuple[PipelineIssue, ...]

record_count: int | None

result: PipelineResult[Any, Any, Any] | None

class acmad_uploader.application.package_upload.PackageRunResult(steps)

Bases: object

Outcome for a package-wide validation or upload run.

steps: tuple[PackageStepResult, ...]

property issues: tuple[PipelineIssue, ...]: Return issues from all package steps in execution order.

property succeeded: bool: Return whether every package step succeeded.

acmad_uploader.application.package_upload.run_package_validation(path, options, *, on_step_started=None, on_step_finished=None)

Validate every known record type in dependency order.

Return type:: PackageRunResult

acmad_uploader.application.package_upload.run_package_upload(path, options, *, on_step_started=None, on_step_finished=None)

Validate and upload every known record type in dependency order.

Return type:: PackageRunResult

Application service for loading a batch of patients.

class acmad_uploader.application.patient.PatientLoadSummary(patients)

Bases: object

Successful patient writes from a load attempt.

patients: tuple[PatientUpsertResult, ...]

class acmad_uploader.application.patient.PatientLoader(repository)

Bases: object

Load patients independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical patient and collect structured failures.

Return type:: StageResult[PatientLoadSummary]

Application service for loading a batch of diagnoses.

class acmad_uploader.application.diagnosis.DiagnosisLoadSummary(diagnoses)

Bases: object

Successful diagnosis writes from a load attempt.

diagnoses: tuple[DiagnosisUpsertResult, ...]

class acmad_uploader.application.diagnosis.DiagnosisLoader(repository)

Bases: object

Load diagnoses independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical diagnosis and collect structured failures.

Return type:: StageResult[DiagnosisLoadSummary]

Application service for loading a batch of gait assessments.

class acmad_uploader.application.gait_assessment.GaitAssessmentLoadSummary(assessments)

Bases: object

Successful gait assessment writes from a load attempt.

assessments: tuple[GaitAssessmentUpsertResult, ...]

class acmad_uploader.application.gait_assessment.GaitAssessmentLoader(repository)

Bases: object

Load assessments independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical gait assessment and collect structured failures.

Return type:: StageResult[GaitAssessmentLoadSummary]

Application service for loading a batch of function assessments.

class acmad_uploader.application.function_assessment.FunctionAssessmentLoadSummary(assessments)

Bases: object

Successful function assessment writes from a load attempt.

assessments: tuple[FunctionAssessmentUpsertResult, ...]

class acmad_uploader.application.function_assessment.FunctionAssessmentLoader(repository)

Bases: object

Load assessments independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical function assessment, collecting failures.

Return type:: StageResult[FunctionAssessmentLoadSummary]

Application service for loading a batch of mocap sessions.

class acmad_uploader.application.mocap_data.MocapDataLoadSummary(sessions)

Bases: object

Successful mocap session writes from a load attempt.

sessions: tuple[MocapDataUpsertResult, ...]

class acmad_uploader.application.mocap_data.MocapDataLoader(repository)

Bases: object

Load sessions independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical mocap session, collecting structured failures.

Return type:: StageResult[MocapDataLoadSummary]

Application service for loading a batch of biomechanics artifacts.

class acmad_uploader.application.biomechanics_artifact.BiomechanicsArtifactLoadSummary(artifacts)

Bases: object

Successful biomechanics artifact writes from a load attempt.

artifacts: tuple[BiomechanicsArtifactUpsertResult, ...]

class acmad_uploader.application.biomechanics_artifact.BiomechanicsArtifactLoader(repository)

Bases: object

Load artifacts independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical artifact, collecting structured failures.

Return type:: StageResult[BiomechanicsArtifactLoadSummary]

Application service for loading a batch of physical examinations.

class acmad_uploader.application.physical_examination.PhysicalExaminationLoadSummary(examinations)

Bases: object

Successful physical examination writes from a load attempt.

examinations: tuple[PhysicalExaminationUpsertResult, ...]

class acmad_uploader.application.physical_examination.PhysicalExaminationLoader(repository)

Bases: object

Load examinations independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical physical examination, collecting failures.

Return type:: StageResult[PhysicalExaminationLoadSummary]

Application service for loading a batch of PROMs family responses.

class acmad_uploader.application.proms.PromsLoadSummary(proms)

Bases: object

Successful PROMs writes from a load attempt.

proms: tuple[PromsUpsertResult, ...]

class acmad_uploader.application.proms.PromsLoader(repository)

Bases: object

Load PROMs independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical PROMs record, collecting structured failures.

Return type:: StageResult[PromsLoadSummary]

Application service for loading a batch of surgery episodes.

class acmad_uploader.application.surgery.SurgeryLoadSummary(surgeries)

Bases: object

Successful surgery episode writes from a load attempt.

surgeries: tuple[SurgeryUpsertResult, ...]

class acmad_uploader.application.surgery.SurgeryLoader(repository)

Bases: object

Load surgeries independently so one API rejection does not hide later ones.

load(canonical)

Upsert every canonical surgery episode, collecting structured failures.

Return type:: StageResult[SurgeryLoadSummary]