Snowflake template¶

CREATE TABLE "DEMO_DB"."PUBLIC"."PRODUCT" ("NAME" STRING, "PRICE" DOUBLE);

INSERT INTO "DEMO_DB"."PUBLIC"."PRODUCT" VALUES ('Apples', 1.24), ('Bananas', 3.54), ('Oranges', 2.34);

API_TOKEN = "<API_TOKEN>" # Make sure this is in the format "Token token-code"
PROJECT_NAME = "<PROJECT_NAME>"
DEPLOYMENT_NAME='snowflake-deployment'
DEPLOYMENT_VERSION='v1'


# Authentication configuration with the snowflake environment

# SNOWFLAKE_ACCOUNT will follow this format "<PROJECT_ID>.<REGION_OF_SNOWFLAKE>.<CLOUD_PROVIDER>"
# Example: "xxxxxx.europe-west4.gcp"
SNOWFLAKE_ACCOUNT = '<INSERT SNOWFLAKE ACCOUNT>'

SNOWFLAKE_USERNAME= '<INSERT SNOWFLAKE USERNAME>'
SNOWFLAKE_PASSWORD = '<INSERT SNOWFLAKE PASSWORD>'
SNOWFLAKE_DATABASE = 'DEMO_DB'



# Import all necessary libraries
import shutil
import os
import ubiops

client = ubiops.ApiClient(ubiops.Configuration(api_key={'Authorization': API_TOKEN}, 
                                               host='https://api.ubiops.com/v2.1'))
api = ubiops.CoreApi(client)

%%writefile snowflake_deployment/deployment.py
"""
The file containing the deployment code is required to be called 'deployment.py' and should contain the 'Deployment'
class and 'request' method.
"""
import snowflake.connector as sf
import os


class Deployment:

    def __init__(self, base_directory, context):
        """
        Initialisation method for the deployment. It can for example be used for loading modules that have to be kept in
        memory or setting up connections. Load your external model files (such as pickles or .h5 files) here.

        :param str base_directory: absolute path to the directory where the deployment.py file is located
        :param dict context: a dictionary containing details of the deployment that might be useful in your code.
            It contains the following keys:
                - deployment (str): name of the deployment
                - version (str): name of the version
                - input_type (str): deployment input type, either 'structured' or 'plain'
                - output_type (str): deployment output type, either 'structured' or 'plain'
                - environment (str): programming environment the deployment is running
                - environment_variables (str): the custom environment variables configured for the deployment.
                    You can also access those as normal environment variables via os.environ
        """

        print("Initialising My Deployment")
        print('Connecting to snowflake database')

        SNOWFLAKE_ACCOUNT = os.environ.get('SNOWFLAKE_ACCOUNT')
        SNOWFLAKE_USERNAME= os.environ.get('SNOWFLAKE_USERNAME')
        SNOWFLAKE_PASSWORD = os.environ.get('SNOWFLAKE_PASSWORD')
        SNOWFLAKE_DATABASE = os.environ.get('SNOWFLAKE_DATABASE')

        try:        
            con = sf.connect(
                user=SNOWFLAKE_USERNAME,
                password=SNOWFLAKE_PASSWORD,
                account=SNOWFLAKE_ACCOUNT,
                database=SNOWFLAKE_DATABASE
            )
            self.cur = con.cursor()

        except Exception as e:
            print('There was a problem connecting to the database!')
            print(e)


    def request(self, data):
        """
        Method for deployment requests, called separately for each individual request.

        :param dict/str data: request input data. In case of deployments with structured data, a Python dictionary
            with as keys the input fields as defined upon deployment creation via the platform. In case of a deployment
            with plain input, it is a string.
        :return dict/str: request output. In case of deployments with structured output data, a Python dictionary
            with as keys the output fields as defined upon deployment creation via the platform. In case of a deployment
            with plain output, it is a string. In this example, a dictionary with the key: output.
        """

        # Price limit
        max_price = data['max_price']
        self.cur.execute(f'SELECT * from product where price < {max_price};')
        items = self.cur.fetchall()

        # No affordable items
        if len(items) == 0:
            return {
                'output': "unfortunately you cannot afford any of the items in our shop!"
            }

        # Format items that the user can purchase
        affordable_items = ', '.join(list(map(lambda x: x[0], items)))
        return {
            "output": f"You can afford to buy the following ({affordable_items})"
        }

# Create the deployment
deployment_template = ubiops.DeploymentCreate(
    name=DEPLOYMENT_NAME,
    description='Snowflake deployment',
    input_type='structured',
    output_type='structured',
    input_fields=[
        {'name':'max_price', 'data_type':'double'}
    ],
    output_fields=[
        {'name':'output', 'data_type':'string'},
    ]
)

api.deployments_create(
    project_name=PROJECT_NAME,
    data=deployment_template
)


# Create secret environment variables for the snowflake connection
api.deployment_environment_variables_create(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    data=ubiops.EnvironmentVariableCreate(
        name='SNOWFLAKE_ACCOUNT', 
        value=SNOWFLAKE_ACCOUNT,
        secret=True
    )
)
api.deployment_environment_variables_create(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    data=ubiops.EnvironmentVariableCreate(
        name='SNOWFLAKE_USERNAME', 
        value=SNOWFLAKE_USERNAME,
        secret=True
    )
)
api.deployment_environment_variables_create(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    data=ubiops.EnvironmentVariableCreate(
        name='SNOWFLAKE_PASSWORD', 
        value=SNOWFLAKE_PASSWORD,
        secret=True
    )
)
api.deployment_environment_variables_create(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    data=ubiops.EnvironmentVariableCreate(
        name='SNOWFLAKE_DATABASE', 
        value=SNOWFLAKE_DATABASE,
        secret=True
    )
)


# Create the version
version_template = ubiops.DeploymentVersionCreate(
    version=DEPLOYMENT_VERSION,
    environment='python3-7',
    instance_type='512mb',
    minimum_instances=0,
    maximum_instances=1,
    maximum_idle_time=1800, # = 30 minutes
    request_retention_mode='none' # we don't need request storage in this example
)

api.deployment_versions_create(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    data=version_template
)

# Zip the deployment package
shutil.make_archive('snowflake_deployment', 'zip', '.', 'snowflake_deployment')

# Upload the zipped deployment package
file_upload_result =api.revisions_file_upload(
    project_name=PROJECT_NAME,
    deployment_name=DEPLOYMENT_NAME,
    version=DEPLOYMENT_VERSION,
    file='snowflake_deployment.zip'
)

api_client.close()