SDK Reference

Client

Low-level access to the Komodo Platform HTTP APIs (IAM, etc.). get_snowflake_connection() constructs a Client internally; use Client directly when you need client.iam or other generated API wrappers.

Import

from komodo import Client

Constructor

def __init__(
    self,
    account_id: str | None = None,
    auth_session: Session | None = None,
    **kwargs,
) -> None

• account_id — Komodo account UUID. If omitted, taken from the session / credentials when available.
• auth_session — Optional Komodo Session (komodo.auth.Session). If omitted, a default Session() is created and connect() is called.

Notable members

Member	Description
auth	Active Session (tokens, refresh).
iam	Generated IAM API client (IamApi).
x_account_id	Resolved account ID for API calls.
komodo_api_client()	Underlying OpenAPI ApiClient.

Async context manager

classmethod async def create(...) -> Client
async def close()
async def __aenter__ / __aexit__

Use async with await Client.create(...) as client: when you need explicit cleanup of the HTTP session.

---

get_snowflake_connection

Returns a DB-API 2.0 Connection to your account’s Snowflake warehouse through the Marmot Development Kit, with query tags applied for observability.

Import

from komodo import get_snowflake_connection

Signature

def get_snowflake_connection(
    account_id: str | None = None,
    jwt: str | None = None,
    client_id: str | None = None,
    client_secret: str | None = None,
    profile: str | None = None,
    use_main_account: bool = False,
    app_id: str | None = None,
    query_tag_extra: Mapping[str, Any] | None = None,
) -> Connection

Parameters

Name	Type	Description
account_id	str | None	Komodo account UUID. Required unless the session / profile supplies it. Cannot be combined with profile if the profile already defines an account.
jwt	str | None	Bearer token for user-style auth. Cannot be combined with client_id / client_secret.
client_id	str | None	Service principal client ID (M2M). Must be used with client_secret and account_id (unless profile supplies them).
client_secret	str | None	Service principal secret (M2M).
profile	str | None	Named section in the credentials file (M2M + account). Mutually exclusive with passing jwt or split client_id/client_secret in the same call.
use_main_account	bool	Route to the shared main Snowflake account instead of the caller’s tenant. Cannot be combined with account_id.
app_id	str | None	Overrides the app id reported to the gateway. Defaults to $KOMODO_APP_ID or kdk.
query_tag_extra	Mapping[str, Any] | None	Optional caller metadata shallow-merged into the structured Snowflake QUERY_TAG under client.properties (e.g. a request or job id, for attribution in query history). Keys and values must be JSON-serializable. Must not contain secrets or PHI — the tag is stored in Snowflake’s query history.

Returns

Connection — DB-API 2.0 connection (Komodo Snowflake proxy). Use conn.cursor() to obtain a KomodoSnowflakeCursor.

Raises

Exception	When
ValueError	Invalid combination of arguments (e.g. JWT + client secret, or profile with explicit creds), or a non-JSON-serializable query_tag_extra.
UnsetAccountError	No account_id and none could be resolved from the session / file.

Example

from komodo import get_snowflake_connection

conn = get_snowflake_connection()
cur = conn.cursor()
cur.execute("SELECT 1")
print(cur.fetchone())
conn.close()

Attaching query-tag metadata

# Tag every query on this connection with caller context for attribution
# in Snowflake query history. Values must be JSON-serializable; no secrets/PHI.
conn = get_snowflake_connection(query_tag_extra={"request_id": "abc-123", "job": "nightly-refresh"})

---

execute_query_async

Async helper: execute_async on the cursor, poll until the query finishes, then attach results to the cursor and return that cursor.

Import

from komodo import execute_query_async

Signature

async def execute_query_async(
    cursor: KomodoSnowflakeCursor,
    query: str,
    query_params: list[str] | dict[str, str] | None = None,
) -> KomodoSnowflakeCursor

Parameters

Name	Description
cursor	Open cursor from get_snowflake_connection().cursor().
query	SQL string.
query_params	Optional bind parameters for the query.

Returns

The same cursor after results are loaded — call fetchall(), fetch_pandas_all(), etc.

Example

import asyncio
from komodo import get_snowflake_connection, execute_query_async

async def main():
    conn = get_snowflake_connection()
    cur = conn.cursor()
    cur = await execute_query_async(cur, "SELECT CURRENT_TIMESTAMP()")
    print(cur.fetchone())
    conn.close()

asyncio.run(main())

---

get_query_diagnostics

Fetch diagnostic detail for a previously executed query so you can self-diagnose failures: the error code/message, execution status, warehouse, and timing Snowflake recorded.

Import

from komodo import get_query_diagnostics

Signature

def get_query_diagnostics(
    connection: Connection,
    query_id: str,
) -> dict[str, Any] | None

Parameters

Name	Type	Description
connection	Connection	An open connection from get_snowflake_connection().
query_id	str	The Snowflake query id (cursor.sfqid, also shown as Query ID: in surfaced errors).

Returns

A dict of the recorded fields (QUERY_ID, QUERY_TEXT, EXECUTION_STATUS, ERROR_CODE, ERROR_MESSAGE, WAREHOUSE_NAME, START_TIME, END_TIME, TOTAL_ELAPSED_TIME), or None if no matching query is found.

Notes

Backed by Snowflake’s INFORMATION_SCHEMA.QUERY_HISTORY table function, scoped to the connection’s role and account. A current database must be set (e.g. USE DATABASE DATA) for INFORMATION_SCHEMA to resolve, and the query must be within the history retention window.

Example

from komodo import get_snowflake_connection, get_query_diagnostics

conn = get_snowflake_connection()
cur = conn.cursor()
cur.execute("USE DATABASE DATA")
try:
    cur.execute("SELECT * FROM SOME_SECURE_VIEW")
except Exception as err:
    print(err)
    print(get_query_diagnostics(conn, cur.sfqid))
conn.close()

---

is_still_running

def is_still_running(map_connection: Connection, query_id: str) -> bool

Returns whether a given Snowflake query id is still running. Used internally by execute_query_async; may raise SnowflakeException if status checks fail.

---

Snowflake-related exceptions

Name	Module	Description
UnsetAccountError	komodo.snowflake	Missing account context when opening a connection.
SnowflakeException	komodo.snowflake	Wrapped errors around query status / Map connection operations.

from komodo.snowflake import UnsetAccountError, SnowflakeException

UnsetAccountError is also available from the top-level komodo package (from komodo import UnsetAccountError).

---

OpenAPI / REST exceptions

Generated clients raise subclasses of OpenApiException (for example ApiException, ApiValueError). Import from komodo:

from komodo import ApiException, ApiValueError, OpenApiException

---

Cursor operations

After cursor = conn.cursor():

Method / attribute	Typical use
execute(sql, params=None)	Run SQL.
fetchone, fetchmany, fetchall	DB-API results.
fetch_pandas_all, fetch_pandas_batches	Pandas DataFrames (Snowflake cursor).
execute_async, get_results_from_sfqid	Advanced / used by execute_query_async.
description	Column metadata after a select.

For SQL and CLI patterns, see Executing Queries and Pandas Integration.

---

SecretsApi

Programmatic access to the Komodo Secrets Service for storing, reading, updating, and deleting secrets. Secrets can be scoped to a user, an application, or shared across the organization. Every synchronous method has an _async counterpart (e.g. create_secret_async) for use in async runtimes.

Import

from komodo.secrets_api import SecretsApi, SecretType

SecretType

Enum that determines the scope/namespace of a secret.

class SecretType(str, Enum):
    USER = "user"
    APPLICATION = "application"
    SHARED = "shared"

Value	Description
SecretType.USER	Personal secrets tied to the authenticated user
SecretType.APPLICATION	Application-scoped secrets (requires app_id)
SecretType.SHARED	Organization-wide secrets accessible by all account members

Class Signature

class SecretsApi:
    def __init__(
        self,
        api_client: Optional[ApiClient] = None,
        session: Optional[Session] = None,
        environment: Optional[str] = None,
        access_token: Optional[str] = None,
        account_id: Optional[str] = None,
    )

Parameter	Type	Description
api_client	ApiClient	Komodo API client instance. Uses the default client if omitted.
session	Session	Komodo auth session. Creates a new session if omitted.
environment	str	"production". Defaults to KOMODO_ENVIRONMENT.
access_token	str	Override token (useful in MCP/service contexts).
account_id	str	Override account ID.

Methods

create_secret / create_secret_async

Store a new secret.

def create_secret(
    self,
    secret_type: SecretType,
    secret_path: str,
    data: Dict[str, Any],
    metadata: Optional[Dict[str, str]] = None,
    app_id: Optional[str] = None,
    platform: bool = False,
) -> Union[SecretResponse, PlatformSecretResponse]

Parameter	Type	Description
secret_type	SecretType	Scope of the secret
secret_path	str	Path for the secret (e.g. "my-api-key")
data	dict	Key-value pairs to store
metadata	dict	Optional metadata
app_id	str	Required when secret_type is APPLICATION
platform	bool	Use the platform secrets endpoint (advanced; default False)

read_secret / read_secret_async

Read a secret’s value and metadata.

def read_secret(
    self,
    secret_type: SecretType,
    secret_path: str,
    app_id: Optional[str] = None,
    platform: bool = False,
) -> Union[SecretResponse, PlatformSecretResponse]

The returned SecretResponse has .data (dict), .metadata (dict or None), .success (bool), and .message (str).

list_secrets / list_secrets_async

List secret paths within a scope.

def list_secrets(
    self,
    secret_type: SecretType,
    prefix: Optional[str] = None,
    app_id: Optional[str] = None,
    platform: bool = False,
    next_token: Optional[str] = None,
) -> Union[SecretListResponse, PlatformSecretListResponse]

Parameter	Type	Description
prefix	str	Filter to paths starting with this prefix
next_token	str	Pagination token from a previous response

The returned SecretListResponse has .secrets (list), .has_more (bool), and .next_token (str or None).

update_secret / update_secret_async

Replace a secret’s data. Creates a new version.

def update_secret(
    self,
    secret_type: SecretType,
    secret_path: str,
    data: Dict[str, Any],
    metadata: Optional[Dict[str, str]] = None,
    app_id: Optional[str] = None,
    platform: bool = False,
) -> Union[SecretResponse, PlatformSecretResponse]

delete_secret / delete_secret_async

Permanently remove a secret.

def delete_secret(
    self,
    secret_type: SecretType,
    secret_path: str,
    app_id: Optional[str] = None,
    platform: bool = False,
) -> Dict[str, Any]

move_secret / move_secret_async

Copy a secret from one scope to another, optionally deleting the original.

def move_secret(
    self,
    source_type: SecretType,
    source_path: str,
    dest_type: SecretType,
    dest_path: str,
    source_app_id: Optional[str] = None,
    dest_app_id: Optional[str] = None,
    delete_original: bool = False,
) -> SecretResponse

Example

from komodo.secrets_api import SecretsApi, SecretType

api = SecretsApi(environment="production")

# Create a user secret
api.create_secret(
    secret_type=SecretType.USER,
    secret_path="my-api-key",
    data={"key": "abc123"},
)

# List all user secrets
result = api.list_secrets(SecretType.USER)
for path in result.secrets:
    print(path)

# Read a secret
secret = api.read_secret(SecretType.USER, "my-api-key")
print(secret.data)  # {"key": "abc123"}

# Update a secret
api.update_secret(
    SecretType.USER,
    "my-api-key",
    data={"key": "new-value"},
)

# Create an app-scoped secret
api.create_secret(
    secret_type=SecretType.APPLICATION,
    secret_path="db-password",
    data={"password": "s3cret"},
    app_id="4cb8894f-90ff-43a8-beba-451f2da3033a",
)

# Delete a secret
api.delete_secret(SecretType.USER, "my-api-key")

Async Example

import asyncio
from komodo.secrets_api import SecretsApi, SecretType

async def main():
    api = SecretsApi(environment="production")

    await api.create_secret_async(
        secret_type=SecretType.USER,
        secret_path="async-key",
        data={"token": "xyz"},
    )

    secret = await api.read_secret_async(SecretType.USER, "async-key")
    print(secret.data)

asyncio.run(main())