Connect to Snowflake from Python Lambda Function

Lets consider that the access credentials in order to connect to Snowflake is available in AWS Secrets Manager.

Below is the sample Python code that demonstrates how to connect to Snowflake:

import os
import base64
import json
import boto3
from botocore.exceptions import ClientError
import snowflake.connector
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.hazmat.primitives import serialization



# Snowflake Access Secrets
sf_access = os.environ['sf_access']
sm_client = boto3.client('secretsmanager', region_name = 'eu-central-1')
get_secret_value_response = sm_client.get_secret_value(
    SecretId = sf_access
)
if 'SecretString' in get_secret_value_response:
    secret = json.loads(get_secret_value_response['SecretString'])
else:
    secret = json.loads(base64.b64decode(get_secret_value_response['SecretBinary']))


# Snowflake Settings
p_key = serialization.load_pem_private_key(
  data = secret['sf_private_key'].encode(),
  password = secret['sf_passphrase'].encode(),
  backend = default_backend()
)

pkb = p_key.private_bytes(
  encoding = serialization.Encoding.DER,
  format = serialization.PrivateFormat.PKCS8,
  encryption_algorithm = serialization.NoEncryption()
)

conn = snowflake.connector.connect(
    account = secret['sf_account'],
    user = secret['sf_user'],
    role = secret['sf_role'],
    private_key = pkb,
    warehouse = secret['sf_warehouse'],
    database = secret['sf_database'],
    schema = secret['sf_schema']
)

print('Connection to Snowflake succeeded')
cur = conn.cursor()


def lambda_handler(event, context):
    cust_id = event['cust_id']
    cust_name = event['cust_name']
    cust_address = event['cust_address']

    query = f"""INSERT INTO mydatabase.myschema.mytable
            (CUST_ID, CUST_NAME, CUST_ADDRESS, CUST_STATUS)
            VALUES ('{cust_id}', '{cust_name}', '{cust_address}', 'JOINED')
        """
    cur.execute(query)

    return True

In this article, we have demonstrated how to connect to Snowflake from an AWS Lambda function written in Python. By following these steps, you can integrate your Snowflake data warehouse with your AWS-based applications and workflows.