imednet.workflows package

Submodules

imednet.workflows.data_extraction module

Provides workflows for extracting specific datasets from iMednet studies.

class imednet.workflows.data_extraction.DataExtractionWorkflow(sdk)

Bases: object

Provides methods for complex data extraction tasks involving multiple iMednet endpoints.

Parameters:

sdk (ImednetSDK) – An instance of the ImednetSDK.

extract_audit_trail(study_key, start_date=None, end_date=None, user_filter=None, **filters)

Extracts the audit trail (record revisions) based on specified filters.

Parameters:

• study_key (str) – The key identifying the study.

• start_date (Optional[str]) – Optional start date filter (YYYY-MM-DD format expected by API).

• end_date (Optional[str]) – Optional end date filter (YYYY-MM-DD format expected by API).

• user_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – Optional dictionary of base filter conditions.

• **filters (Any) – Additional key-value pairs to be added as equality filters.

Return type:

List[RecordRevision]

Returns:

A list of RecordRevision objects matching the criteria.

extract_records_by_criteria(study_key, record_filter=None, subject_filter=None, visit_filter=None, **other_filters)

Extracts records based on criteria spanning subjects, visits, and records.

Parameters:

• study_key (str) – The key identifying the study.

• record_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – Dictionary of conditions for the records endpoint.

• subject_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – Dictionary of conditions for the subjects endpoint.

• visit_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – Dictionary of conditions for the visits endpoint.

• **other_filters (Any) – Additional keyword arguments passed as filters to the records endpoint list method.

Return type:

List[Record]

Returns:

A list of Record objects matching all specified criteria.

imednet.workflows.query_management module

Provides workflows for managing queries within iMednet studies.

class imednet.workflows.query_management.QueryManagementWorkflow(sdk)

Bases: object

Provides methods for common query management tasks.

Parameters:

sdk (ImednetSDK) – An instance of the ImednetSDK.

get_open_queries(study_key, additional_filter=None, **kwargs)

Retrieves all open queries for a given study, potentially filtered further.

An ‘open’ query is defined as one where the query comment with the highest sequence number has its ‘closed’ field set to False.

Note: This method fetches queries based on additional_filter and then filters for the ‘open’ state client-side.

Parameters:

• study_key (str) – The key identifying the study.

• additional_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – An optional dictionary of conditions to apply via the API.

• **kwargs (Any) – Additional keyword arguments passed directly to sdk.queries.list.

Return type:

List[Query]

Returns:

A list of open Query objects matching the criteria.

get_queries_by_site(study_key, site_key, additional_filter=None, **kwargs)

Retrieves all queries for a specific site within a study.

Parameters:

• study_key (str) – The key identifying the study.

• site_key (str) – The name of the site.

• additional_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – Extra conditions to combine with the subject filter.

• **kwargs (Any) – Additional keyword arguments passed directly to sdk.queries.list.

Return type:

List[Query]

Returns:

A list of Query objects for the specified site.

get_queries_for_subject(study_key, subject_key, additional_filter=None, **kwargs)

Retrieves all queries for a specific subject within a study.

Parameters:

• study_key (str) – The key identifying the study.

• subject_key (str) – The key identifying the subject.

• additional_filter (Optional[Dict[str, Union[Any, Tuple[str, Any], List[Any]]]]) – An optional dictionary of conditions to combine with the subject filter.

• **kwargs (Any) – Additional keyword arguments passed directly to sdk.queries.list.

Return type:

List[Query]

Returns:

A list of Query objects for the specified subject.

get_query_state_counts(study_key, **kwargs)

Counts queries grouped by their current state (open/closed/unknown).

The state is determined by the ‘closed’ field of the query comment with the highest sequence number. Queries without any comments are counted as ‘unknown’.

Note: This method fetches all queries matching the base criteria (if any are passed via kwargs) and then performs the aggregation client-side.

Parameters:

• study_key (str) – The key identifying the study.

• **kwargs (Any) – Additional keyword arguments passed directly to sdk.queries.list (e.g., for initial filtering before counting).

Return type:

Dict[str, int]

Returns:

A dictionary with keys ‘open’, ‘closed’, ‘unknown’ and their respective counts.

imednet.workflows.record_mapper module

class imednet.workflows.record_mapper.RecordMapper(sdk)

Bases: object

Maps EDC records for a study into a pandas DataFrame.

Features:

• Fetches variable definitions for column mapping.

• Dynamically creates a Pydantic model for type validation of record data.

• Fetches records, applying server-side filtering where possible.

• Merges metadata and record data.

• Offers choice between variable names or labels for column headers.

• Handles parsing errors gracefully for individual records.

Example

sdk = ImednetSDK(api_key, security_key, base_url) mapper = RecordMapper(sdk) # Get DataFrame with labels as columns, filtered by visit df_labels = mapper.dataframe(study_key=”MYSTUDY”, visit_key=”VISIT1”) # Get DataFrame with variable names as columns df_names = mapper.dataframe(study_key=”MYSTUDY”, use_labels_as_columns=False)

Parameters:

sdk (ImednetSDK) –

dataframe(study_key, visit_key=None, use_labels_as_columns=True, variable_whitelist=None, form_whitelist=None)

Return a pandas.DataFrame of records for a study.

Return type:

DataFrame

Parameters:

• study_key (str) –

• visit_key (str | None) –

• use_labels_as_columns (bool) –

• variable_whitelist (List[str] | None) –

• form_whitelist (List[int] | None) –

imednet.workflows.record_update module

Utilities for submitting and updating records in iMedNet studies.

class imednet.workflows.record_update.RecordUpdateWorkflow(sdk)

Bases: object

Provides workflows for creating or updating records, including batch submission and optional job status monitoring.

Parameters:

sdk (ImednetSDK) – An instance of the ImednetSDK.

async async_create_or_update_records(study_key, records_data, wait_for_completion=False, timeout=300, poll_interval=5)

Asynchronous variant of create_or_update_records().

Return type:

Job

Parameters:

• study_key (str) –

• records_data (List[Dict[str, Any]]) –

• wait_for_completion (bool) –

• timeout (int) –

• poll_interval (int) –

create_new_record(study_key, form_identifier, subject_identifier, data, form_identifier_type='key', subject_identifier_type='key', wait_for_completion=False, timeout=300, poll_interval=5)

Creates a new (unscheduled) record for an existing subject.

Parameters:

• study_key (str) – The study key.

• form_identifier (Union[str, int]) – The form key or ID.

• subject_identifier (Union[str, int]) – The subject key, ID, or OID.

• data (Dict[str, Any]) – The dictionary of record data (variable names and values).

• form_identifier_type (Literal['key', 'id']) – Whether form_identifier is a ‘key’ or ‘id’.

• subject_identifier_type (Literal['key', 'id', 'oid']) – Whether subject_identifier is a ‘key’, ‘id’, or ‘oid’.

• wait_for_completion (bool) – If True, wait for the job to complete.

• timeout (int) – Timeout in seconds for waiting.

• poll_interval (int) – Polling interval in seconds.

Return type:

Job

Returns:

The Job status object.

create_or_update_records(study_key, records_data, wait_for_completion=False, timeout=300, poll_interval=5)

Submit records for creation or update and optionally wait for completion.

Return type:

Job

Parameters:

• study_key (str) –

• records_data (List[Dict[str, Any]]) –

• wait_for_completion (bool) –

• timeout (int) –

• poll_interval (int) –

register_subject(study_key, form_identifier, site_identifier, data, form_identifier_type='key', site_identifier_type='name', wait_for_completion=False, timeout=300, poll_interval=5)

Registers a new subject by submitting a single record.

Parameters:

• study_key (str) – The study key.

• form_identifier (Union[str, int]) – The form key or ID.

• site_identifier (Union[str, int]) – The site name or ID.

• data (Dict[str, Any]) – The dictionary of record data (variable names and values).

• form_identifier_type (Literal['key', 'id']) – Whether form_identifier is a ‘key’ or ‘id’.

• site_identifier_type (Literal['name', 'id']) – Whether site_identifier is a ‘name’ or ‘id’.

• wait_for_completion (bool) – If True, wait for the job to complete.

• timeout (int) – Timeout in seconds for waiting.

• poll_interval (int) – Polling interval in seconds.

Return type:

Job

Returns:

The Job status object.

submit_record_batch(*args, **kwargs)

Return type:

Job

Parameters:

• args (Any) –

• kwargs (Any) –

update_scheduled_record(study_key, form_identifier, subject_identifier, interval_identifier, data, form_identifier_type='key', subject_identifier_type='key', interval_identifier_type='name', wait_for_completion=False, timeout=300, poll_interval=5)

Updates an existing scheduled record for a subject.

Parameters:

• study_key (str) – The study key.

• form_identifier (Union[str, int]) – The form key or ID.

• subject_identifier (Union[str, int]) – The subject key, ID, or OID.

• interval_identifier (Union[str, int]) – The interval name or ID.

• data (Dict[str, Any]) – The dictionary of record data (variable names and values).

• form_identifier_type (Literal['key', 'id']) – Whether form_identifier is a ‘key’ or ‘id’.

• subject_identifier_type (Literal['key', 'id', 'oid']) – Whether subject_identifier is a ‘key’, ‘id’, or ‘oid’.

• interval_identifier_type (Literal['name', 'id']) – Whether interval_identifier is a ‘name’ or ‘id’.

• wait_for_completion (bool) – If True, wait for the job to complete.

• timeout (int) – Timeout in seconds for waiting.

• poll_interval (int) – Polling interval in seconds.

Return type:

Job

Returns:

The Job status object.

RecordUpdateWorkflow uses SchemaCache to validate record data before submission when provided.

imednet.workflows.register_subjects module

Workflow for registering subjects (patients) in iMednet via the Records API. This workflow is self-contained and does not borrow from record_update.py. It provides a simple, robust interface for registering one or more subjects.

class imednet.workflows.register_subjects.RegisterSubjectsWorkflow(sdk)

Bases: object

Manages the registration of subjects using the iMedNet SDK.

Parameters:

sdk (ImednetSDK) –

_sdk

An instance of the ImednetSDK.

Type:

ImednetSDK

register_subjects(study_key, subjects, email_notify=None)

Registers multiple subjects in the specified study. Sites and subject identifiers are validated before submission.

Parameters:

• study_key (str) – The key identifying the study.

• subjects (List[RegisterSubjectRequest]) – A list of RegisterSubjectRequest objects, each defining a subject to be registered.

• email_notify (Optional[str]) – Optional email address to notify upon completion.

Return type:

Job

Returns:

A Job object representing the background job created for the registration request.

Raises:

ApiError – If the API call fails.

imednet.workflows.subject_data module

Provides a workflow to retrieve comprehensive data for a specific subject.

class imednet.workflows.subject_data.SubjectComprehensiveData(**data)

Bases: BaseModel

Structure to hold aggregated data for a subject.

Parameters:

• subject_details (Subject | None) –

• visits (List[Visit]) –

• records (List[Record]) –

• queries (List[Query]) –

model_config: ClassVar[ConfigDict] = {}

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

queries: List[Query]

records: List[Record]

subject_details: Optional[Subject]

visits: List[Visit]

class imednet.workflows.subject_data.SubjectDataWorkflow(sdk)

Bases: object

Provides methods to retrieve comprehensive data related to a specific subject.

Parameters:

sdk (ImednetSDK) – An instance of the ImednetSDK.

get_all_subject_data(study_key, subject_key)

Retrieves subject details, visits, records, and queries for a specific subject.

Parameters:

• study_key (str) – The key identifying the study.

• subject_key (str) – The key identifying the subject.

Return type:

SubjectComprehensiveData

Returns:

A SubjectComprehensiveData object containing the aggregated data.