imednet package

class imednet.AsyncImednetSDK[source]

Bases: _BaseSDK, SDKConvenienceMixin

Async variant of ImednetSDK using the async HTTP client.

Always use this class with async with or call await sdk.aclose() explicitly when done. Using the synchronous context manager (with) or synchronous close() on this class will raise a TypeError.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None, async_client=None)[source]
Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

  • timeout (float) –

  • retries (int) –

  • backoff_factor (float) –

  • retry_policy (RetryPolicy | None) –

  • async_client (AsyncClient | None) –

Return type:

None

async aclose()[source]
Return type:

None

close()[source]

Raises TypeError — use await sdk.aclose() instead.

Return type:

None

codings: AsyncCodingsEndpoint
forms: AsyncFormsEndpoint
intervals: AsyncIntervalsEndpoint
jobs: AsyncJobsEndpoint
queries: AsyncQueriesEndpoint
record_revisions: AsyncRecordRevisionsEndpoint
records: AsyncRecordsEndpoint
property retry_policy: RetryPolicy
sites: AsyncSitesEndpoint
studies: AsyncStudiesEndpoint
subjects: AsyncSubjectsEndpoint
users: AsyncUsersEndpoint
variables: AsyncVariablesEndpoint
visits: AsyncVisitsEndpoint
workflows: WorkflowsNamespaceProtocol
class imednet.BaseClient[source]

Bases: object

Common initialization logic for HTTP clients.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, tracer=None, auth=None)[source]
Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

  • timeout (float | Timeout) –

  • retries (int) –

  • backoff_factor (float) –

  • tracer (opentelemetry.trace.Tracer | None) –

  • auth (AuthStrategy | None) –

Return type:

None

class imednet.Config[source]

Bases: object

Config(api_key: ‘str’, security_key: ‘str’, base_url: ‘Optional[str]’ = None)

__init__(api_key, security_key, base_url=None)
Parameters:

  • api_key (str) –

  • security_key (str) –

  • base_url (str | None) –

Return type:

None

api_key: str
base_url: Optional[str] = None
security_key: str
class imednet.DefaultRetryPolicy[source]

Bases: object

Retry policy with idempotency-aware retry logic.

  • Network errors (httpx.RequestError) and server errors (HTTP 500-599) are only retried for idempotent HTTP methods: GET, PUT, DELETE, HEAD, and OPTIONS.

  • Rate-limit responses (HTTP 429) are retried for all methods because the server rejected the request before processing the payload, so there is no risk of duplicate side-effects.

  • Requests with an unknown or missing method are treated as non-idempotent (fail-safe default): they are not retried on network errors or 5xx responses.

Overriding this behaviour

If you need retries on a POST endpoint that is internally deduplicated (e.g. the server uses an idempotency key), subclass RetryPolicy and pass an instance to the client:

from imednet.core.retry import RetryPolicy, RetryState, IDEMPOTENT_METHODS
import httpx

class IdempotentPostPolicy(RetryPolicy):
    def should_retry(self, state: RetryState) -> bool:
        method = (state.method or "").upper()
        if state.exception:
            return isinstance(state.exception, httpx.RequestError)
        response = state.result
        if isinstance(response, httpx.Response):
            return (
                response.status_code == 429
                or 500 <= response.status_code < 600
            )
        return False

sdk = ImednetSDK(..., retry_policy=IdempotentPostPolicy())
should_retry(state)[source]
Return type:

bool

Parameters:

state (RetryState) –

exception imednet.FilterConflictError[source]

Bases: OrchestratorError

Raised when both whitelist and blacklist are non-empty simultaneously.

The whitelist and blacklist filters are mutually exclusive operations. Providing both simultaneously creates ambiguous behavior and is rejected at validation time before any study resolution occurs.

Example:

# This raises FilterConflictError — a study key cannot be in both:
orchestrator.execute_pipeline(
    my_func,
    whitelist={"STUDY-A", "STUDY-B"},
    blacklist={"STUDY-C"},
)
__init__(whitelist, blacklist)[source]
Parameters:

  • whitelist (set[str]) –

  • blacklist (set[str]) –

Return type:

None

imednet.ImednetClient

alias of ImednetSDK

class imednet.ImednetSDK[source]

Bases: _BaseSDK, SDKConvenienceMixin

Public entry-point for library users.

Provides access to all iMednet API endpoints and maintains configuration.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None, client=None)[source]

Initialize the SDK with credentials and configuration.

Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

  • timeout (float) –

  • retries (int) –

  • backoff_factor (float) –

  • retry_policy (RetryPolicy | None) –

  • client (Client | None) –

Return type:

None

async aclose()[source]
Return type:

None

close()[source]

Close the synchronous client connection and free resources.

Return type:

None

codings: CodingsEndpoint
config: Config
forms: FormsEndpoint
intervals: IntervalsEndpoint
jobs: JobsEndpoint
queries: QueriesEndpoint
record_revisions: RecordRevisionsEndpoint
records: RecordsEndpoint
property retry_policy: RetryPolicy
sites: SitesEndpoint
studies: StudiesEndpoint
subjects: SubjectsEndpoint
users: UsersEndpoint
variables: VariablesEndpoint
visits: VisitsEndpoint
workflows: WorkflowsNamespaceProtocol
class imednet.MultiStudyOrchestrator[source]

Bases: object

Orchestrates pipeline execution across multiple active clinical trial boundaries.

The SDK instance passed at construction is treated as an immutable read-only resource — no worker thread may mutate its transport, authentication state, or connection pool during parallel execution.

Parameters:

  • sdk (Any) – A fully initialized ImednetSDK instance. The orchestrator stores a reference (not a copy) and treats it as immutable.

  • max_workers (int) – Maximum number of concurrent worker threads. Defaults to 4. Set to 1 to force sequential execution (useful for debugging).

Example:

with ImednetSDK(api_key=..., security_key=...) as sdk:
    orchestrator = MultiStudyOrchestrator(sdk, max_workers=8)
    results = orchestrator.execute_pipeline(my_pipeline_func)
__init__(sdk, max_workers=4)[source]
Parameters:

  • sdk (Any) –

  • max_workers (int) –

Return type:

None

execute_pipeline(pipeline_func, whitelist=None, blacklist=None, *args, **kwargs)[source]

Execute a pipeline function concurrently across resolved study contexts.

Return type:

dict[str, OrchestratorResult]

Parameters:

  • pipeline_func (StudyWorkerCallable[Any]) –

  • whitelist (set[str] | None) –

  • blacklist (set[str] | None) –

  • args (Any) –

  • kwargs (Any) –

property max_workers: int

Maximum number of concurrent worker threads.

resolve_active_studies(whitelist=None, blacklist=None)[source]

Query the iMednet registry and apply filtering rules.

Calls self._sdk.studies.list() to fetch the live study inventory, then applies the whitelist OR blacklist (mutually exclusive).

Parameters:

  • whitelist (Optional[set[str]]) – If provided, only studies whose studyKey is in this set are included. Mutually exclusive with blacklist.

  • blacklist (Optional[set[str]]) – If provided, studies whose studyKey is in this set are excluded. Mutually exclusive with whitelist.

Return type:

list[str]

Returns:

Ordered list of study key strings targeting this execution run.

Raises:

FilterConflictError – When both whitelist and blacklist are non-empty simultaneously.

property sdk: Any

The shared read-only SDK instance.

exception imednet.OrchestratorError[source]

Bases: ImednetError

Base exception for all orchestration-layer failures.

Raised when the MultiStudyOrchestrator encounters a structural or configuration error that prevents pipeline execution from starting.

Individual per-study runtime failures are NOT raised as exceptions — they are captured in the OrchestratorResult result matrix with status="FAILED".

class imednet.OrchestratorResult[source]

Bases: TypedDict

Normalized result entry returned per study by execute_pipeline.

data: Optional[Any]
duration_seconds: float
error: Optional[str]
status: str
exception imednet.PluginLoadError[source]

Bases: ImednetError

Raised when a plugin fails to load or does not satisfy the PluginProtocol.

class imednet.PluginProtocol[source]

Bases: Protocol

Protocol that every iMednet plugin factory must satisfy.

A conforming factory is a callable that accepts an SDK instance and returns an object implementing WorkflowsNamespaceProtocol.

Example:

from imednet.plugins import PluginProtocol
from imednet.sdk import ImednetSDK

class MyWorkflows:
    def __init__(self, sdk: ImednetSDK) -> None:
        self.data_extraction = ...
        self.query_management = ...
        self.record_mapper = ...
        self.record_update = ...
        self.subject_data = ...

def create_workflows(sdk: ImednetSDK) -> MyWorkflows:
    return MyWorkflows(sdk)

# Verify conformance at import time (optional, for development use):
assert isinstance(create_workflows, PluginProtocol)
__init__(*args, **kwargs)
class imednet.RetryPolicy[source]

Bases: Protocol

Interface to determine whether a request should be retried.

__init__(*args, **kwargs)
should_retry(state)[source]

Return True to retry the request for the given state.

Return type:

bool

Parameters:

state (RetryState) –

class imednet.RetryState[source]

Bases: object

State information passed to RetryPolicy.

__init__(attempt_number, exception=None, result=None, method=None)
Parameters:

  • attempt_number (int) –

  • exception (BaseException | None) –

  • result (Any | None) –

  • method (str | None) –

Return type:

None

attempt_number: int
exception: Optional[BaseException] = None
method: Optional[str] = None
result: Optional[Any] = None
class imednet.StudyWorkerCallable[source]

Bases: Protocol[T_Output]

Defines the explicit signature required for injected pipeline tasks.

Pipeline callables injected into MultiStudyOrchestrator.execute_pipeline must conform to this protocol.

__init__(*args, **kwargs)
class imednet.WorkflowsNamespaceProtocol[source]

Bases: Protocol

Minimal interface that a workflows namespace object must expose.

Each attribute should be a workflow class instance wired to the SDK.

__init__(*args, **kwargs)
data_extraction: Any
query_management: Any
record_mapper: Any
record_update: Any
subject_data: Any
imednet.load_config(api_key=None, security_key=None, base_url=None)[source]

Return configuration using arguments or environment variables.

Return type:

Config

Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

Subpackages

Submodules

imednet.async_sdk module

imednet.config module

class imednet.config.Config[source]

Bases: object

Config(api_key: ‘str’, security_key: ‘str’, base_url: ‘Optional[str]’ = None)

__init__(api_key, security_key, base_url=None)
Parameters:

  • api_key (str) –

  • security_key (str) –

  • base_url (str | None) –

Return type:

None

api_key: str
base_url: Optional[str] = None
security_key: str
imednet.config.load_config(api_key=None, security_key=None, base_url=None)[source]

Return configuration using arguments or environment variables.

Return type:

Config

Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

imednet.constants module

Common constants used throughout the iMednet SDK.

This module centralizes configuration constants to avoid magic numbers and improve maintainability.

imednet.constants.CONTENT_TYPE_JSON = 'application/json'

JSON content type value.

imednet.constants.DEFAULT_BACKOFF_FACTOR = 1.0

Default backoff factor for exponential retry delays.

imednet.constants.DEFAULT_BASE_URL = 'https://edc.prod.imednetapi.com'

Default base URL for the iMednet API.

imednet.constants.DEFAULT_PAGE_SIZE = 100

Default number of items to fetch per page.

imednet.constants.DEFAULT_RETRIES = 3

Default number of retry attempts for failed requests.

imednet.constants.DEFAULT_TIMEOUT = 30.0

Default timeout in seconds for HTTP requests.

imednet.constants.HEADER_ACCEPT = 'Accept'

HTTP Accept header name.

imednet.constants.HEADER_API_KEY = 'x-api-key'

iMednet API key header name.

imednet.constants.HEADER_CONTENT_TYPE = 'Content-Type'

HTTP Content-Type header name.

imednet.constants.HEADER_EMAIL_NOTIFY = 'x-email-notify'

iMednet email notification header name.

imednet.constants.HEADER_SECURITY_KEY = 'x-imn-security-key'

iMednet security key header name.

imednet.constants.LARGE_PAGE_SIZE = 500

Page size for endpoints with large metadata (forms, intervals, variables).

imednet.constants.MAX_SQLITE_COLUMNS = 2000

Maximum number of columns allowed in a SQLite table.

When exporting data to SQLite, tables exceeding this limit must be split into multiple tables.

imednet.constants.TERMINAL_JOB_STATES = frozenset({'CANCELLED', 'COMPLETED', 'FAILED'})

Job states that indicate the job has finished processing.

These states are used by the job poller to determine when to stop polling. Jobs in these states will not transition to another state.

imednet.discovery module

Runtime discovery utilities for live tests and scripts.

exception imednet.discovery.NoLiveDataError[source]

Bases: RuntimeError

Raised when required live data cannot be found.

imednet.discovery.discover_form_key(sdk, study_key)[source]

Return the first subject record form key for study_key.

Return type:

str

Parameters:
imednet.discovery.discover_interval_name(sdk, study_key)[source]

Return the first non-disabled interval name for study_key.

Return type:

str

Parameters:
imednet.discovery.discover_site_name(sdk, study_key)[source]

Return the first eligible site name for study_key.

A site is eligible when its site_enrollment_status is one of ELIGIBLE_SITE_STATUSES. When no eligible site is found the encountered statuses are logged so callers can distinguish missing data from unsupported status vocabulary.

Return type:

str

Parameters:
imednet.discovery.discover_study_key(sdk)[source]

Return the first study key available for the provided SDK.

Return type:

str

Parameters:

sdk (ImednetSDK) –

imednet.discovery.discover_subject_key(sdk, study_key)[source]

Return the first eligible subject key for study_key.

A subject is eligible when its subject_status is one of ELIGIBLE_SUBJECT_STATUSES. When no eligible subject is found the encountered statuses are logged so callers can distinguish missing data from unsupported status vocabulary.

Return type:

str

Parameters:
imednet.discovery.is_site_eligible(site_enrollment_status)[source]

Return True when site_enrollment_status is usable for live tests.

Return type:

bool

Parameters:

site_enrollment_status (str) –

imednet.discovery.is_subject_eligible(subject_status)[source]

Return True when subject_status is usable for live tests.

Return type:

bool

Parameters:

subject_status (str) –

imednet.plugins module

Plugin contracts for the iMednet SDK.

This module defines the PluginProtocol that any third-party plugin factory must satisfy in order to be discoverable via the imednet.plugins entry-point group, together with the WorkflowsNamespaceProtocol that the factory’s return value must implement.

Minimal plugin skeleton

A plugin package must:

  1. Expose a factory callable under the imednet.plugins entry-point group with the name "workflows" (or another agreed name).

  2. The callable must accept a single argument — the ImednetSDK (or AsyncImednetSDK) instance — and return an object that satisfies WorkflowsNamespaceProtocol.

Example pyproject.toml snippet:

[tool.poetry.plugins."imednet.plugins"]
workflows = "myplugin.namespace:create_workflows"

Example factory:

# myplugin/namespace.py
from imednet.sdk import ImednetSDK

class MyWorkflows:
    def __init__(self, sdk: ImednetSDK) -> None:
        self.data_extraction = ...
        self.query_management = ...
        self.record_mapper = ...
        self.record_update = ...
        self.subject_data = ...

def create_workflows(sdk: ImednetSDK) -> MyWorkflows:
    return MyWorkflows(sdk)
class imednet.plugins.PluginProtocol[source]

Bases: Protocol

Protocol that every iMednet plugin factory must satisfy.

A conforming factory is a callable that accepts an SDK instance and returns an object implementing WorkflowsNamespaceProtocol.

Example:

from imednet.plugins import PluginProtocol
from imednet.sdk import ImednetSDK

class MyWorkflows:
    def __init__(self, sdk: ImednetSDK) -> None:
        self.data_extraction = ...
        self.query_management = ...
        self.record_mapper = ...
        self.record_update = ...
        self.subject_data = ...

def create_workflows(sdk: ImednetSDK) -> MyWorkflows:
    return MyWorkflows(sdk)

# Verify conformance at import time (optional, for development use):
assert isinstance(create_workflows, PluginProtocol)
__init__(*args, **kwargs)
class imednet.plugins.WorkflowsNamespaceProtocol[source]

Bases: Protocol

Minimal interface that a workflows namespace object must expose.

Each attribute should be a workflow class instance wired to the SDK.

__init__(*args, **kwargs)
data_extraction: Any
query_management: Any
record_mapper: Any
record_update: Any
subject_data: Any

imednet.sdk module

Public entry-point for the iMednet SDK.

This module provides the ImednetSDK class which: - Manages configuration and authentication - Exposes all endpoint functionality through a unified interface - Provides context management for proper resource cleanup

class imednet.sdk.AsyncImednetSDK[source]

Bases: _BaseSDK, SDKConvenienceMixin

Async variant of ImednetSDK using the async HTTP client.

Always use this class with async with or call await sdk.aclose() explicitly when done. Using the synchronous context manager (with) or synchronous close() on this class will raise a TypeError.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None, async_client=None)[source]
Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

  • timeout (float) –

  • retries (int) –

  • backoff_factor (float) –

  • retry_policy (RetryPolicy | None) –

  • async_client (AsyncClient | None) –

Return type:

None

async aclose()[source]
Return type:

None

close()[source]

Raises TypeError — use await sdk.aclose() instead.

Return type:

None

codings: AsyncCodingsEndpoint
config: Config
forms: AsyncFormsEndpoint
intervals: AsyncIntervalsEndpoint
jobs: AsyncJobsEndpoint
queries: AsyncQueriesEndpoint
record_revisions: AsyncRecordRevisionsEndpoint
records: AsyncRecordsEndpoint
property retry_policy: RetryPolicy
sites: AsyncSitesEndpoint
studies: AsyncStudiesEndpoint
subjects: AsyncSubjectsEndpoint
users: AsyncUsersEndpoint
variables: AsyncVariablesEndpoint
visits: AsyncVisitsEndpoint
workflows: WorkflowsNamespaceProtocol
class imednet.sdk.ImednetSDK[source]

Bases: _BaseSDK, SDKConvenienceMixin

Public entry-point for library users.

Provides access to all iMednet API endpoints and maintains configuration.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None, client=None)[source]

Initialize the SDK with credentials and configuration.

Parameters:

  • api_key (str | None) –

  • security_key (str | None) –

  • base_url (str | None) –

  • timeout (float) –

  • retries (int) –

  • backoff_factor (float) –

  • retry_policy (RetryPolicy | None) –

  • client (Client | None) –

Return type:

None

async aclose()[source]
Return type:

None

close()[source]

Close the synchronous client connection and free resources.

Return type:

None

codings: CodingsEndpoint
config: Config
forms: FormsEndpoint
intervals: IntervalsEndpoint
jobs: JobsEndpoint
queries: QueriesEndpoint
record_revisions: RecordRevisionsEndpoint
records: RecordsEndpoint
property retry_policy: RetryPolicy
sites: SitesEndpoint
studies: StudiesEndpoint
subjects: SubjectsEndpoint
users: UsersEndpoint
variables: VariablesEndpoint
visits: VisitsEndpoint
workflows: WorkflowsNamespaceProtocol

imednet.sdk_convenience module

Convenience mixin for the iMedNet SDK.

This module contains high-level helper methods that delegate to specific endpoints.

imednet.sdk_convenience.AsyncJobPoller(*args, **kwargs)[source]
Return type:

Any

Parameters:

  • args (Any) –

  • kwargs (Any) –

class imednet.sdk_convenience.AsyncSDKProtocol[source]

Bases: Protocol

__init__(*args, **kwargs)
codings: AsyncCodingsEndpoint
forms: AsyncFormsEndpoint
intervals: AsyncIntervalsEndpoint
jobs: AsyncJobsEndpoint
queries: AsyncQueriesEndpoint
record_revisions: AsyncRecordRevisionsEndpoint
records: AsyncRecordsEndpoint
sites: AsyncSitesEndpoint
studies: AsyncStudiesEndpoint
subjects: AsyncSubjectsEndpoint
users: AsyncUsersEndpoint
variables: AsyncVariablesEndpoint
visits: AsyncVisitsEndpoint
imednet.sdk_convenience.JobPoller(*args, **kwargs)[source]
Return type:

Any

Parameters:

  • args (Any) –

  • kwargs (Any) –

class imednet.sdk_convenience.SDKConvenienceMixin[source]

Bases: object

Mixin class providing convenience methods for the SDK.

These methods wrap endpoint calls to provide a flatter API surface.

async async_get_codings(study_key=None, **filters)[source]

Asynchronously list codings.

Return type:

List[Coding]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_forms(study_key=None, **filters)[source]

Asynchronously list forms.

Return type:

List[Form]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_intervals(study_key=None, **filters)[source]

Asynchronously list intervals.

Return type:

List[Interval]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_job(study_key, batch_id)[source]

Asynchronously get job status.

Return type:

JobStatus

Parameters:
async async_get_queries(study_key=None, **filters)[source]

Asynchronously list queries.

Return type:

List[Query]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_record_revisions(study_key=None, **filters)[source]

Asynchronously list record revisions.

Return type:

List[RecordRevision]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_records(study_key=None, **filters)[source]

Asynchronously list records.

Return type:

List[Record]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_sites(study_key=None, **filters)[source]

Asynchronously list sites.

Return type:

List[Site]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_studies(**filters)[source]

Asynchronously list studies.

Return type:

List[Study]

Parameters:

  • self (AsyncSDKProtocol) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_subjects(study_key=None, **filters)[source]

Asynchronously list subjects.

Return type:

List[Subject]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_users(study_key=None, **filters)[source]

Asynchronously list users.

Return type:

List[User]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_variables(study_key=None, **filters)[source]

Asynchronously list variables.

Return type:

List[Variable]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_get_visits(study_key=None, **filters)[source]

Asynchronously list visits.

Return type:

List[Visit]

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

async async_poll_job(study_key, batch_id, *, interval=5, timeout=300)[source]

Asynchronously poll a job until it reaches a terminal state.

Return type:

JobStatus

Parameters:

  • self (AsyncSDKProtocol) –

  • study_key (str) –

  • batch_id (str) –

  • interval (int) –

  • timeout (int) –

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

Create records in the specified study.

Return type:

Job

Parameters:

  • self (SDKProtocol) –

  • study_key (str) –

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

  • email_notify (bool | str | None) –

get_codings(study_key=None, **filters)[source]

List codings.

Return type:

List[Coding]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_forms(study_key=None, **filters)[source]

List forms.

Return type:

List[Form]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_intervals(study_key=None, **filters)[source]

List intervals.

Return type:

List[Interval]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_job(study_key, batch_id)[source]

Get job status.

Return type:

JobStatus

Parameters:

  • self (SDKProtocol) –

  • study_key (str) –

  • batch_id (str) –

get_queries(study_key=None, **filters)[source]

List queries.

Return type:

List[Query]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_record_revisions(study_key=None, **filters)[source]

List record revisions.

Return type:

List[RecordRevision]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_records(study_key=None, **filters)[source]

List records.

Return type:

List[Record]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_sites(study_key=None, **filters)[source]

List sites.

Return type:

List[Site]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_studies(**filters)[source]

List studies.

Return type:

List[Study]

Parameters:

  • self (SDKProtocol) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_subjects(study_key=None, **filters)[source]

List subjects.

Return type:

List[Subject]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_users(study_key=None, **filters)[source]

List users.

Return type:

List[User]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_variables(study_key=None, **filters)[source]

List variables.

Return type:

List[Variable]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

get_visits(study_key=None, **filters)[source]

List visits.

Return type:

List[Visit]

Parameters:

  • self (SDKProtocol) –

  • study_key (str | None) –

  • filters (str | int | float | bool | None | Tuple[str, str | int | float | bool | None] | List[str | int | float | bool | None]) –

poll_job(study_key, batch_id, *, interval=5, timeout=300)[source]

Poll a job until it reaches a terminal state.

Return type:

JobStatus

Parameters:

  • self (SDKProtocol) –

  • study_key (str) –

  • batch_id (str) –

  • interval (int) –

  • timeout (int) –

class imednet.sdk_convenience.SDKProtocol[source]

Bases: Protocol

Protocol defining the interface required by SDKConvenienceMixin.

__init__(*args, **kwargs)
codings: CodingsEndpoint
forms: FormsEndpoint
intervals: IntervalsEndpoint
jobs: JobsEndpoint
queries: QueriesEndpoint
record_revisions: RecordRevisionsEndpoint
records: RecordsEndpoint
sites: SitesEndpoint
studies: StudiesEndpoint
subjects: SubjectsEndpoint
users: UsersEndpoint
variables: VariablesEndpoint
visits: VisitsEndpoint