Connection#

The kappa_sdk.Connection class is used to establish and manage a connection with a KAPPA-Automate instance. It provides various methods to interact with fields, wells, KW scripts and user tasks.

Initializing a Connection#

The connection can be established in two ways depending on the authentication method:

  • Interactive OAuth Authentication (recommended for human users)

  • Client ID / Secret Authentication (recommended for automated scripts and services)

1. Interactive OAuth Authentication#

This method is intended for human users and does not require a client_id or client_secret. When instantiating the kappa_sdk.Connection class, the OAuth authentication flow happens interactively, meaning the user must log in via a pop-up authentication window.

How it works:

  • If kappa_sdk.Connection is created without client_id and client_secret, the user is redirected to an OAuth login page.

  • The user enters their credentials (or uses Single Sign-On, e.g., Microsoft).

  • After authentication, an access token is generated and stored in the session.

  • This token is automatically used for all subsequent API requests.

  • The token is only valid for 15 minutes. After this period, the system will automatically request a new token.

import kappa_sdk as ka
ka_address = "https://your-ka-instance.com"
connection = ka.Connection(ka_address, verify_ssl=False)

Note

For more details on OAuth 2.0 and authorization, refer to official OAuth 2.0 Documentation

2. Client ID / Secret Authentication#

This method is designed for automated services, scripts, or integrations that need to authenticate without user intervention. It uses a client_id and a client_secret to generate an access token (access_token).

How it works:

  • The application sends a request to the authentication server with client_id and client_secret.

  • The server returns an access_token, which is valid for 15 mins and stored in the session.

  • This token is automatically used for all subsequent API requests.

import kappa_sdk as ka

ka_address = "https://your-ka-instance.com"
client_id = "your-client-id"
client_secret = "your-client-secret"
connection = ka.Connection(ka_address, client_id=client_id, client_secret=client_secret, verify_ssl=False)

Warning

The client id and secret must be securely stored (avoid hardcoding them in scripts).

Advanced Connection Options#

When initializing a kappa_sdk.Connection several options can be configured to customize its behavior:

  • verify_ssl : Set this parameter to False if you need to disable SSL certificate verification (default is True).

  • proxies: Use this parameter if you need to configure proxies.

  • headers: Use this parameter to add custom headers to API requests.

advanced_connection = ka.Connection(
    server_address=ka_address,
    verify_ssl=False,
    proxies={"http": "http://proxy-server:8080", "https": "https://proxy-server:8080"},
    headers={"Custom-Header": "Value"})

Public Methods and Properties#

rest_api#

Returns the interface to directly access KAPPA-Automate’s REST API. For details on how to use the rest_api, see REST API.

rest_api = connection.rest_api

unit_converter#

Provides functions to convert internal units, which are SI units, to specific units and vice versa. For details on how to use the unit_converter, see Unit Converter.

unit_converter = connection.unit_converter

event_bus#

Allows subscription to real-time system events (e.g., data updates). For details on how to subscribe to events, see Event Bus.

event_bus = connection.event_bus

data_types_catalog#

Retrieves the global Data Types catalog of KAPPA-Automate. It allows you to add new data types to this catalog. For more details, see Data Types Catalog.

data_types_catalog = connection.data_types_catalog

well_properties_catalog#

Retrieves the global Well Properties catalog of KAPPA-Automate. It allows you to add new well properties to this catalog, see Well Properties Catalog.

well_properties_catalog = connection.well_properties_catalog

get_fields()#

Return the list of existing fields in your KAPPA-Automate cluster.

fields = connection.get_fields()
for field in fields:
    print(field.name)

run_kw_script()#

Executes a KW script on a given field and well. It uses SpecFlow binding to run the script, but since there is no public documentation for it, do not attempt to use it on your own.

create_field()#

Creates a new field with the specified parameters. The minimal parameters are name, country, field_timezone`and :attr:`unit_system_name.

new_field = connection.create_field("Field", ka.CountryEnum.france_fr, ka.TimezoneEnum.europe_paris, "Oil Field")

migrate_user_tasks()#

Migrates user tasks instances from an existing user task definition to a new one.

Note

The new and old definitions must have the same input parameters to be able to migrate.

You can specify the field name and/or the well name to apply this migration to a specific field and/or well.

connection.migrate_user_tasks("old_definition_name", "new_definition_name", "Field Name", "Well Name")