imednet.spi package¶

Service Provider Interface (SPI) for Imednet Extensions.

This module provides a stable public interface for advanced platform developers, plugins, and providers, decoupling them from internal core implementation details.

Submodules¶

imednet.spi.cli module¶

Service Provider Interface (SPI) for CLI extensions.

imednet.spi.cli.parse_filter_args(filter_args)[source]¶

Parse a list of key=value strings into a dictionary.

Return type:: Optional[dict[str, Any]]
Parameters:: filter_args (list[str] | None) –

imednet.spi.cli.with_sdk(func)[source]¶

Initialize the SDK and pass it to the wrapped command function.

Return type:: Callable[[ParamSpec(P)], TypeVar(R)]
Parameters:: func (Callable[[Concatenate[ImednetSDK, ~P]], R]) –

imednet.spi.constants module¶

SPI for constants.

imednet.spi.endpoints module¶

SPI for endpoints.

class imednet.spi.endpoints.Record¶

Bases: Record

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.spi.endpoints.Variable¶

Bases: Variable

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].

imednet.spi.errors module¶

SPI for error types.

imednet.spi.facade module¶

Facade protocols for the iMednet SDK.

These protocols define the expected interface for the iMednet SDK clients, allowing for easier mocking and decoupling in dependent packages.

class imednet.spi.facade.AsyncImednetFacade[source]¶

Bases: Protocol

Protocol for the asynchronous iMednet SDK facade.

__init__(*args, **kwargs)¶

async async_create_record(study_key, records_data, email_notify=None, *, schema=None)[source]¶

Asynchronously submit a request to create one or more data records.

Parameters:

study_key (str) – The unique identifier for the study.
records_data (list[dict[str, Any]]) – List of dictionaries containing record field data.
email_notify (Union[bool, str, None]) – Optional email notification settings.
schema (Optional[Any]) – Optional validation schema.

Return type:

Job

Returns:

A Job model representing the background operation.

async async_get_codings(study_key, **filters)[source]¶

Asynchronously fetch code list definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Coding]

Returns:

A list of Coding models.

async async_get_forms(study_key, **filters)[source]¶

Asynchronously fetch form definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Form]

Returns:

A list of Form models.

async async_get_intervals(study_key, **filters)[source]¶

Asynchronously fetch visit interval definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Interval]

Returns:

A list of Interval models.

async async_get_job(study_key, batch_id)[source]¶

Asynchronously fetch the status of a specific background job.

Parameters:

study_key (str) – The unique identifier for the study.
batch_id (str) – The unique identifier for the job batch.

Return type:

JobStatus

Returns:

A JobStatus model.

async async_get_queries(study_key, **filters)[source]¶

Asynchronously fetch queries for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Query]

Returns:

A list of Query models.

async async_get_record_revisions(study_key, **filters)[source]¶

Asynchronously fetch record revision history for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[RecordRevision]

Returns:

A list of RecordRevision models.

async async_get_records(study_key, **filters)[source]¶

Asynchronously fetch data records for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Record]

Returns:

A list of Record models.

async async_get_sites(study_key, **filters)[source]¶

Asynchronously fetch site definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Site]

Returns:

A list of Site models.

async async_get_studies(**filters)[source]¶

Asynchronously fetch all studies accessible to the current user.

Parameters:: **filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.
Return type:: list[Study]
Returns:: A list of Study models.

async async_get_subjects(study_key, **filters)[source]¶

Asynchronously fetch subject records for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Subject]

Returns:

A list of Subject models.

async async_get_users(study_key, **filters)[source]¶

Asynchronously fetch user definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[User]

Returns:

A list of User models.

async async_get_variables(study_key, **filters)[source]¶

Asynchronously fetch variable definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Variable]

Returns:

A list of Variable models.

async async_get_visits(study_key, **filters)[source]¶

Asynchronously fetch visit definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Visit]

Returns:

A list of Visit models.

async async_poll_job(study_key, batch_id, timeout=300.0, interval=2.0)[source]¶

Asynchronously wait for a background job to reach a terminal state.

Parameters:

study_key (str) – The unique identifier for the study.
batch_id (str) – The unique identifier for the job batch.
timeout (float) – Maximum time to wait in seconds.
interval (float) – Time between poll attempts in seconds.

Return type:

JobStatus

Returns:

The final JobStatus of the operation.

class imednet.spi.facade.ImednetFacade[source]¶

Bases: Protocol

Protocol for the synchronous iMednet SDK facade.

__init__(*args, **kwargs)¶

property auth: Any¶: Access the authentication provider for the facade.

create_record(study_key, records_data, email_notify=None, *, schema=None)[source]¶

Submit a request to create one or more data records.

Parameters:

study_key (str) – The unique identifier for the study.
records_data (list[dict[str, Any]]) – List of dictionaries containing record field data.
email_notify (Union[bool, str, None]) – Optional email notification settings.
schema (Optional[Any]) – Optional validation schema.

Return type:

Job

Returns:

A Job model representing the background operation.

get_codings(study_key, **filters)[source]¶

Fetch code list definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Coding]

Returns:

A list of Coding models.

get_forms(study_key, **filters)[source]¶

Fetch form definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Form]

Returns:

A list of Form models.

get_intervals(study_key, **filters)[source]¶

Fetch visit interval definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Interval]

Returns:

A list of Interval models.

get_job(study_key, batch_id)[source]¶

Fetch the status of a specific background job.

Parameters:

study_key (str) – The unique identifier for the study.
batch_id (str) – The unique identifier for the job batch.

Return type:

JobStatus

Returns:

A JobStatus model.

get_queries(study_key, **filters)[source]¶

Fetch queries for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Query]

Returns:

A list of Query models.

get_record_revisions(study_key, **filters)[source]¶

Fetch record revision history for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[RecordRevision]

Returns:

A list of RecordRevision models.

get_records(study_key, **filters)[source]¶

Fetch data records for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Record]

Returns:

A list of Record models.

get_sites(study_key, **filters)[source]¶

Fetch site definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Site]

Returns:

A list of Site models.

get_studies(**filters)[source]¶

Fetch all studies accessible to the current user.

Parameters:: **filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.
Return type:: list[Study]
Returns:: A list of Study models.

get_subjects(study_key, **filters)[source]¶

Fetch subject records for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Subject]

Returns:

A list of Subject models.

get_users(study_key, **filters)[source]¶

Fetch user definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[User]

Returns:

A list of User models.

get_variables(study_key, **filters)[source]¶

Fetch variable definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Variable]

Returns:

A list of Variable models.

get_visits(study_key, **filters)[source]¶

Fetch visit definitions for a specific study.

Parameters:

study_key (str) – The unique identifier for the study.
**filters (Union[str, int, float, bool, None, tuple[str, Union[str, int, float, bool, None]], list[Union[str, int, float, bool, None]]]) – Optional API filters to apply.

Return type:

list[Visit]

Returns:

A list of Visit models.

poll_job(study_key, batch_id, timeout=300.0, interval=2.0)[source]¶

Wait for a background job to reach a terminal state.

Parameters:

study_key (str) – The unique identifier for the study.
batch_id (str) – The unique identifier for the job batch.
timeout (float) – Maximum time to wait in seconds.
interval (float) – Time between poll attempts in seconds.

Return type:

JobStatus

Returns:

The final JobStatus of the operation.

imednet.spi.models module¶

SPI for models.

imednet.spi.utils module¶

SPI for utility functions.

class imednet.spi.utils.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) –

exception imednet.spi.utils.JobFailedError[source]¶

Bases: Exception

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

__init__(message, status)[source]¶

Initialize the JobFailedError.

Parameters:

message (str) –
status (JobStatus) –

Return type:

None

class imednet.spi.utils.JobPollSummary[source]¶

Bases: object

Aggregate result of polling multiple jobs.

__init__(study_key, results, failures, elapsed_total)¶

Parameters:

study_key (str) –
results (dict[str, imednet.models.engine.JobStatus]) –
failures (dict[str, Exception]) –
elapsed_total (float) –

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.spi.utils.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.spi.utils.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.spi.utils.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.spi.utils.JobTimeoutError[source]¶

Bases: TimeoutError

Raised when a job does not finish before the timeout.

imednet.spi.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.spi.utils.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.spi.utils.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.spi.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.spi.utils.get_base_url_context()[source]¶

Get the current base URL for the active thread/task context.

Return type:: Optional[str]

imednet.spi.utils.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.spi.utils.get_study_context()[source]¶

Get the current study key for the active thread/task context.

Return type:: Optional[str]

imednet.spi.utils.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.spi.utils.is_missing_value(value)[source]¶

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

Return type:: bool
Parameters:: value (Any) –

imednet.spi.utils.mask_clinical_phi(value)[source]¶

Recursively mask sensitive keys in unstructured data.

Return type:: Any
Parameters:: value (Any) –

imednet.spi.utils.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.spi.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.spi.utils.redact_sensitive_payload(data)[source]¶

Return data with sensitive key values redacted.

Return type:: Any
Parameters:: data (Any) –

imednet.spi.utils.redact_sensitive_text(text)[source]¶

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

Return type:: str
Parameters:: text (Any) –

imednet.spi.utils.reset_base_url_context(token)[source]¶

Reset the active base URL using a token from set_base_url_context().

Return type:: None
Parameters:: token (Token[str | None]) –

imednet.spi.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.spi.utils.set_base_url_context(base_url)[source]¶

Set the active base URL for the current thread/task context.

Return type:: Token[Optional[str]]
Parameters:: base_url (str) –

imednet.spi.utils.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.spi.validation module¶

SPI for validation utilities.

class imednet.spi.validation.AsyncSchemaValidator[source]¶

Bases: BaseSchemaValidator[AsyncImednetFacade]

Validate record payloads using variable metadata from the API (Asynchronous).

__init__(sdk)[source]¶

Initialize the asynchronous schema validator.

Parameters:: sdk (AsyncImednetFacade) –
Return type:: None

async refresh(study_key)[source]¶

Populate the schema cache for study_key from the Variables endpoint.

This method never raises ValidationError; any API errors bubble up as ApiError.

Return type:: None
Parameters:: study_key (str) –

async validate_batch(study_key, records)[source]¶

Validate a batch of record payloads asynchronously.

Return type:

None

Parameters:

study_key (str) –
records (list[dict[str, Any]]) –

async validate_record(study_key, record)[source]¶

Validate a single record payload asynchronously.

Return type:

None

Parameters:

study_key (str) –
record (dict[str, Any]) –

class imednet.spi.validation.DataDictionary[source]¶

Bases: object

Container for data dictionary CSV content.

__init__(business_logic, choices, forms, questions)¶

Parameters:

business_logic (list[dict[str, str]]) –
choices (list[dict[str, str]]) –
forms (list[dict[str, str]]) –
questions (list[dict[str, str]]) –

Return type:

None

class imednet.spi.validation.DataDictionaryLoader[source]¶

Bases: object

Load data dictionary files from various sources.

classmethod from_directory(directory)[source]¶

Load all required CSV files from directory.

Return type:: DataDictionary
Parameters:: directory (Path | str) –

classmethod from_files(*, business_logic, choices, forms, questions)[source]¶

Load a data dictionary from individual CSV files.

Return type:

DataDictionary

Parameters:

business_logic (Path | TextIO) –
choices (Path | TextIO) –
forms (Path | TextIO) –
questions (Path | TextIO) –

classmethod from_zip(source)[source]¶

Load a data dictionary from a ZIP archive containing the required CSVs.

Return type:: DataDictionary
Parameters:: source (Path | BinaryIO) –

class imednet.spi.validation.SchemaCache[source]¶

Bases: BaseSchemaCache[ImednetFacade]

Synchronous implementation of the schema cache.

__init__()[source]¶

Initialize the synchronous schema cache.

Return type:: None

class imednet.spi.validation.SchemaValidator[source]¶

Bases: BaseSchemaValidator[ImednetFacade]

Validate record payloads using variable metadata from the API (Synchronous).

__init__(sdk, *, is_async=False)[source]¶

Initialize the synchronous schema validator.

Parameters:

sdk (ImednetFacade) –
is_async (bool) –

Return type:

None

static __new__(cls, sdk, *args, **kwargs)[source]¶

Create a new instance of the validator, handling deprecation of is_async.

Parameters:

sdk (ImednetFacade) – The iMednet facade instance.
*args (Any) – Positional arguments.
**kwargs (Any) – Keyword arguments.

Returns:

A SchemaValidator or AsyncSchemaValidator instance.

Return type:

Any

refresh(study_key)[source]¶

Populate the schema cache for study_key from the Variables endpoint.

This method never raises ValidationError; any API errors bubble up as ApiError.

Return type:: None
Parameters:: study_key (str) –

validate_batch(study_key, records)[source]¶

Validate a batch of record payloads.

Return type:

None

Parameters:

study_key (str) –
records (list[dict[str, Any]]) –

validate_record(study_key, record)[source]¶

Validate a single record payload.

Return type:

None

Parameters:

study_key (str) –
record (dict[str, Any]) –