imednet.integrations package

Integration helpers for exporting study data.

class imednet.integrations.ExportSink[source]

Bases: ABC

Abstract base class for all export sinks.

Subclasses must implement write_batch(), flush(), and close(). The context-manager protocol is provided by this class.

Parameters

config:

Shared sink configuration. Defaults to SinkConfig with all values at their defaults.

__init__(config)[source]

Initialize the export sink with a configuration.

Parameters:

config (SinkConfig) –

Return type:

None

abstract close()[source]

Release all resources held by this sink (connections, file handles).

Implementations must be idempotent — calling close() on an already closed sink must not raise.

Return type:

None

abstract flush()[source]

Flush any internal buffers to the destination.

Return type:

None

Raises:

~imednet.errors.ExportError

On flush failure.

abstract write_batch(records, *, batch_id)[source]

Write one batch of records to the destination.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

Parameters

records:

Sequence of records to write. The concrete type depends on the export path:

  • Tabular path – pandas.DataFrame rows or plain dicts produced by RecordMapper.

  • Structure-preserving path – typed Record instances from DataExtractionWorkflow.

  • Warehouse path – pyarrow.RecordBatch or dicts destined for a staged Parquet file.

batch_id:

Caller-supplied idempotency key. Recommended format: "<study_key>/<form_key>/<batch_number>".

Returns:

int

Number of records successfully written.

Raises:

~imednet.errors.ExportBatchError

When the batch cannot be written after all retries.

class imednet.integrations.PartitionedStorageEngine[source]

Bases: Protocol

Interface for writing tables into a partitioned dataset.

__init__(*args, **kwargs)
write_form_table(table, *, base_dir, study_key, form_key)[source]

Persist a form table into a partitioned dataset layout.

Return type:

None

Parameters:

  • table (Any) –

  • base_dir (str) –

  • study_key (str) –

  • form_key (str) –

class imednet.integrations.PyArrowDatasetPartitionedStorageEngine[source]

Bases: PartitionedStorageEngine

Partitioned storage engine powered by pyarrow.dataset.write_dataset.

__init__(compression='snappy', use_dictionary=True, existing_data_behavior='overwrite_or_ignore', staging_dir_name='.imednet_staging')
Parameters:

  • compression (str) –

  • use_dictionary (bool) –

  • existing_data_behavior (str) –

  • staging_dir_name (str) –

Return type:

None

write_form_table(table, *, base_dir, study_key, form_key)[source]

Persist a form table into a partitioned dataset layout using PyArrow.

This implementation uses a staging directory and atomic renames to ensure that data is written consistently.

Parameters:

  • table (Any) – The pyarrow table to write.

  • base_dir (str) – The base directory for the Hive dataset.

  • study_key (str) – The iMednet study key (used for partitioning).

  • form_key (str) – The iMednet form key (used for partitioning).

Raises:

RuntimeError – If the write operation fails or produces unexpected output.

Return type:

None

class imednet.integrations.SinkConfig[source]

Bases: object

Shared configuration for all export sinks.

Parameters

study_key:

Mandatory study key for the export.

batch_size:

Number of records per ExportSink.write_batch() call.

max_retries:

Maximum number of retry attempts on transient errors.

retry_backoff:

Base delay in seconds between retry attempts. The actual delay grows exponentially: retry_backoff * 2 ** attempt.

idempotent:

When True, sinks use upsert / replace semantics so that replaying a batch with the same batch_id produces no duplicate data.

__init__(study_key, batch_size=500, max_retries=3, retry_backoff=1.0, idempotent=True, extra=<factory>, quality_gate_enabled=False, min_schema_readiness_score=100.0, tracer=None)
Parameters:

  • study_key (str) –

  • batch_size (int) –

  • max_retries (int) –

  • retry_backoff (float) –

  • idempotent (bool) –

  • extra (dict[str, Any]) –

  • quality_gate_enabled (bool) –

  • min_schema_readiness_score (float) –

  • tracer (Any | None) –

Return type:

None

imednet.integrations.export_to_csv(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to a CSV file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export_to_duckdb(sdk, study_key, db_path, table_name, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None)[source]

Export study records to a DuckDB table using native DataFrame registration.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • db_path (str) –

  • table_name (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

Parameters

sdk:

Authenticated SDK instance used to fetch study records.

study_key:

Study identifier to export.

db_path:

Path to the target .duckdb database file.

table_name:

Name of the destination DuckDB table.

use_labels_as_columns:

When True, variable labels are used for DataFrame column names.

variable_whitelist:

Optional list of variable names to include.

form_whitelist:

Optional list of form IDs to include.

Raises:

ImportError

If the optional duckdb dependency is not installed.

imednet.integrations.export_to_duckdb_by_form(sdk, study_key, db_path, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None)[source]

Export records to separate DuckDB tables for each form.

Each form is exported to a table named after form.form_key.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • db_path (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

Parameters

sdk:

Authenticated SDK instance used to fetch study records.

study_key:

Study identifier to export.

db_path:

Path to the target .duckdb database file.

use_labels_as_columns:

When True, variable labels are used for DataFrame column names.

variable_whitelist:

Optional list of variable names to include.

form_whitelist:

Optional list of form IDs to include.

Raises:

ImportError

If the optional duckdb dependency is not installed.

imednet.integrations.export_to_excel(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to an Excel workbook.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export_to_hive_parquet(sdk, study_key, base_dir, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, chunk_size=5000)[source]

Export study records to a Hive-partitioned Parquet directory layout.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • base_dir (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • chunk_size (int) –

imednet.integrations.export_to_json(sdk, study_key, path, *, use_labels_as_columns=False, hierarchical=False, **kwargs)[source]

Export study records to a JSON file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • hierarchical (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

hierarchical:

When True, generates a nested tree (Subject > Visit > Form) suitable for Veeva Vault integrations instead of a flat tabular layout.

imednet.integrations.export_to_long_sql(sdk, study_key, table_name, conn_str, *, chunk_size=1000)[source]

Export records to a normalized long-format SQL table.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • table_name (str) –

  • conn_str (str) –

  • chunk_size (int) –

imednet.integrations.export_to_parquet(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to a Parquet file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export_to_sql(sdk, study_key, table, conn_str, if_exists='replace', *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, **kwargs)[source]

Export study records to a SQL table.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • table (str) –

  • conn_str (str) –

  • if_exists (Literal['fail', 'replace', 'append', 'delete_rows']) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export_to_sql_by_form(sdk, study_key, conn_str, if_exists='replace', *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, **kwargs)[source]

Export records to separate SQL tables for each form.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • conn_str (str) –

  • if_exists (Literal['fail', 'replace', 'append', 'delete_rows']) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • kwargs (Any) –

imednet.integrations.hive_parquet_query(base_dir)[source]

Return the DuckDB read_parquet query string for the given Hive base directory.

Return type:

str

Parameters:

base_dir (str) –

Submodules

imednet.integrations.dispatcher module

Unified functional facade for data export.

This module provides a single entry point export() for all data persistence tasks, supporting both tabular procedural functions and object-oriented sink classes.

class imednet.integrations.dispatcher.ExportRegistry[source]

Bases: object

Central registry mapping target types to their implementations.

__init__()[source]

Initialize an empty ExportRegistry.

Return type:

None

get_config_class(target_type)[source]

Retrieve a registered config class, or SinkConfig as default.

Return type:

type[SinkConfig]

Parameters:

target_type (str) –

get_sink(target_type)[source]

Retrieve a registered sink class, or None if not found.

Return type:

Optional[type[ExportSink]]

Parameters:

target_type (str) –

get_tabular(target_type)[source]

Retrieve a registered tabular function, or None if not found.

Return type:

Optional[Callable[..., Any]]

Parameters:

target_type (str) –

register_sink(target_type, sink_class)[source]

Register an object-oriented sink class for a target type.

Return type:

None

Parameters:
register_tabular(target_type, func)[source]

Register a tabular procedural function for a target type.

Return type:

None

Parameters:

  • target_type (str) –

  • func (Callable[[...], Any]) –

imednet.integrations.dispatcher.export(target, sdk, study_key, *, config=None, **kwargs)[source]

Unified entry point for data export.

Routes the export request to either a tabular function or a sink class based on the target identifier.

Parameters:

  • target (str) – The target destination type (e.g., ‘csv’, ‘snowflake’, ‘mongodb’).

  • sdk (ImednetSDK) – Authenticated SDK instance used to fetch study records.

  • study_key (str) – Study identifier to export.

  • config (Optional[SinkConfig]) – Optional sink configuration (for sink-based targets).

  • **kwargs (Any) – Target-specific configuration parameters.

Returns:

Typically None. For sink targets: The total number of records successfully written.

Return type:

For tabular targets

Raises:

ValueError – If the target type is not registered or unsupported.

imednet.integrations.dispatcher.register_sink_target(target_type, sink_class)[source]

Register an object-oriented sink class for a target type.

Return type:

None

Parameters:
imednet.integrations.dispatcher.register_tabular_target(target_type, func)[source]

Register a tabular procedural function for a target type.

Return type:

None

Parameters:

  • target_type (str) –

  • func (Callable[[...], Any]) –

imednet.integrations.enrichment module

Data enrichment and transformation pipeline.

This module provides the EnrichmentPipeline for transforming raw API data using configured mappings, terminology lookups, and PHI masking rules.

class imednet.integrations.enrichment.CentralizedMapper[source]

Bases: object

Unified Mapper with Enrichment Engine for data sinks.

__init__(mode='document', post_processor=None)[source]

Initialize the mapper.

Parameters:

  • mode (str) – “tabular” for flattened structures, “document” for nested structures.

  • post_processor (Optional[Callable[[dict[str, Any]], dict[str, Any]]]) – Optional function to apply destination-specific formatting.

map_record(record, study_key=None)[source]

Map a clinical record to the unified destination format.

Return type:

dict[str, Any]

Parameters:

  • record (Any) –

  • study_key (str | None) –

class imednet.integrations.enrichment.EnrichmentPipeline[source]

Bases: object

Pipeline for transforming and enriching study data.

Applies a set of rules defined in a StudyConfiguration to data objects, including PHI masking, terminology mapping, and custom business logic.

__init__(config)[source]

Initialize the enrichment pipeline.

Parameters:

config (StudyConfiguration) – The study configuration containing enrichment rules.

process(data)[source]

Process the input data through the enrichment pipeline.

Parameters:

data (Any) – The raw data to be enriched.

Return type:

Any

Returns:

The enriched and transformed data.

imednet.integrations.export module

Export helpers built on top of RecordMapper.

class imednet.integrations.export.TabularCSVSink[source]

Bases: ExportSink

Sink for exporting data to CSV format.

__init__(path, config)[source]

Initialize the CSV sink.

Parameters:
close()[source]

Release all resources held by this sink (connections, file handles).

Implementations must be idempotent — calling close() on an already closed sink must not raise.

Return type:

None

flush()[source]

Flush any internal buffers to the destination.

Return type:

None

Raises:

~imednet.errors.ExportError

On flush failure.

write_batch(records, *, batch_id)[source]

Write a batch of records to the sink.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet.integrations.export.TabularSQLSink[source]

Bases: ExportSink

Sink for exporting data to SQL databases.

__init__(table, engine, config)[source]

Initialize the SQL sink.

Parameters:
close()[source]

Close the sink.

Return type:

None

flush()[source]

Flush the sink.

Return type:

None

write_batch(records, *, batch_id)[source]

Write a batch of records to the sink.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet.integrations.export.TabularSinkConfig[source]

Bases: SinkConfig

Configuration for tabular sinks.

__init__(study_key, batch_size=500, max_retries=3, retry_backoff=1.0, idempotent=True, extra=<factory>, quality_gate_enabled=False, min_schema_readiness_score=100.0, tracer=None, manifest_path=None, use_labels_as_columns=False, sanitize=False, variable_whitelist=None, form_whitelist=None, pandas_kwargs=<factory>)
Parameters:

  • study_key (str) –

  • batch_size (int) –

  • max_retries (int) –

  • retry_backoff (float) –

  • idempotent (bool) –

  • extra (dict[str, Any]) –

  • quality_gate_enabled (bool) –

  • min_schema_readiness_score (float) –

  • tracer (Any | None) –

  • manifest_path (str | None) –

  • use_labels_as_columns (bool) –

  • sanitize (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • pandas_kwargs (dict[str, Any]) –

Return type:

None

imednet.integrations.export.export_to_csv(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to a CSV file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export.export_to_duckdb(sdk, study_key, db_path, table_name, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None)[source]

Export study records to a DuckDB table using native DataFrame registration.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • db_path (str) –

  • table_name (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

Parameters

sdk:

Authenticated SDK instance used to fetch study records.

study_key:

Study identifier to export.

db_path:

Path to the target .duckdb database file.

table_name:

Name of the destination DuckDB table.

use_labels_as_columns:

When True, variable labels are used for DataFrame column names.

variable_whitelist:

Optional list of variable names to include.

form_whitelist:

Optional list of form IDs to include.

Raises:

ImportError

If the optional duckdb dependency is not installed.

imednet.integrations.export.export_to_duckdb_by_form(sdk, study_key, db_path, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None)[source]

Export records to separate DuckDB tables for each form.

Each form is exported to a table named after form.form_key.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • db_path (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

Parameters

sdk:

Authenticated SDK instance used to fetch study records.

study_key:

Study identifier to export.

db_path:

Path to the target .duckdb database file.

use_labels_as_columns:

When True, variable labels are used for DataFrame column names.

variable_whitelist:

Optional list of variable names to include.

form_whitelist:

Optional list of form IDs to include.

Raises:

ImportError

If the optional duckdb dependency is not installed.

imednet.integrations.export.export_to_excel(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to an Excel workbook.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export.export_to_json(sdk, study_key, path, *, use_labels_as_columns=False, hierarchical=False, **kwargs)[source]

Export study records to a JSON file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • hierarchical (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

hierarchical:

When True, generates a nested tree (Subject > Visit > Form) suitable for Veeva Vault integrations instead of a flat tabular layout.

imednet.integrations.export.export_to_long_sql(sdk, study_key, table_name, conn_str, *, chunk_size=1000)[source]

Export records to a normalized long-format SQL table.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • table_name (str) –

  • conn_str (str) –

  • chunk_size (int) –

imednet.integrations.export.export_to_parquet(sdk, study_key, path, *, use_labels_as_columns=False, **kwargs)[source]

Export study records to a Parquet file.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • path (str) –

  • use_labels_as_columns (bool) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export.export_to_sql(sdk, study_key, table, conn_str, if_exists='replace', *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, **kwargs)[source]

Export study records to a SQL table.

All exports now inherit StudyConfiguration rules by default.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • table (str) –

  • conn_str (str) –

  • if_exists (Literal['fail', 'replace', 'append', 'delete_rows']) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • kwargs (Any) –

Parameters

use_labels_as_columns:

When True, variable labels are used for column names instead of variable names.

imednet.integrations.export.export_to_sql_by_form(sdk, study_key, conn_str, if_exists='replace', *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, **kwargs)[source]

Export records to separate SQL tables for each form.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • conn_str (str) –

  • if_exists (Literal['fail', 'replace', 'append', 'delete_rows']) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • kwargs (Any) –

imednet.integrations.parquet module

Hive-partitioned Parquet integration helpers.

imednet.integrations.parquet.export_to_hive_parquet(sdk, study_key, base_dir, *, use_labels_as_columns=False, variable_whitelist=None, form_whitelist=None, chunk_size=5000)[source]

Export study records to a Hive-partitioned Parquet directory layout.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • base_dir (str) –

  • use_labels_as_columns (bool) –

  • variable_whitelist (list[str] | None) –

  • form_whitelist (list[int] | None) –

  • chunk_size (int) –

imednet.integrations.parquet.hive_parquet_query(base_dir)[source]

Return the DuckDB read_parquet query string for the given Hive base directory.

Return type:

str

Parameters:

base_dir (str) –

imednet.integrations.parquet_engine module

Partitioned Parquet storage engines.

class imednet.integrations.parquet_engine.PartitionedStorageEngine[source]

Bases: Protocol

Interface for writing tables into a partitioned dataset.

__init__(*args, **kwargs)
write_form_table(table, *, base_dir, study_key, form_key)[source]

Persist a form table into a partitioned dataset layout.

Return type:

None

Parameters:

  • table (Any) –

  • base_dir (str) –

  • study_key (str) –

  • form_key (str) –

class imednet.integrations.parquet_engine.PyArrowDatasetPartitionedStorageEngine[source]

Bases: PartitionedStorageEngine

Partitioned storage engine powered by pyarrow.dataset.write_dataset.

__init__(compression='snappy', use_dictionary=True, existing_data_behavior='overwrite_or_ignore', staging_dir_name='.imednet_staging')
Parameters:

  • compression (str) –

  • use_dictionary (bool) –

  • existing_data_behavior (str) –

  • staging_dir_name (str) –

Return type:

None

write_form_table(table, *, base_dir, study_key, form_key)[source]

Persist a form table into a partitioned dataset layout using PyArrow.

This implementation uses a staging directory and atomic renames to ensure that data is written consistently.

Parameters:

  • table (Any) – The pyarrow table to write.

  • base_dir (str) – The base directory for the Hive dataset.

  • study_key (str) – The iMednet study key (used for partitioning).

  • form_key (str) – The iMednet form key (used for partitioning).

Raises:

RuntimeError – If the write operation fails or produces unexpected output.

Return type:

None

imednet.integrations.sink_base module

Shared base classes and helpers for all export sinks.

Architecture decision

The SDK provides three export paths:

Tabular path (RecordMapper + pandas.DataFrame)

Flattens record data into a wide DataFrame and writes it to CSV, Excel, JSON, SQL, DuckDB, or Parquet. All functions in imednet.integrations.export follow this path.

Structure-preserving path (DataExtractionWorkflow + typed records)

Traverses the clinical data hierarchy Study → Subject → Visit → Record and materialises the relationships into a destination that can represent them natively — a property graph (Neo4j) or a nested document store (MongoDB).

Warehouse path (staged Parquet + native bulk loader)

Writes one Parquet file per form to an intermediate staging area and then invokes the destination’s native bulk-loading command (e.g. Snowflake COPY INTO).

All three paths share the contracts defined in this module: SinkConfig, the ExportSink ABC, the import-guard helper _require_optional_dep(), and the credential-redaction helper _redact_uri().

Shared contracts

  • Batching – callers split records into batches and call ExportSink.write_batch() once per batch. The batch_id parameter is a caller-supplied idempotency key (e.g. "<study_key>/<form_key>/<n>").

  • Chunk sizing – SinkConfig.batch_size controls the number of records per batch (default 500).

  • Retries – sinks must honour SinkConfig.max_retries and use SinkConfig.retry_backoff as the base delay between attempts.

  • Idempotent writes – when SinkConfig.idempotent is True (default) sinks must use upsert semantics or CREATE OR REPLACE so that replaying a batch with the same batch_id produces no duplicate data.

  • Error propagation – transient errors are retried; permanent errors raise ExportBatchError (includes batch_id) or ExportConfigurationError.

  • Logging – sinks use logging.getLogger(__name__) and must not log raw credentials or full URIs. Pass URIs through _redact_uri() before logging.

Optional dependency conventions

  • Each sink module calls _require_optional_dep() at connection time (not at import time) so that importing the module never fails due to a missing optional library.

  • Extras keys follow the pattern imednet[<key>]:

    pip install 'imednet[neo4j]'
pip install 'imednet[mongodb]'
pip install 'imednet[snowflake]'

Public-API exposure rules

  • imednet.integrations re-exports only the tabular helpers by default (backward compatibility).

  • The three new sink classes (Neo4jExportSink, MongoDbExportSink, SnowflakeExportSink) and SinkConfig are importable from their respective submodules and are also re-exported from imednet.integrations via explicit names.

  • Airflow helpers in apache_airflow_providers_imednet.export wrap only the tabular path; graph/document/warehouse sinks are not wrapped there.

class imednet.integrations.sink_base.ExportSink[source]

Bases: ABC

Abstract base class for all export sinks.

Subclasses must implement write_batch(), flush(), and close(). The context-manager protocol is provided by this class.

Parameters

config:

Shared sink configuration. Defaults to SinkConfig with all values at their defaults.

__init__(config)[source]

Initialize the export sink with a configuration.

Parameters:

config (SinkConfig) –

Return type:

None

abstract close()[source]

Release all resources held by this sink (connections, file handles).

Implementations must be idempotent — calling close() on an already closed sink must not raise.

Return type:

None

abstract flush()[source]

Flush any internal buffers to the destination.

Return type:

None

Raises:

~imednet.errors.ExportError

On flush failure.

abstract write_batch(records, *, batch_id)[source]

Write one batch of records to the destination.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

Parameters

records:

Sequence of records to write. The concrete type depends on the export path:

  • Tabular path – pandas.DataFrame rows or plain dicts produced by RecordMapper.

  • Structure-preserving path – typed Record instances from DataExtractionWorkflow.

  • Warehouse path – pyarrow.RecordBatch or dicts destined for a staged Parquet file.

batch_id:

Caller-supplied idempotency key. Recommended format: "<study_key>/<form_key>/<batch_number>".

Returns:

int

Number of records successfully written.

Raises:

~imednet.errors.ExportBatchError

When the batch cannot be written after all retries.

class imednet.integrations.sink_base.SinkConfig[source]

Bases: object

Shared configuration for all export sinks.

Parameters

study_key:

Mandatory study key for the export.

batch_size:

Number of records per ExportSink.write_batch() call.

max_retries:

Maximum number of retry attempts on transient errors.

retry_backoff:

Base delay in seconds between retry attempts. The actual delay grows exponentially: retry_backoff * 2 ** attempt.

idempotent:

When True, sinks use upsert / replace semantics so that replaying a batch with the same batch_id produces no duplicate data.

__init__(study_key, batch_size=500, max_retries=3, retry_backoff=1.0, idempotent=True, extra=<factory>, quality_gate_enabled=False, min_schema_readiness_score=100.0, tracer=None)
Parameters:

  • study_key (str) –

  • batch_size (int) –

  • max_retries (int) –

  • retry_backoff (float) –

  • idempotent (bool) –

  • extra (dict[str, Any]) –

  • quality_gate_enabled (bool) –

  • min_schema_readiness_score (float) –

  • tracer (Any | None) –

Return type:

None

imednet.integrations.sink_base.iter_batches(records, batch_size)[source]

Yield records in chunks of batch_size.

Return type:

Iterator[Sequence[Any]]

Parameters:

  • records (Sequence[Any]) –

  • batch_size (int) –