Managing Snowflake functions and stored procedures with Python

You can use Python to manage user-defined functions (UDFs) and stored procedures in Snowflake. When you create a UDF or procedure, you write its logic in one of the supported handler languages, then create it using the Snowflake Python APIs. For more information about UDFs and procedures, see Extending Snowflake with Functions and Procedures.

Prerequisites

The examples in this topic assume that you’ve added code to connect with Snowflake and to create a Root object from which to use the Snowflake Python APIs.

For example, the following code uses connection parameters defined in a configuration file to create a connection to Snowflake:

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

Using the resulting Session object, the code creates a Root object to use the API’s types and methods. For more information, see Connect to Snowflake with the Snowflake Python APIs.

Managing user-defined functions (UDFs)

You can manage user-defined functions (UDFs), which you can write to extend the system to perform operations that are not available through the built-in system-defined functions provided by Snowflake. After you create a UDF, you can reuse it multiple times. For more information, see User-defined functions overview.

Note

Calling UDFs by using the API is currently not supported.

The Snowflake Python APIs represents UDFs with two separate types:

  • UserDefinedFunction: Exposes a UDF’s properties such as its name, list of arguments, return type, and function definition.

  • UserDefinedFunctionResource: Exposes methods you can use to fetch a corresponding UserDefinedFunction object, rename the UDF, and drop the UDF.

Creating a UDF

To create a UDF, first create a UserDefinedFunction object, and then create a UserDefinedFunctionCollection object from the API Root object. Using UserDefinedFunctionCollection.create, add the new UDF to Snowflake.

When you create a UDF, you specify a handler whose code is written in one of the following supported languages.

Python

Code in the following example creates a UserDefinedFunction object that represents a UDF named my_python_function in the my_db database and the my_schema schema, with the specified arguments, return type, language, and UDF Python definition:

from snowflake.core.user_defined_function import (
    PythonFunction,
    ReturnDataType,
    UserDefinedFunction
)

function_of_python = UserDefinedFunction(
    "my_python_function",
    arguments=[],
    return_type=ReturnDataType(datatype="VARIANT"),
    language_config=PythonFunction(runtime_version="3.8", packages=[], handler="udf"),
    body="""
def udf():
    return {"key": "value"}
    """,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_python)
Copy

Java

Code in the following example creates a UserDefinedFunction object that represents a UDF named my_java_function in the my_db database and the my_schema schema, with the specified arguments, return type, language, and UDF Java definition:

from snowflake.core.user_defined_function import (
    Argument,
    JavaFunction,
    ReturnDataType,
    UserDefinedFunction
)

function_body = """
    class TestFunc {
        public static String echoVarchar(String x) {
            return x;
        }
    }
"""

function_of_java = UserDefinedFunction(
    name="my_java_function",
    arguments=[Argument(name="x", datatype="STRING")],
    return_type=ReturnDataType(datatype="VARCHAR", nullable=True),
    language_config=JavaFunction(
        handler="TestFunc.echoVarchar",
        runtime_version="11",
        target_path=target_path,
        packages=[],
        called_on_null_input=True,
        is_volatile=True,
    ),
    body=function_body,
    comment="test_comment",
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_java)
Copy

JavaScript

Code in the following example creates a UserDefinedFunction object that represents a UDF named my_javascript_function in the my_db database and the my_schema schema, with the specified arguments, return type, language, and UDF JavaScript definition:

from snowflake.core.user_defined_function import (
    Argument,
    ReturnDataType,
    JavaScriptFunction
)

function_body = """
    if (D <= 0) {
        return 1;
    } else {
        var result = 1;
        for (var i = 2; i <= D; i++) {
            result = result * i;
        }
        return result;
    }
"""

function_of_javascript = UserDefinedFunction(
    name="my_javascript_function",
    arguments=[Argument(name="d", datatype="DOUBLE")],
    return_type=ReturnDataType(datatype="DOUBLE"),
    language_config=JavaScriptFunction(),
    body=function_body,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_javascript)
Copy

Scala

Code in the following example creates a UserDefinedFunction object that represents a UDF named my_scala_function in the my_db database and the my_schema schema, with the specified arguments, return type, language, and UDF Scala definition:

from snowflake.core.user_defined_function import (
    Argument,
    ReturnDataType,
    ScalaFunction
)

function_body = """
    class Echo {
        def echoVarchar(x : String): String = {
            return x
        }
    }
"""

function_of_scala = UserDefinedFunction(
    name="my_scala_function",
    arguments=[Argument(name="x", datatype="VARCHAR")],
    return_type=ReturnDataType(datatype="VARCHAR"),
    language_config=ScalaFunction(
        runtime_version="2.12", handler="Echo.echoVarchar", target_path=target_path, packages=[]
    ),
    body=function_body,
    comment="test_comment",
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_scala)
Copy

SQL

Code in the following example creates a UserDefinedFunction object that represents a UDF named my_sql_function in the my_db database and the my_schema schema, with the specified arguments, return type, language, and UDF SQL definition:

from snowflake.core.user_defined_function import (
    Argument,
    ReturnDataType,
    SQLFunction
)

function_body = """3.141592654::FLOAT"""

function_of_sql = UserDefinedFunction(
    name="my_sql_function",
    arguments=[],
    return_type=ReturnDataType(datatype="FLOAT"),
    language_config=SQLFunction(),
    body=function_body,
)

root.databases["my_db"].schemas["my_schema"].user_defined_functions.create(function_of_sql)
Copy

Getting UDF details

You can get information about a UDF by calling the UserDefinedFunctionResource.fetch method, which returns a UserDefinedFunction object.

Code in the following example fetches information about the my_javascript_function(DOUBLE) UDF in the my_db database and the my_schema schema:

Note

When getting a UDF resource object, you must specify the full signature (the UDF name and its parameter data types) because UDFs can be overloaded.

my_udf = root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"].fetch()
print(my_udf.to_dict())
Copy

Listing UDFs

You can list UDFs using the UserDefinedFunctionCollection.iter method, which returns a PagedIter iterator of UserDefinedFunction objects.

Code in the following example lists UDFs whose name starts with my_java in the my_db database and the my_schema schema, and then prints the name of each:

udf_iter = root.databases["my_db"].schemas["my_schema"].user_defined_functions.iter(like="my_java%")
for udf_obj in udf_iter:
    print(udf_obj.name)
Copy

Renaming a UDF

You can rename a UDF with a UserDefinedFunctionResource object.

Code in the following example gets the my_javascript_function(DOUBLE) UDF resource object in the my_db database and the my_schema schema, and then renames the UDF to my_other_js_function while also moving it to the my_other_db database and the my_other_schema schema:

root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"].rename(
    "my_other_js_function",
    target_database = "my_other_database",
    target_schema = "my_other_schema"
)
Copy

Dropping a UDF

You can drop a UDF with a UserDefinedFunctionResource object.

Code in the following example gets the my_javascript_function(DOUBLE) UDF resource object and then drops the UDF:

my_udf_res = root.databases["my_db"].schemas["my_schema"].user_defined_functions["my_javascript_function(DOUBLE)"]
my_udf_res.drop()
Copy

Managing stored procedures

You can manage stored procedures, which you can write to extend the system with procedural code that executes SQL. In a stored procedure, you can use programmatic constructs to perform branching and looping. After you create a stored procedure, you can reuse it multiple times. For more information, see Stored procedures overview.

The Snowflake Python APIs represents procedures with two separate types:

  • Procedure: Exposes a procedure’s properties such as its name, list of arguments, return type, and procedure definition.

  • ProcedureResource: Exposes methods you can use to fetch a corresponding Procedure object, call the procedure, and drop the procedure.

Creating a procedure

To create a procedure, first create a Procedure object, and then create a ProcedureCollection object from the API Root object. Using ProcedureCollection.create, add the new procedure to Snowflake.

Code in the following example creates a Procedure object that represents a procedure named my_procedure in the my_db database and the my_schema schema, with the specified arguments, return type, and SQL procedure definition:

from snowflake.core.procedure import Argument, ColumnType, Procedure, ReturnTable, SQLFunction

procedure = Procedure(
    name="my_procedure",
    arguments=[Argument(name="id", datatype="VARCHAR")],
    return_type=ReturnTable(
        column_list=[
            ColumnType(name="id", datatype="NUMBER),
            ColumnType(name="price", datatype="NUMBER"),
        ]
    ),
    language_config=SQLFunction(),
    body="
        DECLARE
            res RESULTSET DEFAULT (SELECT * FROM invoices WHERE id = :id);
        BEGIN
            RETURN TABLE(res);
        END;
    ",
)

procedures = root.databases["my_db"].schemas["my_schema"].procedures
procedures.create(procedure)
Copy

Calling a procedure

You can call a procedure with a ProcedureResource object.

Code in the following example gets the my_procedure(NUMBER, NUMBER) procedure resource object, creates a CallArgumentList object, and then calls the procedure using that list of arguments.

Note

When getting a procedure resource object, you must specify the full signature (the procedure name and its parameter data types) because procedures can be overloaded.

from snowflake.core.procedure import CallArgument, CallArgumentList

procedure_reference = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"]
call_argument_list = CallArgumentList(call_arguments=[
    CallArgument(name="id", datatype="NUMBER", value=1),
])
procedure_reference.call(call_argument_list)
Copy

Getting procedure details

You can get information about a procedure by calling the ProcedureResource.fetch method, which returns a Procedure object.

Code in the following example fetches information about the my_procedure(NUMBER, NUMBER) procedure in the my_db database and the my_schema schema:

my_procedure = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"].fetch()
print(my_procedure.to_dict())
Copy

Listing procedures

You can list procedures using the ProcedureCollection.iter method, which returns a PagedIter iterator of Procedure objects.

Code in the following example lists procedures whose name starts with my in the my_db database and the my_schema schema, and then prints the name of each:

procedure_iter = root.databases["my_db"].schemas["my_schema"].procedures.iter(like="my%")
for procedure_obj in procedure_iter:
    print(procedure_obj.name)
Copy

Dropping a procedure

You can drop a procedure with a ProcedureResource object.

Code in the following example gets the my_procedure(NUMBER, NUMBER) procedure resource object in the my_db database and the my_schema schema, and then drops the procedure.

my_procedure_res = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"]
my_procedure_res.drop()
Copy
Language: English