imednet.utils package

Utils init module.

imednet.utils.JsonDict

alias of dict[str, Any]

imednet.utils.build_filter_string(filters, and_connector=';', or_connector=',')[source]

Return a filter string constructed according to iMednet rules.

Each key in filters is converted to camelCase. Raw values imply equality, tuples allow explicit operators, and lists generate multiple equality filters joined by or_connector. Conditions are then joined by and_connector.

Example: >>> from imednet.utils.filters import build_filter_string >>> build_filter_string({‘age’: (‘>’, 30), ‘status’: ‘active’}) ‘age>30;status==active’ >>> build_filter_string({‘type’: [‘A’, ‘B’]}) ‘type==A,type==B’

Return type:

str

Parameters:

  • filters (dict[str, Union[Any, tuple[str, Any], list[Any]]]) –

  • and_connector (str) –

  • or_connector (str) –

imednet.utils.configure_json_logging(level=20)[source]

Configure root logger to emit JSON formatted logs.

Return type:

None

Parameters:

level (int) –

imednet.utils.format_iso_datetime(dt)[source]

Format a datetime object into an ISO 8601 string ending with ‘Z’.

Parameters:

dt (datetime) – datetime object (naive or timezone-aware).

Return type:

str

Returns:

A string representing the datetime in ISO 8601 format.

imednet.utils.parse_iso_datetime(date_str)[source]

Parse an ISO 8601 date/time string into a datetime object.

Handles timestamps ending with ‘Z’ as UTC.

Parameters:

date_str (str) – ISO 8601 formatted date/time string.

Return type:

datetime

Returns:

A timezone-aware datetime object.

Raises:

ValueError – If the input string is not a valid ISO format.

imednet.utils.sanitize_base_url(url)[source]

Return base URL without trailing slashes or /api suffix.

Return type:

str

Parameters:

url (str) –

imednet.utils.sanitize_csv_formula(value)[source]

Sanitize a value to prevent CSV/Excel Formula Injection.

Prefixes strings starting with =, +, -, or @ (even after whitespace) with a single quote. Lists and tuples are recursively sanitized.

Return type:

Any

Parameters:

value (Any) –

imednet.utils.validate_partition_key(key)[source]

Reject partition keys that could escape or reshape the target directory.

Return type:

None

Parameters:

key (str) –

Submodules

imednet.utils.arrow module

PyArrow serialization helpers.

imednet.utils.arrow.to_arrow_table(data_records, schema=None)[source]

Serialize record dictionaries (or Pydantic-like objects) into a pyarrow.Table.

Parameters:

  • data_records (list[dict[str, Any] | _ModelDumpable]) – Record payloads to serialize. Each item must be a dictionary or expose a model_dump() method that returns a dictionary.

  • schema (Optional[Schema]) – Optional explicit Arrow schema. When provided, output columns follow schema order and types; when omitted, columns and types are inferred. Naive datetime values are interpreted as UTC. When schema inference is used, datetime columns use microsecond precision. Boolean strings accept true/false, 1/0, yes/no, y/n, and t/f.

Return type:

Table

Returns:

A fully initialized pyarrow.Table with deterministic columns and null values for missing or empty-string inputs.

Raises:

  • ImportError – If pyarrow is not installed.

  • TypeError – If a record is not dict-like and does not expose model_dump.

imednet.utils.dates module

Utility functions for parsing and formatting ISO date/time strings.

imednet.utils.dates.format_iso_datetime(dt)[source]

Format a datetime object into an ISO 8601 string ending with ‘Z’.

Parameters:

dt (datetime) – datetime object (naive or timezone-aware).

Return type:

str

Returns:

A string representing the datetime in ISO 8601 format.

imednet.utils.dates.parse_iso_datetime(date_str)[source]

Parse an ISO 8601 date/time string into a datetime object.

Handles timestamps ending with ‘Z’ as UTC.

Parameters:

date_str (str) – ISO 8601 formatted date/time string.

Return type:

datetime

Returns:

A timezone-aware datetime object.

Raises:

ValueError – If the input string is not a valid ISO format.

imednet.utils.db module

Centralized database connection utilities.

imednet.utils.db.get_sqlite_connection(db_path, timeout=30.0, busy_timeout_ms=30000, foreign_keys=False)[source]

Return a SQLite connection configured for concurrent access.

Return type:

Connection

Parameters:

  • db_path (str | Path) –

  • timeout (float) –

  • busy_timeout_ms (int) –

  • foreign_keys (bool) –

imednet.utils.db.sqlite_connection(db_path, timeout=30.0, busy_timeout_ms=30000, foreign_keys=False)[source]

Context manager yielding a safely configured SQLite connection.

Return type:

Iterator[Connection]

Parameters:

  • db_path (str | Path) –

  • timeout (float) –

  • busy_timeout_ms (int) –

  • foreign_keys (bool) –

imednet.utils.filters module

Utility functions for building API filter strings.

This module provides functionality to construct filter query parameters for iMednet API endpoints based on the reference documentation.

imednet.utils.filters.build_filter_string(filters, and_connector=';', or_connector=',')[source]

Return a filter string constructed according to iMednet rules.

Each key in filters is converted to camelCase. Raw values imply equality, tuples allow explicit operators, and lists generate multiple equality filters joined by or_connector. Conditions are then joined by and_connector.

Example: >>> from imednet.utils.filters import build_filter_string >>> build_filter_string({‘age’: (‘>’, 30), ‘status’: ‘active’}) ‘age>30;status==active’ >>> build_filter_string({‘type’: [‘A’, ‘B’]}) ‘type==A,type==B’

Return type:

str

Parameters:

  • filters (dict[str, Union[Any, tuple[str, Any], list[Any]]]) –

  • and_connector (str) –

  • or_connector (str) –

imednet.utils.job_poller module

Utility for polling job status.

class imednet.utils.job_poller.AsyncJobPoller[source]

Bases: BaseJobPoller

Asynchronously poll a job until completion.

__init__(get_job, fetch_result=None)[source]

Initialize the asynchronous job poller.

Parameters:

  • get_job (Callable[[str, str], Awaitable[JobStatus]]) – Awaitable callable that takes study_key and batch_id and returns JobStatus.

  • fetch_result (Optional[Callable[[str], Awaitable[Any]]]) – Optional awaitable callable to fetch the result data.

Return type:

None

async async_poll_many(study_key, batch_ids, interval=5.0, timeout=300.0, *, on_progress=None, fail_fast=False)[source]

Asynchronously poll multiple jobs concurrently via asyncio.gather.

All jobs are polled in parallel. Results are collected regardless of individual failures unless fail_fast=True.

Return type:

JobPollSummary

Parameters:

  • study_key (str) –

  • batch_ids (list[str]) –

  • interval (float) –

  • timeout (float) –

  • on_progress (JobProgressCallback | None) –

  • fail_fast (bool) –

async run(study_key, batch_id, interval=5.0, timeout=300.0, *, on_progress=None)[source]

Asynchronously poll a job until completion.

Return type:

JobStatus

Parameters:

  • study_key (str) –

  • batch_id (str) –

  • interval (float) –

  • timeout (float) –

  • on_progress (JobProgressCallback | None) –

class imednet.utils.job_poller.BaseJobPoller[source]

Bases: object

Base class for polling a job until it reaches a terminal state.

exception imednet.utils.job_poller.JobFailedError[source]

Bases: Exception

Raised when a job completes with a FAILED or CANCELLED state.

__init__(message, status)[source]

Initialize the JobFailedError.

Parameters:
Return type:

None

class imednet.utils.job_poller.JobPollSummary[source]

Bases: object

Aggregate result of polling multiple jobs.

__init__(study_key, results, failures, elapsed_total)
Parameters:
Return type:

None

property all_successful: bool

Return True if all jobs were successful.

property any_failed: bool

Return True if any job failed or timed out.

class imednet.utils.job_poller.JobPoller[source]

Bases: BaseJobPoller

Synchronously poll a job until completion.

__init__(get_job, fetch_result=None)[source]

Initialize the synchronous job poller.

Parameters:

  • get_job (Callable[[str, str], JobStatus]) – Callable that takes study_key and batch_id and returns JobStatus.

  • fetch_result (Optional[Callable[[str], Any]]) – Optional callable to fetch the result data from the result URL.

Return type:

None

poll_many(study_key, batch_ids, interval=5.0, timeout=300.0, *, on_progress=None, fail_fast=False)[source]

Poll multiple jobs concurrently using threading.

Return type:

JobPollSummary

Parameters:

  • study_key (str) –

  • batch_ids (list[str]) –

  • interval (float) –

  • timeout (float) –

  • on_progress (JobProgressCallback | None) –

  • fail_fast (bool) –

Parameters

study_keystr

The study key.

batch_idsList[str]

The batch IDs to monitor.

intervalfloat

Seconds between poll cycles.

timeoutfloat

Seconds before a job is considered timed out.

fail_fastbool

If True, raise JobTimeoutError as soon as any single job times out. If False (default), continue polling remaining jobs and collect all failures into JobPollSummary.failures.

on_progressOptional[JobProgressCallback]

Called for each job on each poll cycle. Thread-safe: the callback must be safe to call from multiple threads simultaneously.

run(study_key, batch_id, interval=5.0, timeout=300.0, *, on_progress=None)[source]

Synchronously poll a job until completion.

Return type:

JobStatus

Parameters:

  • study_key (str) –

  • batch_id (str) –

  • interval (float) –

  • timeout (float) –

  • on_progress (JobProgressCallback | None) –

class imednet.utils.job_poller.JobProgressCallback[source]

Bases: Protocol

Callable invoked on each poll cycle with the current job status.

Parameters

batch_idstr

The batch ID being polled.

statusJobStatus

The current status object from the jobs endpoint.

elapsedfloat

Seconds elapsed since polling started for this job.

__init__(*args, **kwargs)
class imednet.utils.job_poller.JobStatusEvent[source]

Bases: object

Structured event for job status updates.

__init__(batch_id, status, poll_number, elapsed, is_terminal)
Parameters:

  • batch_id (str) –

  • status (JobStatus) –

  • poll_number (int) –

  • elapsed (float) –

  • is_terminal (bool) –

Return type:

None

exception imednet.utils.job_poller.JobTimeoutError[source]

Bases: TimeoutError

Raised when a job does not finish before the timeout.

imednet.utils.job_poller.evaluate_job_state(job)[source]

Evaluate a job’s status.

Return type:

bool

Returns:

True if the job completed successfully. False if the job is still in progress.

Raises:

JobFailedError – If the job ended in a failed or cancelled state.

Parameters:

job (JobStatus) –

imednet.utils.json_logging module

JSON logging configuration utility.

imednet.utils.json_logging.configure_json_logging(level=20)[source]

Configure root logger to emit JSON formatted logs.

Return type:

None

Parameters:

level (int) –

imednet.utils.pandas module

Pandas helpers for working with iMednet models.

imednet.utils.pandas.export_records_csv(sdk, study_key, file_path, *, flatten=True)[source]

Fetch all records for study_key and write them to file_path.

Parameters are passed to records_to_dataframe() and the resulting DataFrame is written with pandas.DataFrame.to_csv() using index=False.

Return type:

None

Parameters:

  • sdk (ImednetSDK) –

  • study_key (str) –

  • file_path (str) –

  • flatten (bool) –

imednet.utils.pandas.records_to_dataframe(records, *, flatten=False)[source]

Convert a list of Record to a DataFrame.

Each record is converted using pydantic.BaseModel.model_dump() with by_alias=False. If flatten is True the record_data column is expanded using pandas.json_normalize() so that each variable becomes a column in the resulting DataFrame.

Return type:

DataFrame

Parameters:

imednet.utils.secrets module

Secret redaction utilities.

imednet.utils.secrets.redact_sensitive_payload(data)[source]

Return data with sensitive key values redacted.

Return type:

Any

Parameters:

data (Any) –

imednet.utils.security module

Security utilities.

class imednet.utils.security.SensitivityRegistry[source]

Bases: object

Registry for managing sensitive fields (PHI) that should be masked in logs.

__init__()[source]

Initialize the registry with default sensitive and exempt keys.

Return type:

None

add_exempt_key(key)[source]

Add a key to the exemption list.

Return type:

None

Parameters:

key (str) –

add_sensitive_key(key)[source]

Add a key to the sensitive list if it is not exempted.

Return type:

None

Parameters:

key (str) –

is_sensitive(key)[source]

Check if a key is considered sensitive and not exempted.

Return type:

bool

Parameters:

key (str) –

remove_exempt_key(key)[source]

Remove a key from the exemption list.

Return type:

None

Parameters:

key (str) –

remove_sensitive_key(key)[source]

Remove a key from the sensitive list.

Return type:

None

Parameters:

key (str) –

imednet.utils.security.mask_clinical_phi(value)[source]

Recursively mask sensitive keys in unstructured data.

Return type:

Any

Parameters:

value (Any) –

imednet.utils.security.sanitize_csv_formula(value)[source]

Sanitize a value to prevent CSV/Excel Formula Injection.

Prefixes strings starting with =, +, -, or @ (even after whitespace) with a single quote. Lists and tuples are recursively sanitized.

Return type:

Any

Parameters:

value (Any) –

imednet.utils.security.validate_header_value(value)[source]

Validate that a header value does not contain newline characters.

Parameters:

value (str) – The header value to validate.

Raises:

ClientError – If the value contains newline characters.

Return type:

None

imednet.utils.security.validate_partition_key(key)[source]

Reject partition keys that could escape or reshape the target directory.

Return type:

None

Parameters:

key (str) –

imednet.utils.serialization module

Data serialization and flattening utilities.

imednet.utils.serialization.flatten(data, prefix='')[source]

Recursively flatten a nested dict/list into dot-notation keys.

Return type:

dict[str, Any]

Parameters:

  • data (Any) –

  • prefix (str) –

imednet.utils.typing module

Type definitions for imednet SDK utilities.

imednet.utils.typing.FilterScalar

A single filter value accepted by endpoint list and convenience methods.

Supported forms:

  • A scalar str, int, float, or bool – implies equality. # noqa: RUF003

  • A two-element tuple (operator, scalar) – e.g. (">", 30). # noqa: RUF003

  • A list of scalars – generates OR-joined equality filters. # noqa: RUF003

alias of Optional[Union[str, int, float, bool]]

imednet.utils.typing.ItemId

Accepted types for an endpoint item-ID parameter (path segment or filter value).

alias of Union[str, int]

imednet.utils.typing.JsonDict

Generic JSON object type (mapping of string keys to arbitrary values).

imednet.utils.url module

URL manipulation and sanitization utilities.

imednet.utils.url.build_safe_path(base_path, *segments)[source]

Build a normalized relative path using HTTPX URL joining.

Parameters:

  • base_path (str) – Base path segment to start from.

  • *segments (Any) – Additional path segments. Non-string values are stringified.

Return type:

str

Returns:

A slash-normalized relative path with URL-encoded segments decoded (for example, "a%2Fb" becomes "a/b").

imednet.utils.url.redact_sensitive_text(text)[source]

Scan and redact sensitive URIs, connection strings, and query parameters in text.

Return type:

str

Parameters:

text (Any) –

imednet.utils.url.redact_url_query(url, sensitive_params=None)[source]

Return URL with sensitive query parameters redacted.

Return type:

str

Parameters:

  • url (str) –

  • sensitive_params (set[str] | None) –

imednet.utils.url.sanitize_base_url(url)[source]

Return base URL without trailing slashes or /api suffix.

Return type:

str

Parameters:

url (str) –

imednet.utils.validators module

Validation and normalization utilities for API response data.

imednet.utils.validators.is_boolean_token(value)[source]

Check if a string represents a boolean value (e.g., ‘true’, ‘yes’, ‘0’).

Return type:

bool

Parameters:

value (str) –

imednet.utils.validators.is_missing_value(value)[source]

Check if a value is None or an empty/whitespace-only string.

Return type:

bool

Parameters:

value (Any) –

imednet.utils.validators.parse_bool(v)[source]

Normalize boolean values from various representations.

Accepts bool, str, int, float and returns a bool.

Defaults to False for unknown or unparsable types (e.g. None, [], object()). String representations like ‘1.0’, ‘inf’, and ‘nan’ are treated as truthy via float fallback.

Return type:

bool

Parameters:

v (Any) –

Example

>>> from imednet.utils.validators import parse_bool
>>> parse_bool("yes")
True
>>> parse_bool("1.0")
True
>>> parse_bool(None)
False
imednet.utils.validators.parse_datetime(v)[source]

Parse an ISO datetime string, numeric timestamp, or return a sentinel value.

The SDK historically returns datetime(1969, 4, 20, 16, 20) when a timestamp field is empty. This helper mirrors that behaviour for backward compatibility.

Parameters:

v (str | int | float | datetime) – Date string, numeric timestamp (seconds since epoch), or datetime object. Numeric values are assumed to be UTC timestamps.

Return type:

datetime

imednet.utils.validators.parse_dict_or_default(v, default_factory=<class 'dict'>)[source]

Normalize dictionary values, defaulting if None. Ensures result is a dict.

Return type:

dict[str, Any]

Parameters:

  • v (Any) –

  • default_factory (Callable[[], dict[str, Any]]) –

imednet.utils.validators.parse_int_or_default(v, default=0, strict=False)[source]

Normalize integer values, defaulting if None or empty string.

If strict=True, raise ValueError on parse failure.

Return type:

int

Parameters:

  • v (Any) –

  • default (int) –

  • strict (bool) –

imednet.utils.validators.parse_list_or_default(v, default_factory=<class 'list'>)[source]

Normalize list values, defaulting if None. Ensures result is a list.

Return type:

list[TypeVar(T)]

Parameters:

  • v (Any) –

  • default_factory (Callable[[], list[~T]]) –

imednet.utils.validators.parse_str_or_default(v, default='')[source]

Normalize string values, defaulting if None.

Return type:

str

Parameters:

  • v (Any) –

  • default (str) –