Custom API with Snowflake and Python Flask

Many developers want to share some of the metrics of snowflake usage / data stored in snowflake over a custom API. Recent application uses the API to retrieve the data from back end. These API acts bridge between front end and back-end applications. Often many of the users reache on creating custom API with snowflake metrics using python flask API. In this post I am going to create custom API gives the snowflake usage metrics of the given warehouse.

Prerequisites:
· Snowflake account and Privileges to access the WAREHOUSE_METERING_HISTORY.

· Python > 3.8

· Ability to install all the package mentioned in the requirement.txt

Project Folder Structure:

Lets understand the folder structure of the project

Python_services: Root Folder
Get_wh_credit.py is the main script

Snowflake Setup:

Get Sriganesh Palani’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The API needs a warehouse to query the data to return to the caller. To create the database and warehouse, connect to Snowflake and run the following commands in the Snowflake console or using SnowSQL:

USE ROLE ACCOUNTADMIN;
CREATE WAREHOUSE CUSTOM_API_WH WITH WAREHOUSE_SIZE='small';

Create the User and Role in Snowflake:
Lets create an new role and assign the warehouse for the created role.

USE ROLE ACCOUNTADMIN;

CREATE ROLE CUSTOM_API_ROLE;

GRANT USAGE ON WAREHOUSE CUSTOM_API_WH TO ROLE CUSTOM_API_ROLE;

CREATE USER "CUSTOM_API" PASSWORD='Sriga@2013' DEFAULT_ROLE=CUSTOM_API_ROLE DEFAULT_WAREHOUSE=CUSTOM_API_WH MUST_CHANGE_PASSWORD=FALSE;

GRANT ROLE CUSTOM_API_ROLE TO USER CUSTOM_API;

For this use case you need role which has the access to the SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY.

We created above user, role and warehouse to test our custom API

Execution Flow:

Main Script:
Lets understand the main script:

import json
from flask import Flask
from flask import jsonify, Response, Request
from flask import request
from python_services.src import wh_credit_calc


app = Flask(__name__)


@app.route('/getwhmetrics', methods=['POST'])
def get_response():
    """
    Function will get te response from the user and validate the json element:
    If all mandatory fields available in the input JSON
    result will be returned
    :return:
    """
    input_request = request.get_json()
    print(input_request["RequestData"])
    l_metrics = wh_credit_calc.CalcCredit(input_request)
    l_data = l_metrics.validate_schema()

    output_response = {
        "responseStatus": {
            "responseType": "WareHouse Metrics Data",
            "errorMessage": None,
            "errorCode": None,
            "status": "SUCCESS",
            "d":l_data
        }}
    js = json.dumps(output_response, indent=4)
    return Response(js,mimetype='application/json')


if __name__ == "__main__":
    app.run(debug=True)

Response:

Yes! our custom API is working as expected, since given warehouse is newly created one hence no usage as of now.

Now we have created simple custom API which interact the snowflake database and pull the metrics for the given request. If you have any questions, please leave them in the comments below or reach out me on https://www.linkedin.com/in/sriganeshpalani/