imednet_sinks package

Unified Sinks plugin for iMednet.

class imednet_sinks.MongoDbExportSink[source]

Bases: ExportSink

Export iMednet records as MongoDB document envelopes.

Parameters

config:

Mandatory MongoDbSinkConfig.

raises ~imednet.errors.ExportConfigurationError:

When the client cannot connect to the server.

raises ImportError:

When the pymongo package is not installed.

__init__(config)[source]

Initialize the MongoDB export sink.

Parameters:

config (MongoDbSinkConfig) – Mandatory sink configuration.

Return type:

None

close()[source]

Close the underlying PyMongo client.

Return type:

None

flush()[source]

No-op: MongoDB writes are committed per bulk operation.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to MongoDB using upsert (idempotent) or insert.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.MongoDbSinkConfig[source]

Bases: SinkConfig

Configuration for MongoDB export sink.

__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, uri='', database='', collection='')
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) –

  • uri (str) –

  • database (str) –

  • collection (str) –

Return type:

None

class imednet_sinks.Neo4jExportSink[source]

Bases: ExportSink

Export iMednet records as Neo4j graph nodes and relationships.

Parameters

config:

Mandatory Neo4jSinkConfig.

raises ~imednet.errors.ExportConfigurationError:

When the driver cannot connect to the database.

raises ImportError:

When the neo4j package is not installed.

__init__(config)[source]

Initialize the Neo4j export sink.

Parameters:

config (Neo4jSinkConfig) – Mandatory Neo4j-specific sink configuration.

Return type:

None

close()[source]

Close the underlying Neo4j driver connection.

Return type:

None

flush()[source]

No-op: Neo4j writes are committed per transaction.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to Neo4j using MERGE (idempotent) or CREATE.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.Neo4jSinkConfig[source]

Bases: SinkConfig

Extended SinkConfig for Neo4j.

Parameters

database:

Target Neo4j database name (default "neo4j").

__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, uri='', auth=('', ''), database='neo4j')
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) –

  • uri (str) –

  • auth (tuple[str, str]) –

  • database (str) –

Return type:

None

class imednet_sinks.SnowflakeExportSink[source]

Bases: ExportSink

Stage Parquet files and bulk-load them into Snowflake.

Parameters

config:

SnowflakeSinkConfig containing all connection details and staging paths.

raises ~imednet.errors.ExportConfigurationError:

When the Snowflake connector cannot be initialised or the required configuration values are missing.

raises ImportError:

When snowflake-connector-python or pyarrow are not installed.

__init__(config)[source]

Initialize the Snowflake export sink.

Parameters:

config (SnowflakeSinkConfig) – Mandatory Snowflake-specific sink configuration.

Return type:

None

close()[source]

Close the Snowflake connection and clean up temporary staging files.

Return type:

None

flush()[source]

No-op: each batch is committed individually.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to Snowflake via Parquet staging + COPY INTO.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.SnowflakeSinkConfig[source]

Bases: SinkConfig

Configuration for SnowflakeExportSink.

Parameters

account:

Snowflake account identifier (<org>-<account> or legacy format).

user:

Snowflake user name.

password:

Snowflake password. Never logged.

database:

Target database.

schema:

Target schema.

warehouse:

Virtual warehouse used for the COPY INTO command.

stage:

Snowflake internal stage name (e.g. "MY_STAGE").

table:

Destination table name inside database.*schema*.

stage_prefix:

Path prefix inside the stage (default "imednet").

local_staging_dir:

Local directory used to write Parquet files before PUT. Defaults to a temporary directory created by tempfile.

manifest_path:

Optional path to a JSON-lines file where each loaded batch is recorded.

__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, account='', user='', password='', database='', schema='PUBLIC', warehouse='', stage='', table='', stage_prefix='imednet', local_staging_dir=None, manifest_path=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) –

  • account (str) –

  • user (str) –

  • password (str) –

  • database (str) –

  • schema (str) –

  • warehouse (str) –

  • stage (str) –

  • table (str) –

  • stage_prefix (str) –

  • local_staging_dir (str | PathLike[str] | None) –

  • manifest_path (str | PathLike[str] | None) –

Return type:

None

imednet_sinks.export_to_mongodb(sdk, study_key, uri='', database='', collection='', *, config=None)[source]

Export study records to MongoDB using MongoDbExportSink.

Return type:

int

Parameters:
imednet_sinks.export_to_neo4j(sdk, study_key, uri='', auth=('', ''), *, config=None)[source]

Export study records to Neo4j using Neo4jExportSink.

Return type:

int

Parameters:
imednet_sinks.export_to_snowflake(sdk, study_key, *, config)[source]

Export study records to Snowflake using SnowflakeExportSink.

Return type:

int

Parameters:

Submodules

imednet_sinks.document module

MongoDB document-envelope export sink.

This module implements the structure-preserving export path for a MongoDB destination. Records fetched via DataExtractionWorkflow are wrapped in a consistent document envelope that preserves the clinical hierarchy.

Document envelope

Each document stored in MongoDB represents a single iMednet Record and carries the metadata needed to locate and de-duplicate it:

{
    "_id": "<study_key>/<record_id>",
    "study_key":     "MYSTUDY",
    "record_id":     1234,
    "subject_id":    99,
    "subject_key":   "SUBJ-001",
    "visit_id":      42,
    "form_id":       7,
    "form_key":      "BASELINE",
    "record_status": "Complete",
    "deleted":       false,
    "date_created":  "2024-01-01T08:00:00Z",
    "date_modified": "2024-01-15T12:00:00Z",
    "record_data":   { "var1": "value1", "var2": 99 },
    "exported_at":   "2024-01-15T12:00:00Z"
}

The _id field is a composite key <study_key>/<record_id> which guarantees uniqueness within a collection and enables efficient upserts.

Optional dependency

Requires pymongo (install via pip install 'imednet[mongodb]'). The client is imported lazily at connection time.

Idempotency

When SinkConfig.idempotent is True (default) the sink uses bulk_write with UpdateOne(..., upsert=True) operations so that re-running an export for the same batch updates existing documents rather than inserting duplicates.

Usage

from imednet.integrations.document import MongoDbExportSink
from imednet_workflows.data_extraction import DataExtractionWorkflow

records = DataExtractionWorkflow(sdk).extract_records_by_criteria(
    study_key="MYSTUDY",
)

with MongoDbExportSink(
    uri="******localhost:27017",
    database="imednet",
    collection="records",
) as sink:
    for i, batch in enumerate(batched(records, 500)):
        sink.write_batch(batch, batch_id=f"MYSTUDY/all/{i}")
class imednet_sinks.document.MongoDbExportSink[source]

Bases: ExportSink

Export iMednet records as MongoDB document envelopes.

Parameters

config:

Mandatory MongoDbSinkConfig.

raises ~imednet.errors.ExportConfigurationError:

When the client cannot connect to the server.

raises ImportError:

When the pymongo package is not installed.

__init__(config)[source]

Initialize the MongoDB export sink.

Parameters:

config (MongoDbSinkConfig) – Mandatory sink configuration.

Return type:

None

close()[source]

Close the underlying PyMongo client.

Return type:

None

flush()[source]

No-op: MongoDB writes are committed per bulk operation.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to MongoDB using upsert (idempotent) or insert.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.document.MongoDbSinkConfig[source]

Bases: SinkConfig

Configuration for MongoDB export sink.

__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, uri='', database='', collection='')
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) –

  • uri (str) –

  • database (str) –

  • collection (str) –

Return type:

None

imednet_sinks.document.export_to_mongodb(sdk, study_key, uri='', database='', collection='', *, config=None)[source]

Export study records to MongoDB using MongoDbExportSink.

Return type:

int

Parameters:

imednet_sinks.graph module

Neo4j graph export sink.

This module implements the structure-preserving export path for a Neo4j property-graph destination. Records fetched via DataExtractionWorkflow are written as graph nodes and relationships that mirror the clinical data hierarchy.

Graph shape

Nodes

  • (:Study   {study_key})

  • (:Subject {subject_key, study_key})

  • (:Visit   {visit_id, subject_key, study_key})

  • (:Record  {record_id, form_id, visit_id, subject_key, study_key, **record_data})

Relationships

  • (:Study)-[:HAS_SUBJECT]->(:Subject)

  • (:Subject)-[:HAS_VISIT]->(:Visit)

  • (:Visit)-[:HAS_RECORD]->(:Record)

Optional dependency

Requires neo4j (install via pip install 'imednet[neo4j]'). The driver is imported lazily at connection time so that importing this module never fails when neo4j is not installed.

Idempotency

When SinkConfig.idempotent is True (default) the sink uses MERGE on the node’s primary key property so that re-running an export for the same batch updates existing nodes rather than creating duplicates.

Usage

from imednet.integrations.graph import Neo4jExportSink, Neo4jSinkConfig
from imednet_workflows.data_extraction import DataExtractionWorkflow

records = DataExtractionWorkflow(sdk).extract_records_by_criteria(
    study_key="MYSTUDY",
)

config = Neo4jSinkConfig(batch_size=200)
with Neo4jExportSink(
    uri="bolt://localhost:7687",
    auth=("neo4j", "password"),
    config=config,
) as sink:
    for i, batch in enumerate(batched(records, config.batch_size)):
        sink.write_batch(batch, batch_id=f"MYSTUDY/all/{i}")
class imednet_sinks.graph.Neo4jExportSink[source]

Bases: ExportSink

Export iMednet records as Neo4j graph nodes and relationships.

Parameters

config:

Mandatory Neo4jSinkConfig.

raises ~imednet.errors.ExportConfigurationError:

When the driver cannot connect to the database.

raises ImportError:

When the neo4j package is not installed.

__init__(config)[source]

Initialize the Neo4j export sink.

Parameters:

config (Neo4jSinkConfig) – Mandatory Neo4j-specific sink configuration.

Return type:

None

close()[source]

Close the underlying Neo4j driver connection.

Return type:

None

flush()[source]

No-op: Neo4j writes are committed per transaction.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to Neo4j using MERGE (idempotent) or CREATE.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.graph.Neo4jSinkConfig[source]

Bases: SinkConfig

Extended SinkConfig for Neo4j.

Parameters

database:

Target Neo4j database name (default "neo4j").

__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, uri='', auth=('', ''), database='neo4j')
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) –

  • uri (str) –

  • auth (tuple[str, str]) –

  • database (str) –

Return type:

None

imednet_sinks.graph.export_to_neo4j(sdk, study_key, uri='', auth=('', ''), *, config=None)[source]

Export study records to Neo4j using Neo4jExportSink.

Return type:

int

Parameters:

imednet_sinks.plugin module

Entry point for the iMednet sinks plugin.

This module provides the SinksNamespace which aggregates all available data export sinks (MongoDB, Neo4j, Snowflake) under a single namespace on the SDK instance.

class imednet_sinks.plugin.SinksNamespace[source]

Bases: object

Namespace for accessing data export sinks.

This class is typically instantiated by the SDK and exposed via sdk.sinks. It provides properties to access sink-specific export functions and class definitions.

property MongoDbExportSink: Any

Sink class for exporting records to MongoDB.

property MongoDbSinkConfig: Any

Configuration class for MongoDB sinks.

property Neo4jExportSink: Any

Sink class for exporting records to Neo4j.

property Neo4jSinkConfig: Any

Configuration class for Neo4j sinks.

property SnowflakeExportSink: Any

Sink class for exporting records to Snowflake.

property SnowflakeSinkConfig: Any

Configuration class for Snowflake sinks.

__init__(sdk)[source]

Initialize the sinks namespace.

Parameters:

sdk (Union[ImednetFacade, AsyncImednetFacade]) – An instance of the iMednet SDK (synchronous or asynchronous).

property export_to_mongodb: Any

Helper function to export records to MongoDB.

property export_to_neo4j: Any

Helper function to export records to Neo4j.

property export_to_snowflake: Any

Helper function to export records to Snowflake.

imednet_sinks.plugin.create_sinks(sdk)[source]

Factory function to create a new SinksNamespace instance.

Return type:

SinksNamespace

Parameters:

sdk (ImednetFacade | AsyncImednetFacade) –

imednet_sinks.warehouse module

Snowflake warehouse export sink.

This module implements the warehouse export path for a Snowflake destination. Study records are:

  1. Written to Parquet files in a local staging directory (one file per batch).

  2. Uploaded to the configured Snowflake internal stage via PUT.

  3. Bulk-loaded into the target table with COPY INTO ... FROM @<stage>.

This two-phase approach decouples data preparation from bulk ingestion, allows the Parquet files to be independently audited or re-uploaded, and leverages Snowflake’s native columnar loader for best throughput.

Manifest

After each successful COPY INTO, a manifest entry is appended to SinkConfig.extra["manifest_path"] (if provided):

{
    "batch_id":   "MYSTUDY/FORM1/0",
    "stage_path": "@MY_STAGE/imednet/MYSTUDY/FORM1/batch_0.parquet",
    "row_count":  500,
    "loaded_at":  "2024-01-15T12:00:00Z"
}

Optional dependencies

  • snowflake-connector-python (pip install 'imednet[snowflake]')

  • pyarrow (included in imednet[snowflake])

Both are imported lazily at connection / write time.

Idempotency

When SinkConfig.idempotent is True (default) the sink uses COPY INTO ... FORCE = FALSE so that Snowflake skips files that have already been loaded, making re-runs safe. Set idempotent = False to force re-ingestion of previously loaded files.

Usage

from imednet.integrations.warehouse import SnowflakeExportSink, SnowflakeSinkConfig

config = SnowflakeSinkConfig(
    account="myorg-myaccount",
    user="loader",
    **{"password": os.environ["SF_PASS"]},  # keep credentials out of source code
    database="IMEDNET_DB",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
    stage="MY_STAGE",
    table="RECORDS",
    stage_prefix="imednet",
    local_staging_dir="/tmp/imednet_stage",
)
with SnowflakeExportSink(config=config) as sink:
    for i, batch in enumerate(batched(records, config.batch_size)):
        sink.write_batch(batch, batch_id=f"MYSTUDY/FORM1/{i}")
class imednet_sinks.warehouse.SnowflakeExportSink[source]

Bases: ExportSink

Stage Parquet files and bulk-load them into Snowflake.

Parameters

config:

SnowflakeSinkConfig containing all connection details and staging paths.

raises ~imednet.errors.ExportConfigurationError:

When the Snowflake connector cannot be initialised or the required configuration values are missing.

raises ImportError:

When snowflake-connector-python or pyarrow are not installed.

__init__(config)[source]

Initialize the Snowflake export sink.

Parameters:

config (SnowflakeSinkConfig) – Mandatory Snowflake-specific sink configuration.

Return type:

None

close()[source]

Close the Snowflake connection and clean up temporary staging files.

Return type:

None

flush()[source]

No-op: each batch is committed individually.

Return type:

None

write_batch(records, *, batch_id)[source]

Write records to Snowflake via Parquet staging + COPY INTO.

Return type:

int

Parameters:

  • records (Sequence[Any]) –

  • batch_id (str) –

class imednet_sinks.warehouse.SnowflakeSinkConfig[source]

Bases: SinkConfig

Configuration for SnowflakeExportSink.

Parameters

account:

Snowflake account identifier (<org>-<account> or legacy format).

user:

Snowflake user name.

password:

Snowflake password. Never logged.

database:

Target database.

schema:

Target schema.

warehouse:

Virtual warehouse used for the COPY INTO command.

stage:

Snowflake internal stage name (e.g. "MY_STAGE").

table:

Destination table name inside database.*schema*.

stage_prefix:

Path prefix inside the stage (default "imednet").

local_staging_dir:

Local directory used to write Parquet files before PUT. Defaults to a temporary directory created by tempfile.

manifest_path:

Optional path to a JSON-lines file where each loaded batch is recorded.

__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, account='', user='', password='', database='', schema='PUBLIC', warehouse='', stage='', table='', stage_prefix='imednet', local_staging_dir=None, manifest_path=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) –

  • account (str) –

  • user (str) –

  • password (str) –

  • database (str) –

  • schema (str) –

  • warehouse (str) –

  • stage (str) –

  • table (str) –

  • stage_prefix (str) –

  • local_staging_dir (str | PathLike[str] | None) –

  • manifest_path (str | PathLike[str] | None) –

Return type:

None

imednet_sinks.warehouse.export_to_snowflake(sdk, study_key, *, config)[source]

Export study records to Snowflake using SnowflakeExportSink.

Return type:

int

Parameters: