imednet_workflows.uat package

UAT specification models.

class imednet_workflows.uat.BatchSubmission[source]

Bases: ImednetBaseModel

Tracks a single API call to records.create().

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class imednet_workflows.uat.BulkRecordSubmissionWorkflow[source]

Bases: object

Two-phase bulk record submission for UAT scenarios.

Phase 1: Submit all RegisterSubjectRequest payloads and await job completion. Phase 2: Submit all UpdateScheduledRecordRequest and CreateNewRecordRequest payloads.

Parameters

sdkImednetFacade

Initialized synchronous SDK instance.

batch_sizeint

Maximum number of records per API call. Default: 100.

await_registrationbool

If True, block until all Phase 1 jobs reach terminal state before submitting Phase 2 records. Default: True.

registration_timeoutfloat

Seconds to wait for Phase 1 jobs. Default: 600.

skip_existing_subjectsbool

If True, query the API for existing UAT-tagged subjects and skip registration for any that already exist. Default: True.

__init__(sdk, batch_size=100, await_registration=True, registration_timeout=600.0, skip_existing_subjects=True)[source]

Initialize the workflow.

Parameters:

  • sdk (ImednetFacade) –

  • batch_size (int) –

  • await_registration (bool) –

  • registration_timeout (float) –

  • skip_existing_subjects (bool) –

Return type:

None

submit(study_key, record_sets, *, email_notify=None, keyword_tag='UAT')[source]

Execute the two-phase submission.

Returns a SubmissionResult with all jobs and metadata. Raises BulkSubmissionError if Phase 1 jobs fail and Phase 2 depends on them.

Return type:

SubmissionResult

Parameters:
exception imednet_workflows.uat.BulkSubmissionError[source]

Bases: Exception

Raised when Phase 1 registration jobs fail, blocking Phase 2 submission.

__init__(message, failed_batches)[source]

Initialize the error.

Parameters:

  • message (str) – The error message.

  • failed_batches (list[BatchSubmission]) – List of batches that failed during submission.

Return type:

None

class imednet_workflows.uat.EditCheckResultStatus[source]

Bases: str, Enum

Possible outcomes for an edit check verification rule.

class imednet_workflows.uat.EditCheckVerificationReport[source]

Bases: object

Consolidated report for an automated edit check verification run.

__init__(study_key, total_rules=0, passed_rules=0, failed_rules=0, skipped_rules=0, results=<factory>)
Parameters:

  • study_key (str) –

  • total_rules (int) –

  • passed_rules (int) –

  • failed_rules (int) –

  • skipped_rules (int) –

  • results (list[dict[str, Any]]) –

Return type:

None

class imednet_workflows.uat.GeneratedRecordSet[source]

Bases: ImednetBaseModel

Output of the generator for a single form spec.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class imednet_workflows.uat.RecordTestType[source]

Bases: str, Enum

What kind of record submission is being tested.

class imednet_workflows.uat.StudySchemaInspector[source]

Bases: object

Fetch and index all structural metadata for an iMednet study.

Supports both sync and async SDK clients. When an async client is used, metadata calls are executed concurrently via asyncio.gather. The in-memory cache is per inspector instance. It can be shared by concurrent tasks in a single event loop, but is not thread-safe across threads/processes; use external synchronization or separate inspector instances for that access pattern.

__init__(sdk)[source]

Initialize the study schema inspector.

Parameters:

sdk (ImednetFacade | AsyncImednetFacade) – An instance of the synchronous or asynchronous iMednet SDK facade.

Return type:

None

async async_inspect(study_key, *, force_refresh=False)[source]

Asynchronously fetch metadata concurrently and return a study snapshot.

Return type:

StudySnapshot

Parameters:

  • study_key (str) –

  • force_refresh (bool) –

clear_cache(study_key=None)[source]

Clear snapshot cache for one study key or all keys.

Return type:

None

Parameters:

study_key (str | None) –

inspect(study_key, *, force_refresh=False)[source]

Synchronously fetch and return a study snapshot.

Return type:

StudySnapshot

Parameters:

  • study_key (str) –

  • force_refresh (bool) –

class imednet_workflows.uat.StudySnapshot[source]

Bases: ImednetBaseModel

Point-in-time snapshot of a study’s structural metadata.

active_sites()[source]

Return sites that are actively enrolling.

Return type:

list[Site]

enrollment_forms()[source]

Return forms with form_type indicating subject registration.

Return type:

list[Form]

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_StudySnapshot__context)[source]

Initialize derived lookup dictionaries after model initialization.

Return type:

None

Parameters:

_StudySnapshot__context (object) –

scheduled_forms()[source]

Return scheduled (non-enrollment, non-unscheduled) forms.

Return type:

list[Form]

unscheduled_forms()[source]

Return unscheduled forms.

Return type:

list[Form]

class imednet_workflows.uat.SubmissionResult[source]

Bases: ImednetBaseModel

Aggregate result of a full UAT submission run.

property all_batch_ids: list[str]

Return all batch IDs from all submitted jobs.

property all_batches: list[imednet_workflows.uat.submission.BatchSubmission]

Return all submission batches from both phases.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property total_records_submitted: int

Return the total count of records submitted across all batches.

class imednet_workflows.uat.SyntheticRecordGenerator[source]

Bases: object

Generate synthetic record payloads from a UATSpecification and StudySnapshot.

Parameters

seedOptional[int]

Random seed for reproducible generation. None means non-deterministic.

localestr

Locale (no longer used, kept for backwards compatibility).

__init__(seed=None, locale='en_US')[source]

Initialize the generator.

Parameters:

  • seed (Optional[int]) – Optional seed for reproducibility.

  • locale (str) – Ignored.

Return type:

None

generate(spec, snapshot)[source]

Generate all record payloads for the enabled forms in the spec.

Returns one GeneratedRecordSet per enabled UATFormSpec, in the correct submission order: RegisterSubject records first, then scheduled/unscheduled.

Return type:

list[GeneratedRecordSet]

Parameters:
class imednet_workflows.uat.UATExecutionEngine[source]

Bases: object

Automated Edit Check Verification execution engine.

__init__(sdk, data_dictionary)[source]

Initialize the UAT execution engine.

Parameters:

  • sdk (ImednetFacade) – An instance of the iMednet SDK facade.

  • data_dictionary (DataDictionary) – The data dictionary containing business rules.

generate_negative_test_case(rule)[source]

Generate a data payload designed to violate the given rule.

Return type:

dict[str, Any]

Parameters:

rule (dict[str, Any]) –

run_verification(study_key, form_scope=None, site_name='TestSite')[source]

Run the end-to-end UAT verification against the EDC API.

Return type:

EditCheckVerificationReport

Parameters:

  • study_key (str) –

  • form_scope (str | None) –

  • site_name (str) –

class imednet_workflows.uat.UATFormSpec[source]

Bases: UATBaseModel

Form-level UAT test specification.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_subject_count(value)[source]

Enforce safe subject count bounds.

Return type:

int

Parameters:

value (int) –

classmethod validate_unique_variable_names(variables)[source]

Ensure variables are uniquely identified by variable_name.

Return type:

list[UATVariableSpec]

Parameters:

variables (list[imednet_workflows.uat.models.UATVariableSpec]) –

class imednet_workflows.uat.UATRunPhase[source]

Bases: str, Enum

Phases of a UAT workflow execution.

class imednet_workflows.uat.UATRunResult[source]

Bases: ImednetBaseModel

Complete result of a UATWorkflow.run() execution.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property overall_success: bool

Return True if all jobs in the run were successful.

summary()[source]

Return a human-readable summary string suitable for CLI output.

Return type:

str

class imednet_workflows.uat.UATSpecification[source]

Bases: UATBaseModel

Root model for a complete UAT test plan.

enabled_forms()[source]

Return only enabled forms.

Return type:

list[UATFormSpec]

forms_by_type(test_type)[source]

Return enabled forms matching a specific test type.

Return type:

list[UATFormSpec]

Parameters:

test_type (RecordTestType) –

classmethod from_json(payload)[source]

Load a UAT specification from JSON text, bytes, or file path.

Return type:

UATSpecification

Parameters:

payload (str | bytes | Path) –

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_spec_version(value)[source]

Pin this release to v1.0 schema.

Return type:

str

Parameters:

value (str) –

class imednet_workflows.uat.UATSpecificationBuilder[source]

Bases: object

Auto-generate a UATSpecification from a StudySnapshot.

The generated spec uses SYNTHETIC strategy for all variables. Users can then edit the JSON output to apply FIXED or SKIP strategies for specific variables before feeding it back into UATWorkflow.run().

build(snapshot, *, subject_count=2, site_name=None, enabled_form_keys=None, seed=None)[source]

Build a complete UATSpecification from a StudySnapshot.

Return type:

UATSpecification

Parameters:

  • snapshot (StudySnapshot) –

  • subject_count (int) –

  • site_name (str | None) –

  • enabled_form_keys (list[str] | None) –

  • seed (int | None) –

Parameters

snapshotStudySnapshot

The study metadata snapshot from StudySchemaInspector.

subject_countint

Number of test subjects to register. Default 2.

site_nameOptional[str]

Site to register subjects at. If None, uses the first active site.

enabled_form_keysOptional[List[str]]

Restrict to a subset of forms. If None, all non-disabled forms are included.

seedOptional[int]

Passed through to UATSpecification for traceability.

class imednet_workflows.uat.UATSubjectSpec[source]

Bases: UATBaseModel

Defines synthetic test subjects to register before form submission.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_subject_count(value)[source]

Enforce safe subject count bounds.

Return type:

int

Parameters:

value (int) –

class imednet_workflows.uat.UATVariableSpec[source]

Bases: UATBaseModel

Variable-level test specification.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_fixed_strategy(data)[source]

Require a fixed value when FIXED strategy is selected.

Return type:

Any

Parameters:

data (Any) –

class imednet_workflows.uat.UATWorkflow[source]

Bases: object

End-to-end UAT execution coordinator.

Composes StudySchemaInspector, UATSpecificationBuilder, SyntheticRecordGenerator, BulkRecordSubmissionWorkflow, and JobPoller into a single pipeline.

__init__(sdk, *, batch_size=100, poll_interval=10.0, poll_timeout=600.0, seed=None, on_progress=None)[source]

Initialize the UAT workflow orchestrator.

Parameters:

  • sdk (ImednetFacade) –

  • batch_size (int) –

  • poll_interval (float) –

  • poll_timeout (float) –

  • seed (int | None) –

  • on_progress (JobProgressCallback | None) –

Return type:

None

build_spec(snapshot, **kwargs)[source]

Auto-generate a UATSpecification from the StudySnapshot.

Return type:

UATSpecification

Parameters:

  • snapshot (Any) –

  • kwargs (Any) –

generate(spec, snapshot)[source]

Create synthetic record payloads via SyntheticRecordGenerator.

Return type:

list[GeneratedRecordSet]

Parameters:
inspect(study_key)[source]

Fetch study metadata via StudySchemaInspector.

Return type:

Any

Parameters:

study_key (str) –

monitor(study_key, submission_result)[source]

Poll all resulting jobs to completion via JobPoller.

Return type:

Any

Parameters:
run(study_key, *, spec=None, subject_count=2, site_name=None, enabled_form_keys=None, email_notify=None)[source]

Execute the full UAT pipeline and return a UATRunResult.

Return type:

UATRunResult

Parameters:

  • study_key (str) –

  • spec (UATSpecification | None) –

  • subject_count (int) –

  • site_name (str | None) –

  • enabled_form_keys (list[str] | None) –

  • email_notify (bool | str | None) –

submit(study_key, record_sets, email_notify=None)[source]

Execute the two-phase bulk submission.

Return type:

SubmissionResult

Parameters:
class imednet_workflows.uat.VariableTestStrategy[source]

Bases: str, Enum

How a variable’s test value should be determined.

Submodules

imednet_workflows.uat.engine module

Engine for automated UAT (User Acceptance Testing) and edit check verification.

class imednet_workflows.uat.engine.EditCheckResultStatus[source]

Bases: str, Enum

Possible outcomes for an edit check verification rule.

class imednet_workflows.uat.engine.EditCheckVerificationReport[source]

Bases: object

Consolidated report for an automated edit check verification run.

__init__(study_key, total_rules=0, passed_rules=0, failed_rules=0, skipped_rules=0, results=<factory>)
Parameters:

  • study_key (str) –

  • total_rules (int) –

  • passed_rules (int) –

  • failed_rules (int) –

  • skipped_rules (int) –

  • results (list[dict[str, Any]]) –

Return type:

None

class imednet_workflows.uat.engine.UATExecutionEngine[source]

Bases: object

Automated Edit Check Verification execution engine.

__init__(sdk, data_dictionary)[source]

Initialize the UAT execution engine.

Parameters:

  • sdk (ImednetFacade) – An instance of the iMednet SDK facade.

  • data_dictionary (DataDictionary) – The data dictionary containing business rules.

generate_negative_test_case(rule)[source]

Generate a data payload designed to violate the given rule.

Return type:

dict[str, Any]

Parameters:

rule (dict[str, Any]) –

run_verification(study_key, form_scope=None, site_name='TestSite')[source]

Run the end-to-end UAT verification against the EDC API.

Return type:

EditCheckVerificationReport

Parameters:

  • study_key (str) –

  • form_scope (str | None) –

  • site_name (str) –

imednet_workflows.uat.generator module

Synthetic record payload generator for UAT.

class imednet_workflows.uat.generator.GeneratedRecordSet[source]

Bases: ImednetBaseModel

Output of the generator for a single form spec.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class imednet_workflows.uat.generator.SyntheticRecordGenerator[source]

Bases: object

Generate synthetic record payloads from a UATSpecification and StudySnapshot.

Parameters

seedOptional[int]

Random seed for reproducible generation. None means non-deterministic.

localestr

Locale (no longer used, kept for backwards compatibility).

__init__(seed=None, locale='en_US')[source]

Initialize the generator.

Parameters:

  • seed (Optional[int]) – Optional seed for reproducibility.

  • locale (str) – Ignored.

Return type:

None

generate(spec, snapshot)[source]

Generate all record payloads for the enabled forms in the spec.

Returns one GeneratedRecordSet per enabled UATFormSpec, in the correct submission order: RegisterSubject records first, then scheduled/unscheduled.

Return type:

list[GeneratedRecordSet]

Parameters:

imednet_workflows.uat.inspector module

Study metadata inspector for UAT specification generation.

class imednet_workflows.uat.inspector.FormVariableMap[source]

Bases: TypedDict

Mapping of a form to its associated variables.

class imednet_workflows.uat.inspector.StudySchemaInspector[source]

Bases: object

Fetch and index all structural metadata for an iMednet study.

Supports both sync and async SDK clients. When an async client is used, metadata calls are executed concurrently via asyncio.gather. The in-memory cache is per inspector instance. It can be shared by concurrent tasks in a single event loop, but is not thread-safe across threads/processes; use external synchronization or separate inspector instances for that access pattern.

__init__(sdk)[source]

Initialize the study schema inspector.

Parameters:

sdk (ImednetFacade | AsyncImednetFacade) – An instance of the synchronous or asynchronous iMednet SDK facade.

Return type:

None

async async_inspect(study_key, *, force_refresh=False)[source]

Asynchronously fetch metadata concurrently and return a study snapshot.

Return type:

StudySnapshot

Parameters:

  • study_key (str) –

  • force_refresh (bool) –

clear_cache(study_key=None)[source]

Clear snapshot cache for one study key or all keys.

Return type:

None

Parameters:

study_key (str | None) –

inspect(study_key, *, force_refresh=False)[source]

Synchronously fetch and return a study snapshot.

Return type:

StudySnapshot

Parameters:

  • study_key (str) –

  • force_refresh (bool) –

class imednet_workflows.uat.inspector.StudySnapshot[source]

Bases: ImednetBaseModel

Point-in-time snapshot of a study’s structural metadata.

active_sites()[source]

Return sites that are actively enrolling.

Return type:

list[Site]

enrollment_forms()[source]

Return forms with form_type indicating subject registration.

Return type:

list[Form]

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(_StudySnapshot__context)[source]

Initialize derived lookup dictionaries after model initialization.

Return type:

None

Parameters:

_StudySnapshot__context (object) –

scheduled_forms()[source]

Return scheduled (non-enrollment, non-unscheduled) forms.

Return type:

list[Form]

unscheduled_forms()[source]

Return unscheduled forms.

Return type:

list[Form]

imednet_workflows.uat.models module

Pydantic models describing a complete UAT specification.

class imednet_workflows.uat.models.RecordTestType[source]

Bases: str, Enum

What kind of record submission is being tested.

class imednet_workflows.uat.models.UATFormSpec[source]

Bases: UATBaseModel

Form-level UAT test specification.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_subject_count(value)[source]

Enforce safe subject count bounds.

Return type:

int

Parameters:

value (int) –

classmethod validate_unique_variable_names(variables)[source]

Ensure variables are uniquely identified by variable_name.

Return type:

list[UATVariableSpec]

Parameters:

variables (list[imednet_workflows.uat.models.UATVariableSpec]) –

class imednet_workflows.uat.models.UATSpecification[source]

Bases: UATBaseModel

Root model for a complete UAT test plan.

enabled_forms()[source]

Return only enabled forms.

Return type:

list[UATFormSpec]

forms_by_type(test_type)[source]

Return enabled forms matching a specific test type.

Return type:

list[UATFormSpec]

Parameters:

test_type (RecordTestType) –

classmethod from_json(payload)[source]

Load a UAT specification from JSON text, bytes, or file path.

Return type:

UATSpecification

Parameters:

payload (str | bytes | Path) –

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_spec_version(value)[source]

Pin this release to v1.0 schema.

Return type:

str

Parameters:

value (str) –

class imednet_workflows.uat.models.UATSubjectSpec[source]

Bases: UATBaseModel

Defines synthetic test subjects to register before form submission.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_subject_count(value)[source]

Enforce safe subject count bounds.

Return type:

int

Parameters:

value (int) –

class imednet_workflows.uat.models.UATVariableSpec[source]

Bases: UATBaseModel

Variable-level test specification.

model_config: ClassVar[ConfigDict] = {'alias_generator': <function to_camel>, 'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_fixed_strategy(data)[source]

Require a fixed value when FIXED strategy is selected.

Return type:

Any

Parameters:

data (Any) –

class imednet_workflows.uat.models.VariableTestStrategy[source]

Bases: str, Enum

How a variable’s test value should be determined.

imednet_workflows.uat.orchestrator module

End-to-end UAT execution orchestrator.

class imednet_workflows.uat.orchestrator.UATRunPhase[source]

Bases: str, Enum

Phases of a UAT workflow execution.

class imednet_workflows.uat.orchestrator.UATRunResult[source]

Bases: ImednetBaseModel

Complete result of a UATWorkflow.run() execution.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property overall_success: bool

Return True if all jobs in the run were successful.

summary()[source]

Return a human-readable summary string suitable for CLI output.

Return type:

str

class imednet_workflows.uat.orchestrator.UATSpecificationBuilder[source]

Bases: object

Auto-generate a UATSpecification from a StudySnapshot.

The generated spec uses SYNTHETIC strategy for all variables. Users can then edit the JSON output to apply FIXED or SKIP strategies for specific variables before feeding it back into UATWorkflow.run().

build(snapshot, *, subject_count=2, site_name=None, enabled_form_keys=None, seed=None)[source]

Build a complete UATSpecification from a StudySnapshot.

Return type:

UATSpecification

Parameters:

  • snapshot (StudySnapshot) –

  • subject_count (int) –

  • site_name (str | None) –

  • enabled_form_keys (list[str] | None) –

  • seed (int | None) –

Parameters

snapshotStudySnapshot

The study metadata snapshot from StudySchemaInspector.

subject_countint

Number of test subjects to register. Default 2.

site_nameOptional[str]

Site to register subjects at. If None, uses the first active site.

enabled_form_keysOptional[List[str]]

Restrict to a subset of forms. If None, all non-disabled forms are included.

seedOptional[int]

Passed through to UATSpecification for traceability.

class imednet_workflows.uat.orchestrator.UATWorkflow[source]

Bases: object

End-to-end UAT execution coordinator.

Composes StudySchemaInspector, UATSpecificationBuilder, SyntheticRecordGenerator, BulkRecordSubmissionWorkflow, and JobPoller into a single pipeline.

__init__(sdk, *, batch_size=100, poll_interval=10.0, poll_timeout=600.0, seed=None, on_progress=None)[source]

Initialize the UAT workflow orchestrator.

Parameters:

  • sdk (ImednetFacade) –

  • batch_size (int) –

  • poll_interval (float) –

  • poll_timeout (float) –

  • seed (int | None) –

  • on_progress (JobProgressCallback | None) –

Return type:

None

build_spec(snapshot, **kwargs)[source]

Auto-generate a UATSpecification from the StudySnapshot.

Return type:

UATSpecification

Parameters:

  • snapshot (Any) –

  • kwargs (Any) –

generate(spec, snapshot)[source]

Create synthetic record payloads via SyntheticRecordGenerator.

Return type:

list[GeneratedRecordSet]

Parameters:
inspect(study_key)[source]

Fetch study metadata via StudySchemaInspector.

Return type:

Any

Parameters:

study_key (str) –

monitor(study_key, submission_result)[source]

Poll all resulting jobs to completion via JobPoller.

Return type:

Any

Parameters:
run(study_key, *, spec=None, subject_count=2, site_name=None, enabled_form_keys=None, email_notify=None)[source]

Execute the full UAT pipeline and return a UATRunResult.

Return type:

UATRunResult

Parameters:

  • study_key (str) –

  • spec (UATSpecification | None) –

  • subject_count (int) –

  • site_name (str | None) –

  • enabled_form_keys (list[str] | None) –

  • email_notify (bool | str | None) –

submit(study_key, record_sets, email_notify=None)[source]

Execute the two-phase bulk submission.

Return type:

SubmissionResult

Parameters:

imednet_workflows.uat.submission module

Two-phase bulk record submission workflow for UAT.

class imednet_workflows.uat.submission.BatchSubmission[source]

Bases: ImednetBaseModel

Tracks a single API call to records.create().

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class imednet_workflows.uat.submission.BulkRecordSubmissionWorkflow[source]

Bases: object

Two-phase bulk record submission for UAT scenarios.

Phase 1: Submit all RegisterSubjectRequest payloads and await job completion. Phase 2: Submit all UpdateScheduledRecordRequest and CreateNewRecordRequest payloads.

Parameters

sdkImednetFacade

Initialized synchronous SDK instance.

batch_sizeint

Maximum number of records per API call. Default: 100.

await_registrationbool

If True, block until all Phase 1 jobs reach terminal state before submitting Phase 2 records. Default: True.

registration_timeoutfloat

Seconds to wait for Phase 1 jobs. Default: 600.

skip_existing_subjectsbool

If True, query the API for existing UAT-tagged subjects and skip registration for any that already exist. Default: True.

__init__(sdk, batch_size=100, await_registration=True, registration_timeout=600.0, skip_existing_subjects=True)[source]

Initialize the workflow.

Parameters:

  • sdk (ImednetFacade) –

  • batch_size (int) –

  • await_registration (bool) –

  • registration_timeout (float) –

  • skip_existing_subjects (bool) –

Return type:

None

submit(study_key, record_sets, *, email_notify=None, keyword_tag='UAT')[source]

Execute the two-phase submission.

Returns a SubmissionResult with all jobs and metadata. Raises BulkSubmissionError if Phase 1 jobs fail and Phase 2 depends on them.

Return type:

SubmissionResult

Parameters:
exception imednet_workflows.uat.submission.BulkSubmissionError[source]

Bases: Exception

Raised when Phase 1 registration jobs fail, blocking Phase 2 submission.

__init__(message, failed_batches)[source]

Initialize the error.

Parameters:

  • message (str) – The error message.

  • failed_batches (list[BatchSubmission]) – List of batches that failed during submission.

Return type:

None

class imednet_workflows.uat.submission.SubmissionResult[source]

Bases: ImednetBaseModel

Aggregate result of a full UAT submission run.

property all_batch_ids: list[str]

Return all batch IDs from all submitted jobs.

property all_batches: list[imednet_workflows.uat.submission.BatchSubmission]

Return all submission batches from both phases.

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'populate_by_name': True, 'str_strip_whitespace': True, 'validate_by_alias': True, 'validate_by_name': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property total_records_submitted: int

Return the total count of records submitted across all batches.