Bulk Submission Guide

The Bulk Record Submission Workflow manages the complexity of uploading large volumes of data by orchestrating a two-phase ingestion process. It ensures that required subject registrations complete before any subsequent data is uploaded.

Two-Phase Process

When submitting records in bulk, the operations are automatically separated into two phases:

1. Registration Phase: All REGISTER_SUBJECT test types are batched and submitted. The orchestrator polls until these registrations complete successfully.

2. Data Phase: All other record types (e.g., CREATE_NEW_RECORD, UPDATE_SCHEDULED_RECORD) are subsequently submitted, preventing missing subject errors.

Visual Workflow

The following diagram illustrates the internal two-phase orchestration:

sequenceDiagram participant User participant Orchestrator as BulkRecordSubmissionWorkflow participant API as iMednet API participant Poller as PollingManager User->>Orchestrator: submit(Study, RecordSets) Note over Orchestrator,API: Phase 1: Registration Orchestrator->>API: create_record(Registration Batches) API-->>Orchestrator: Job IDs (Registration) Orchestrator->>Poller: poll(Registration Job IDs) Poller-->>Orchestrator: Completed Note over Orchestrator,API: Phase 2: Data Submission Orchestrator->>API: create_record(Data Batches) API-->>Orchestrator: Job IDs (Data) Orchestrator-->>User: SubmissionResult

Usage Example

The following script sets up a bulk submission instance and illustrates the two-phase workflow:

from imednet import ImednetSDK, load_config
from imednet_workflows.uat import BulkRecordSubmissionWorkflow, GeneratedRecordSet, RecordTestType


def main():
    try:
        cfg = load_config()
    except ValueError as e:
        print(f"Skipping execution: {e}")
        return

    # In a real environment, you'd execute against an active study.
    try:
        with ImednetSDK(
            api_key=cfg.api_key,
            security_key=cfg.security_key,
            base_url=cfg.base_url,
        ) as sdk:
            BulkRecordSubmissionWorkflow(
                sdk, batch_size=500, skip_existing_subjects=True, await_registration=True
            )

            # Example record sets (normally produced by a generator or mapped from source data)
            GeneratedRecordSet(
                form_key="REG",
                form_name="Registration Form",
                test_type=RecordTestType.REGISTER_SUBJECT,
                payloads=[{"subject_id": "SUBJ_1", "age": 45}],
                subject_keys=["SUBJ_1"],
            )

            GeneratedRecordSet(
                form_key="VITALS",
                form_name="Vitals Form",
                test_type=RecordTestType.CREATE_NEW_RECORD,
                payloads=[{"subject_id": "SUBJ_1", "hr": 72, "bp": "120/80"}],
                subject_keys=["SUBJ_1"],
            )

            # Submit in two phases: Registration first, then Data
            # result = workflow.submit("MY_STUDY", [registration_set, data_set])
            # print(f"Submitted {result.total_records_submitted} records across {len(result.all_batches)} batches.")
    except Exception as e:
        print(f"Failed: {e}")


if __name__ == "__main__":
    main()

Error Handling

If the registration phase fails (i.e. one or more jobs return a FAILED state), the orchestrator will raise a BulkSubmissionError immediately, avoiding cascading failures on the dependent data phase.