Transport

The ixmp4.transport module defines how ixmp4 routes data-layer operations to their execution target. All transports share a common abstract base class (Transport) and can either call service methods directly using a local database session (DirectTransport, AuthorizedTransport) or use a remote ixmp4 http server ():class:~ixmp4.transport.HttpxTransport).

Transport selection is handled automatically by get_transport() based on the platform’s configured DSN:

A DSN starting with http or https -> HttpxTransport
Any other DSN (sqlite://..., postgresql://...) -> DirectTransport

If a direct connection to a database fails and the platform also has an HTTP URL configured, Platform automatically falls back to HttpxTransport.

class ixmp4.transport.Transport

Bases: ABC

Abstract base class for all ixmp4 transport backends.

A transport is holds context for routing data-layer operations to their execution target, which can be either a local SQLAlchemy session (DirectTransport) or a remote ixmp4 HTTP server (HttpxTransport).

ixmp4.transport.cached_create_engine(dsn: str, **kwargs: Any) → Engine

Create and cache a SQLAlchemy engine for dsn.

The result is memoized so that repeated calls with the same DSN reuse the existing engine.

Parameters:

dsn – Database connection string (SQLAlchemy DSN format).
**kwargs – Additional keyword arguments forwarded to sqlalchemy.create_engine().

class ixmp4.transport.DirectTransport(session: Session, ping_database: bool = True, check_alembic_version: bool = True)

Bases: Transport

Transport that operates directly with a local SQLAlchemy database session.

session

The active SQLAlchemy ORM session used for all database operations.

Type:: sqlalchemy.orm.session.Session

classmethod check_dsn(dsn: str) → str: Normalise dsn to use the psycopg driver prefix where required.

classmethod from_dsn(dsn: str, *args: Any, **kwargs: Any) → DirectTransport

Create a DirectTransport from a connection-string DSN.

The DSN is first passed through resolve_dsn_env_tokens() to expand any {env:VAR} placeholders, then normalised via check_dsn() before a database engine is created.

Parameters:

dsn – Database connection string.
*args – Positional arguments forwarded to the constructor.
**kwargs – Keyword arguments forwarded to the constructor.

Returns:

A transport instance connected to the specified database.

Return type:

DirectTransport

Raises:

ProgrammingError – If the dialect prefix in dsn is not sqlite or postgresql.

get_database_url() → URL | None: Return the SQLAlchemy URL of the bound engine, or None.

close() → None: Roll back any open transaction, close the session, and dispose the engine.

check_versioning_compatiblity() → None

Raise OperationNotSupported unless the underlying database is PostgreSQL.

Versioning (row-level history tracking) relies on PostgreSQL-specific features and is not available on SQLite.

class ixmp4.transport.AuthorizedTransport(session: Session, auth_ctx: AuthorizationContext, platform: PlatformProtocol, ping_database: bool = True, check_alembic_version: bool = True)

Bases: DirectTransport

A DirectTransport decorated with authorisation context.

This transport holds the current AuthorizationContext and exposes a separate unauthorized_transport that can be used for operations that must bypass permission checks.

platform

The platform being accessed, used for permission look-ups.

Type:: toolkit.auth.context.PlatformProtocol

auth_ctx

The current user’s authorisation context.

Type:: toolkit.auth.context.AuthorizationContext

unauthorized_transport

A second DirectTransport bound to the same session that skips authorisation checks.

Type:: ixmp4.transport.DirectTransport

class ixmp4.transport.HttpxTransport(client: Client | TestClient[Litestar], settings: ClientSettings, check_root: bool = True)

Bases: Transport, ServiceClient

Transport that communicates with a remote ixmp4 server over HTTP.

http_client

The underlying HTTP client used to issue requests.

Type:: httpx.Client | litestar.testing.client.sync_client.TestClient[litestar.app.Litestar]

settings

Client-side configuration (timeouts, concurrency, retries, …).

Type:: ixmp4.conf.settings.ClientSettings

executor

Thread-pool used for concurrent data-layer calls.

Type:: concurrent.futures.thread.ThreadPoolExecutor

direct

Optional DirectTransport used for in-process test clients that need a live database session alongside the ASGI app.

Type:: ixmp4.transport.DirectTransport | None

backoff_maximum: Upper bound (seconds) on the exponential back-off delay. Default: 16.

backoff_factor: Multiplier for the exponential back-off calculation. Default: 0.5.

backoff_exp_base: Base of the exponent used in back-off. Default: 2.

check_root() → None

Verify connectivity and compatibility with the remote server.

Sends a GET / request to the server’s root endpoint, checks that the client and server ixmp4 versions match, and validates that the manager URL configured on the server matches the one used by the client (when ManagerAuth is in use).

Raises:: ImproperlyConfigured – If the manager URLs on the client and server disagree.

classmethod from_url(url: str, settings: ClientSettings | None = None, auth: Auth | None = None) → HttpxTransport

Create an HttpxTransport from a plain URL string.

Configures an httpx.Client with HTTP/2, the timeout from settings, and optionally a SelfSignedAuth handler when settings.secret_hs256 is set.

Parameters:

url – Base URL of the remote ixmp4 server (e.g. https://host/v1/myplatform/).
settings – Client configuration. Falls back to the default Settings when None.
auth – Authentication handler to attach to the client. When None and settings.secret_hs256 is set, a SelfSignedAuth is created automatically.

Returns:

A transport instance connected to url.

Return type:

HttpxTransport

classmethod from_asgi(asgi: Litestar, settings: ClientSettings, direct: DirectTransport | None = None, raise_server_exceptions: bool = True) → HttpxTransport

Create an HttpxTransport backed by an in-process ASGI app.

Used in the test suite to exercise the full HTTP stack without a real network connection. The root-check is intentionally skipped because the ASGI test client is not yet live at construction time.

Parameters:

asgi – A fully constructed Litestar application.
settings – Client configuration object.
direct – An optional DirectTransport to attach as transport.direct — useful when tests need both HTTP and direct-database access.
raise_server_exceptions – Forwarded to TestClient. When True, server-side exceptions propagate to the test.

Returns:

A transport instance connected to the ASGI app.

Return type:

HttpxTransport

request(method: str, path: str, **kwargs: Any) → Response

Issue an HTTP request and retry automatically on HTTP 429.

Retries up to settings.retries times. The delay between attempts is determined by get_retry_delay_seconds().

Parameters:

method – HTTP method string (e.g. "GET", "POST").
path – Path relative to the client’s base URL.
**kwargs – Additional keyword arguments forwarded to httpx.Client.request().

Returns:

The first non-429 response, or the last response if the retry budget is exhausted.

Return type:

httpx.Response

get_retry_delay_seconds(response: Response, attempt: int) → float

Calculate the retry delay for a rate-limited response.

Honours the Retry-After response header when present (both numeric seconds and HTTP-date formats are supported). Falls back to exponential back-off: min(backoff_maximum, backoff_factor * backoff_exp_base ** attempt).

Parameters:

response – The HTTP 429 response whose headers may contain Retry-After.
attempt – The current retry attempt, used to compute the exponential back-off.

Returns:

Seconds to wait before the next attempt (always ≥ 0).

Return type:

float