DSIS Python Client

Python SDK for the DSIS (DecisionSpace Integration Server) API. Handles dual-token authentication (Azure AD + DSIS) and provides a fluent query builder for OData access.

Installation

pip install dsis-client

For protobuf bulk data decoding (grids, horizons, seismic):

pip install dsis-model-sdk[protobuf]

Quick Start

import os
from dsis_client import DSISClient, DSISConfig, Environment, QueryBuilder

config = DSISConfig(
    environment=Environment.DEV,
    tenant_id=os.getenv("DSIS_TENANT_ID"),
    client_id=os.getenv("DSIS_CLIENT_ID"),
    client_secret=os.getenv("DSIS_CLIENT_SECRET"),
    access_app_id=os.getenv("DSIS_ACCESS_APP_ID"),
    dsis_username=os.getenv("DSIS_USERNAME"),
    dsis_password=os.getenv("DSIS_PASSWORD"),
    subscription_key_dsauth=os.getenv("DSIS_SUBSCRIPTION_KEY_DSAUTH"),
    subscription_key_dsdata=os.getenv("DSIS_SUBSCRIPTION_KEY_DSDATA"),
    dsis_site="qa",
)

client = DSISClient(config)

query = (
    QueryBuilder(
        model_name="OpenWorksCommonModel",
        district_id="OpenWorksCommonModel_OW_SV4TSTA-OW_SV4TSTA",
        project="SNORRE",
    )
    .schema("Well")
    .select("name", "depth", "status")
    .filter("depth gt 1000")
    .expand("wellbores")
)

for well in client.execute_query(query):
    print(well)

QueryBuilder

QueryBuilder is the primary way to query data. It uses method chaining and IS the query object (no .build() needed).

query = (
    QueryBuilder(
        model_name="OW5000",
        district_id="OpenWorks_OW_SV4TSTA_SingleSource-OW_SV4TSTA",
        project="SNORRE",
    )
    .schema("Fault")
    .select("fault_id,fault_type,fault_name")
    .filter("fault_type eq 'NORMAL'")
    .expand("interpretations")
)

for item in client.execute_query(query):
    print(item)

Type-safe casting with model classes

Pass a model class to .schema() and use cast=True to get typed results:

from dsis_model_sdk.models.common import Basin

query = (
    QueryBuilder(model_name="OpenWorksCommonModel", district_id=dist, project=prj)
    .schema(Basin)
    .select("basin_name,basin_id,native_uid")
)

for basin in client.execute_query(query, cast=True):
    print(basin.basin_name)  # IDE autocomplete works

Pagination

execute_query() automatically follows odata.nextLink across all pages. Control with max_pages:

# All pages (default)
all_items = list(client.execute_query(query))

# First page only (max 1000 items)
first_page = list(client.execute_query(query, max_pages=1))

Bulk Data (Protobuf)

Fetch binary data (horizons, log curves, seismic) with get_bulk_data() or stream large datasets with get_bulk_data_stream():

from dsis_model_sdk.models.common import HorizonData3D
from dsis_model_sdk.protobuf import decode_horizon_data

binary_data = client.get_bulk_data(
    schema=HorizonData3D,
    native_uid="46075",
    district_id=district_id,
    project=project,
)
decoded = decode_horizon_data(binary_data)

For large datasets, stream in chunks:

for chunk in client.get_bulk_data_stream(
    schema=SeismicDataSet3D,
    native_uid=seismic,
    query=query,
    chunk_size=10 * 1024 * 1024,  # 10 MB
    stream_retries=2,
):
    process(chunk)

Set stream_retries to retry transient failures while reading streamed chunks. Retries reopen the stream and assume the endpoint returns the same bytes across reconnects. The timeout value on streamed downloads applies to connection setup and waiting for the next bytes, not to the full transfer duration.

Error Handling

from dsis_client import DSISAuthenticationError, DSISAPIError, DSISConfigurationError

try:
    client = DSISClient(config)
    for item in client.execute_query(query):
        print(item)
except DSISConfigurationError as e:
    print(f"Bad config: {e}")
except DSISAuthenticationError as e:
    print(f"Auth failed: {e}")
except DSISAPIError as e:
    print(f"Request failed: {e}")

Logging

import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("dsis_client").setLevel(logging.DEBUG)