What is Column-Level Security in Snowflake?

				
					import json
from cryptography.fernet import Fernet
import pandas as pd
import base64
import hashlib
import os

class PIITokenizer:
    def __init__(self, secret_key: str):
        """
        Initialize the tokenizer with a secret key.
        :param secret_key: A string used to derive an encryption key.
        """
        # Derive a 32-byte encryption key from the secret key
        key = hashlib.sha256(secret_key.encode()).digest()
        self.key = base64.urlsafe_b64encode(key[:32])
        self.fernet = Fernet(self.key)

    def tokenize(self, input_json: str) -> str:
        """
        Tokenize sensitive PII data.
        :param data: The sensitive data to be tokenized.
        :return: The tokenized data.
        """
        # Parse the input JSON string into a Python dictionary
        input_data = json.loads(input_json)
       
        # Convert the data into a DataFrame
        df = pd.DataFrame(input_data["data"], columns=["row_number", "value", "action"])

        # Check if the DataFrame is empty
        if df.empty:
            return json.dumps({"data": None})

        # Apply the specified action (encrypt or decrypt) to the "value" column
        df["value"] = df.apply(
            lambda row: (
                ""
                if row["value"] == ""
                else None
                if row["value"] is None
                else self.fernet.encrypt(row["value"].encode()).decode()
                if row["action"] == "encrypt"
                else self.fernet.decrypt(row["value"].encode()).decode()
                if row["action"] == "decrypt"
                else "Invalid operation Type"
            ),
            axis=1
        )
   
        # Convert the processed DataFrame back to the required JSON format
        output_data = {"data": df[["row_number", "value"]].values.tolist()}
        return json.dumps(output_data)
   
def lambda_handler(event, context):
    # 200 is the HTTP status code for "ok".
    status_code = 200
    # The return value will contain an array of arrays (one inner array per input row).
    array_of_rows_to_return = [ ]
    try:
        # From the input parameter named "event", get the body, which contains
        # the input rows.
        secret_key = os.environ.get('secret_key')#The secret key is stored as an env variable.
        tokenizer = PIITokenizer(secret_key)
        event_body = event["body"]
        json_compatible_string_to_return = tokenizer.tokenize(event_body)
    except Exception as err:
        # 400 implies some type of error.
        status_code = 400
        # Tell the caller what this function could not handle.
        raise err
    # Return the return value and HTTP status code.
    return {
        'statusCode': status_code,
        'body': json_compatible_string_to_return
    }