Is there a way to reference environment variables in connections.toml

[ReinierKop]

What are you trying to accomplish?
I am trying to integrate SchemaChange into our dagster/dbt/snowflake stack. We use dotenv to manage our configurations across dev/test/staging/ci/prod, so I'd like to be able to reference environment variables throughout our stack.

Specifically, I'd like to be able to specify environment variables in connections.toml, but my attempts to do so so far have failed, and I've not seen examples in the issue history nor the documentation that tell me how and if I can do this.

What options have you tried so far?

I have a connections.toml file like this:

[myconnection]
account = "${SNOWFLAKE_ACCOUNT}"
user = "${SNOWFLAKE_USER}"
password = "${SNOWFLAKE_PASSWORD}"
warehouse = "${SNOWFLAKE_WAREHOUSE}"
database = "${SNOWFLAKE_DATABASE}"

I then run schemachange -f migrations --connections-file-path ./connections.toml --connection-name myconnection

Expectation: connected successfully to our snowflake warehouse.
Reality: 404 Not Found: post https://${SNOWFLAKE_ACCOUNT}.snowflakecomputing.com:443/session/v1/login-request? .... (error truncated).

I've also tried this with environment variables formatted as "$SNOWFLAKE_ACCOUNT"

Hope someone can point me in the right direction, this seems rather trivial and I am sure I am missing something :-)

[scotthcollins]

I use a custom python app in ADO as a work around. (I initially was going to look for the pattern {{ and }} and replace based on that, however, it took me less thought and time to just inject the whole of the config content. Do as you wish with the given examples :) )

Given a toml template as:

An ADO Pipeline task as:
** $(DatabaseNames) is an ado var list of database names, ie: 'db1','db2' &c

  - task: PythonScript@0
    displayName: 'Update Connections TOML File'
    inputs:
      scriptSource: 'filePath'
      scriptPath: '$(Pipeline.Workspace)/Deploy/python/create_toml_file.py'
    env:
      connections_file_path: $(Pipeline.Workspace)/Deploy/schemachange_config/connections.toml
      database_names: $(DatabaseNames) # pipeline variable array
      warehouse_name: $(SnowFlake-ComputeWarehouse) # plucked from KV
      snowflake_account: $(SnowFlake-Host) # plucked from KV
      snowflake_user: $(RSA-Admin) # plucked from KV
      snowflake_private_key: $(RSA-Admin-PrivateKey) # plucked from KV
      is_rsa: $(Is_RSA) # set in azure PS task condition: if RSA-* exist and are not null
       #If NOT rsa, these are required
      snowflake_password: $(SNOWFLAKE_PASSWORD) # plucked from KV
      snowflake_pw_user: $(snowUser) # plucked from KV

the create_toml_file.py as (note I, personally am using the private_key, and not private_key_file as our stuff comes from the KV)

import os
import stat
import sys

def fix_permissions(file_path):
    """
    Fixes permissions for the specified file.
    Sets the file to be readable and writable only by the owner.
    """
    try:
        # Set file permissions to 600 (read/write for owner only)
        os.chmod(file_path, stat.S_IRUSR | stat.S_IWUSR)
        # Change ownership to the current user
        #os.chown(file_path, os.getuid(), os.getgid())
        print(f"Fixed permissions for {file_path}", flush=True)
    except PermissionError as e:
        message = f"Failed to fix permissions for {file_path}: {e}"
        raise PermissionError(message) from e
    except FileNotFoundError as e:
        message = f"File not found: {file_path}: {e}"
        raise FileNotFoundError(message) from e
    except OSError as e:
        message = f"OS error: {e}"
        raise OSError(message) from e


def add_connection_sections(config):
    """
    Adds new connection sections to the TOML file based on database names.
    """
    try:
        file_path = config['file_path']
        # Read the existing content of the file
        with open(file_path, 'r') as file:
            content = file.read()

        # Prepare new connection sections
        new_sections = []

        if config['is_rsa']:
            authenticator = "SNOWFLAKE_JWT"
            authentication_type = (
                f"private_key = \"{config['snowflake_password']}\""
            )
        else:
            authenticator = "snowflake"
            authentication_type = (
                f"password = \"{config['snowflake_password']}\""
            )

        for db_name in config['database_names']:
            db_name = db_name.strip()  # Clean up any extra whitespace
            section = f"""
            [{db_name}]
            account = "{config['snowflake_account']}"
            user = "{config['snowflake_user']}"
            database = "{db_name}"
            warehouse = "{config['warehouse_name']}"
            role = "SYSADMIN"
            {authentication_type}
            authenticator = "{authenticator}"
            """
            new_sections.append(section.strip())

        # Append new sections to the file content
        updated_content = content + "\n\n" + "\n\n".join(new_sections)

        # Write the updated content back to the file
        with open(file_path, 'w') as file:
            file.write(updated_content)

        print(
            f"Added new connection sections for databases: "
            f"{', '.join(config['database_names'])}",
            flush=True
        )

        # Fix permissions for the file
        fix_permissions(file_path)

    except OSError as e:
        message = f"File operation error: {e}"
        raise OSError(message) from e
    except KeyError as e:
        message = f"Missing configuration key: {e}"
        raise KeyError(message) from e
    except ValueError as e:
        message = f"Value error: {e}"
        raise ValueError(message) from e


def validate_environment_variables():
    """
    Validates required environment variables.
    Raises a ValueError if any are missing.
    Returns a dictionary of validated environment variables.
    """
    # Retrieve environment variables
    file_path = os.environ.get('connections_file_path')
    database_names = os.environ.get('database_names', None)
    warehouse_name = os.environ.get('warehouse_name')
    snowflake_account = os.environ.get('snowflake_account')
    is_rsa = os.environ.get('is_rsa', 'false').lower() == 'true'

    if is_rsa:
        snowflake_user = os.environ.get('snowflake_user')
        snowflake_password = os.environ.get('snowflake_private_key')
    else:
        snowflake_user = os.environ.get('snowflake_pw_user')
        snowflake_password = os.environ.get('snowflake_password')
    
    # Validate required environment variables
    missing_vars = []
    if not file_path:
        missing_vars.append("connections_file_path")
    if not database_names:
        missing_vars.append("database_names")
    if not warehouse_name:
        missing_vars.append("warehouse_name")
    if not snowflake_account:
        missing_vars.append("snowflake_account")
    if not snowflake_user:
        if is_rsa:
            missing_vars.append("snowflake_user")
        else:     
            missing_vars.append("snowflake_pw_user")
    if not snowflake_password:
        if is_rsa:
            missing_vars.append("snowflake_private_key")
        else:
            missing_vars.append("snowflake_password")

    if missing_vars:
        message = (
            f"Error: Missing required environment variables: "
            f"{', '.join(missing_vars)}"
        )
        print(
            f"##vso[task.logissue type=error]{message}",
            flush=True
        )  # Log error to pipeline
        raise RuntimeError(message)

    # Return validated environment variables as a dictionary.
    # This ensures all required variables are present.
    return {
        'file_path': file_path,
        'database_names': database_names.replace(" ", "").split(','),
        'warehouse_name': warehouse_name,
        'snowflake_account': snowflake_account,
        'snowflake_user': snowflake_user,
        'snowflake_password': snowflake_password,
        'is_rsa': is_rsa
    }


if __name__ == "__main__":
    try:
        # Validate environment variables
        config = validate_environment_variables()

        # Add new connection sections
        add_connection_sections(config)
    except (RuntimeError, ValueError, KeyError) as e:
        message = f"Parameter Validation Error: {e}"
        # Log error to ADO
        print(f"##vso[task.logissue type=error]{message}", flush=True)
        sys.exit(1)
    except (OSError, IOError) as e:
        message = f"An error occurred: {e}"
        # Log error to ADO
        print(f"##vso[task.logissue type=error]{message}", flush=True)
        sys.exit(1)
    except Exception as e:
        message = f"Unexpected error: {e}"
        # Log error to ADO
        print(f"##vso[task.logissue type=error]{message}", flush=True)
        sys.exit(1)

[sfc-gh-tmathew]

With release 4.1.0 you will be able to set ENvironment variables that does not have to be defined in the connections.toml file. Please check the updated docs and examples to revisit your need.