imednet.core package

Re-exports core components for easier access.

exception imednet.core.ApiError[source]

Bases: ImednetError

Raised for generic API errors (non-2xx HTTP status codes).

status_code: HTTP status code returned by the API.

response: Parsed JSON or raw text of the error response.

__init__(response, status_code=None)[source]

Parameters:

response (Dict[str, Any] | str | Any) –
status_code (int | None) –

Return type:

None

class imednet.core.AsyncClient[source]

Bases: HTTPClientBase[AsyncClient, AsyncRequestExecutor]

Asynchronous variant of Client.

async aclose()[source]

Close the underlying async HTTP client.

Return type:: None

async get(path, params=None, **kwargs)[source]

Return type:

Response

Parameters:

path (str) –
params (Dict[str, Any] | None) –
kwargs (Any) –

async post(path, json=None, **kwargs)[source]

Return type:

Response

Parameters:

path (str) –
json (Any | None) –
kwargs (Any) –

class imednet.core.AsyncPaginator[source]

Bases: BasePaginator[AsyncRequestorProtocol]

Asynchronous variant of Paginator.

exception imednet.core.AuthenticationError[source]

Bases: ApiError

Raised when authentication to the API fails (HTTP 401).

exception imednet.core.AuthorizationError[source]

Bases: ApiError

Raised when access to the API is forbidden (HTTP 403).

exception imednet.core.BadRequestError[source]

Bases: ValidationError

Raised for HTTP 400 bad requests.

class imednet.core.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.core.Client[source]

Bases: HTTPClientBase[Client, SyncRequestExecutor]

Core HTTP client for the iMednet API.

base_url: Base URL for API requests.

timeout: Default timeout for requests.

retries: Number of retry attempts for transient errors.

backoff_factor: Multiplier for exponential backoff.

close()[source]

Close the underlying HTTP client.

Return type:: None

get(path, params=None, **kwargs)[source]

Make a GET request.

Parameters:

path (str) – URL path or full URL.
params (Optional[Dict[str, Any]]) – Query parameters.
kwargs (Any) –

Return type:

Response

post(path, json=None, **kwargs)[source]

Make a POST request.

Parameters:

path (str) – URL path or full URL.
json (Optional[Any]) – JSON body for the request.
kwargs (Any) –

Return type:

Response

exception imednet.core.ConflictError[source]

Bases: ApiError

Raised for HTTP 409 conflict errors.

class imednet.core.Context[source]

Bases: object

Compatibility shim backed by thread/task-local context variables.

clear_default_study_key()[source]

Return type:: None

property default_study_key: str | None

set_default_study_key(study_key)[source]

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

class imednet.core.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.core.ForbiddenError[source]

Bases: AuthorizationError

Raised for HTTP 403 forbidden errors.

class imednet.core.HTTPClientBase[source]

Bases: BaseClient, ABC, Generic[ClientT, ExecutorT]

Shared logic for synchronous and asynchronous HTTP clients.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, tracer=None, retry_policy=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 (Any | None) –
retry_policy (RetryPolicy | None) –
auth (AuthStrategy | None) –

Return type:

None

property retry_policy: RetryPolicy

exception imednet.core.ImednetError[source]

Bases: Exception

Base exception for all iMednet SDK errors.

exception imednet.core.NotFoundError[source]

Bases: ApiError

Raised when a requested resource is not found (HTTP 404).

exception imednet.core.PaginationError[source]

Bases: ClientError

Raised when pagination metadata is malformed or inconsistent.

class imednet.core.Paginator[source]

Bases: BasePaginator[RequestorProtocol]

Iterate synchronously over paginated API results.

exception imednet.core.RateLimitError[source]

Bases: ApiError

Raised when the API rate limit is exceeded (HTTP 429).

exception imednet.core.RequestError[source]

Bases: ImednetError

Raised when a network request fails after retries.

class imednet.core.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.core.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

exception imednet.core.ServerError[source]

Bases: ApiError

Raised when the API returns a server error (HTTP 5xx).

exception imednet.core.UnauthorizedError[source]

Bases: AuthenticationError

Raised for HTTP 401 unauthorized errors.

exception imednet.core.ValidationError[source]

Bases: ApiError

Raised when a request is malformed or validation fails (HTTP 400).

Subpackages

Submodules

imednet.core.async_client module

Asynchronous HTTP client for the iMednet API.

class imednet.core.async_client.AsyncClient[source]

Bases: HTTPClientBase[AsyncClient, AsyncRequestExecutor]

Asynchronous variant of Client.

async aclose()[source]

Close the underlying async HTTP client.

Return type:: None

async get(path, params=None, **kwargs)[source]

Return type:

Response

Parameters:

path (str) –
params (Dict[str, Any] | None) –
kwargs (Any) –

async post(path, json=None, **kwargs)[source]

Return type:

Response

Parameters:

path (str) –
json (Any | None) –
kwargs (Any) –

imednet.core.base_client module

class imednet.core.base_client.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

imednet.core.client module

Core HTTP client for interacting with the iMednet REST API.

This module defines the Client class which handles: - Authentication headers (API key and security key). - Configuration of base URL, timeouts, and retry logic. - Making HTTP GET and POST requests. - Error mapping to custom errors. - Context-manager support for automatic cleanup.

class imednet.core.client.Client[source]

Bases: HTTPClientBase[Client, SyncRequestExecutor]

Core HTTP client for the iMednet API.

base_url: Base URL for API requests.

timeout: Default timeout for requests.

retries: Number of retry attempts for transient errors.

backoff_factor: Multiplier for exponential backoff.

close()[source]

Close the underlying HTTP client.

Return type:: None

get(path, params=None, **kwargs)[source]

Make a GET request.

Parameters:

path (str) – URL path or full URL.
params (Optional[Dict[str, Any]]) – Query parameters.
kwargs (Any) –

Return type:

Response

post(path, json=None, **kwargs)[source]

Make a POST request.

Parameters:

path (str) – URL path or full URL.
json (Optional[Any]) – JSON body for the request.
kwargs (Any) –

Return type:

Response

imednet.core.context module

Thread- and task-local study context helpers.

class imednet.core.context.Context[source]

Bases: object

Compatibility shim backed by thread/task-local context variables.

clear_default_study_key()[source]

Return type:: None

property default_study_key: str | None

set_default_study_key(study_key)[source]

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

imednet.core.context.clear_study_context()[source]

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

Return type:: None

imednet.core.context.get_current_study()[source]

Get the current study key or raise when no study context is set.

Return type:: str

imednet.core.context.get_study_context()[source]

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

Return type:: Optional[str]

imednet.core.context.reset_study_context(token)[source]

Reset the active study key using a token from set_study_context().

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

imednet.core.context.set_study_context(study_key)[source]

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

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

imednet.core.context.study_context(study_key)[source]

Temporarily set the active study key for the current thread/task context.

Return type:: Iterator[None]
Parameters:: study_key (str) –

imednet.core.factory module

Factory for creating HTTP clients.

This module encapsulates the creation logic for Client and AsyncClient, handling authentication, configuration, and retry policies.

class imednet.core.factory.ClientFactory[source]

Bases: object

Factory for creating API clients.

static create_async_client(config, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None)[source]

Create an asynchronous client.

Return type:

AsyncClient

Parameters:

config (Config) –
timeout (float) –
retries (int) –
backoff_factor (float) –
retry_policy (RetryPolicy | None) –

static create_client(config, timeout=30.0, retries=3, backoff_factor=1.0, retry_policy=None)[source]

Create a synchronous client.

Return type:

Client

Parameters:

config (Config) –
timeout (float) –
retries (int) –
backoff_factor (float) –
retry_policy (RetryPolicy | None) –

imednet.core.http_client_base module

class imednet.core.http_client_base.HTTPClientBase[source]

Bases: BaseClient, ABC, Generic[ClientT, ExecutorT]

Shared logic for synchronous and asynchronous HTTP clients.

__init__(api_key=None, security_key=None, base_url=None, timeout=30.0, retries=3, backoff_factor=1.0, tracer=None, retry_policy=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 (Any | None) –
retry_policy (RetryPolicy | None) –
auth (AuthStrategy | None) –

Return type:

None

property retry_policy: RetryPolicy

imednet.core.paginator module

Pagination helpers for iterating through API responses.

class imednet.core.paginator.AsyncJsonListPaginator[source]

Bases: AsyncPaginator

Asynchronous variant of JsonListPaginator.

class imednet.core.paginator.AsyncPaginator[source]

Bases: BasePaginator[AsyncRequestorProtocol]

Asynchronous variant of Paginator.

class imednet.core.paginator.BasePaginator[source]

Bases: Generic[ClientT]

Shared paginator implementation.

__init__(client, path, params=None, page_size=100, page_param='page', size_param='size', data_key='data', metadata_key='metadata')[source]

Parameters:

client (ClientT) –
path (str) –
params (Dict[str, Any] | None) –
page_size (int) –
page_param (str) –
size_param (str) –
data_key (str) –
metadata_key (str) –

Return type:

None

property cursor: int | None: The next page cursor (0-based page index), or None when exhausted.

class imednet.core.paginator.JsonListPaginator[source]

Bases: Paginator

Paginator for endpoints returning a raw list.

class imednet.core.paginator.Paginator[source]

Bases: BasePaginator[RequestorProtocol]

Iterate synchronously over paginated API results.

imednet.core.parsing module

Centralized model parsing strategy for the iMednet SDK.

This module provides a consistent approach to parsing API responses into Pydantic models, eliminating duplicated parsing logic across endpoints.

class imednet.core.parsing.ModelParser[source]

Bases: object

Stateful parser that can be configured with a specific model.

This class provides a convenient interface for repeated parsing operations with the same model type.

Example

>>> from imednet.models.studies import Study
>>> parser = ModelParser(Study)
>>> studies = [parser.parse(data) for data in api_response]

__init__(model)[source]

Initialize parser with a model type.

Parameters:: model (Type[BaseModel]) – The model class to parse data into
Return type:: None

parse(data)[source]

Parse raw data into a model instance.

Parameters:: data (Any) – Raw data (usually dict) to parse
Return type:: BaseModel
Returns:: Parsed model instance

parse_many(items)[source]

Parse a list of raw data items into model instances.

Parameters:: items (list[Any]) – List of raw data items
Return type:: list[BaseModel]
Returns:: List of parsed model instances

imednet.core.parsing.get_model_parser(model)[source]

Return the appropriate parsing function for a model.

This function implements a strategy pattern for model parsing: 1. If the model has a custom from_json classmethod, use it 2. Otherwise, fall back to Pydantic’s model_validate

Parameters:: model (Type[TypeVar(T, bound= BaseModel)]) – The model class to get a parser for
Return type:: Callable[[Any], TypeVar(T, bound= BaseModel)]
Returns:: A callable that takes raw data and returns a model instance

Example

>>> from imednet.models.studies import Study
>>> parser = get_model_parser(Study)
>>> study = parser({"study_name": "Test", "study_key": "123"})

imednet.core.protocols module

Core protocols for the HTTP client to decouple endpoints from concrete implementations.

class imednet.core.protocols.AsyncRequestorProtocol[source]

Bases: Protocol

Protocol for asynchronous HTTP clients.

__init__(*args, **kwargs)

aclose()[source]

Close the underlying client.

Return type:: Awaitable[None]

get(path, params=None, **kwargs)[source]

Make a GET request.

Return type:

Awaitable[Response]

Parameters:

path (str) –
params (Dict[str, Any] | None) –
kwargs (Any) –

post(path, json=None, **kwargs)[source]

Make a POST request.

Return type:

Awaitable[Response]

Parameters:

path (str) –
json (Any | None) –
kwargs (Any) –

property retry_policy: RetryPolicy: Get the retry policy.

class imednet.core.protocols.ClientProvider[source]

Bases: Protocol

Protocol for classes that provide access to synchronous and asynchronous clients.

__init__(*args, **kwargs)

class imednet.core.protocols.ParamProcessor[source]

Bases: Protocol

Protocol for parameter processing strategies.

__init__(*args, **kwargs)

process_filters(filters)[source]

Process filters to extract special parameters.

Parameters:

filters (Dict[str, Any]) – The dictionary of filters provided to the endpoint.

Returns:

The modified filters dictionary (for query string).
A dictionary of special parameters (to be merged into extra_params).

Return type:

A tuple containing

class imednet.core.protocols.RequestorProtocol[source]

Bases: Protocol

Protocol for synchronous HTTP clients.

__init__(*args, **kwargs)

close()[source]

Close the underlying client.

Return type:: None

get(path, params=None, **kwargs)[source]

Make a GET request.

Return type:

Response

Parameters:

path (str) –
params (Dict[str, Any] | None) –
kwargs (Any) –

post(path, json=None, **kwargs)[source]

Make a POST request.

Return type:

Response

Parameters:

path (str) –
json (Any | None) –
kwargs (Any) –

property retry_policy: RetryPolicy: Get the retry policy.

imednet.core.retry module

class imednet.core.retry.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) –

imednet.core.retry.IDEMPOTENT_METHODS: frozenset[str] = frozenset({'DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'}): HTTP methods that are safe to retry automatically. Mutating methods such as POST and PATCH are intentionally excluded because a server may have already processed the request before the connection dropped, and retrying would risk creating duplicate records or corrupting state.

class imednet.core.retry.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.core.retry.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