Gerenciamento de funções Snowflake e procedimentos armazenados com Python¶

É possível usar Python para gerenciar funções definidas pelo usuário (UDFs) e procedimentos armazenados no Snowflake. Ao criar uma UDF ou procedimento, você escreve sua lógica em uma das linguagens de manipulador compatíveis e, em seguida, o cria usando o Snowflake Python APIs. Para obter mais informações sobre UDFs e procedimentos, consulte Como estender o Snowflake com funções e procedimentos.

Pré-requisitos¶

Os exemplos neste tópico pressupõem que você tenha adicionado código para se conectar ao Snowflake e criar um objeto Root a partir do qual usar o Snowflake Python APIs.

Por exemplo, o seguinte código usa parâmetros de conexão definidos em um arquivo de configuração para criar uma conexão com o Snowflake:

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

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

Usando o objeto Session resultante, o código cria um objeto Root para usar os tipos e métodos de API. Para obter mais informações, consulte Conexão ao Snowflake com o Snowflake Python APIs.

Gerenciamento de funções definidas pelo usuário (UDFs)¶

É possível gerenciar funções definidas pelo usuário (UDFs), que podem ser escritas para estender o sistema para executar operações que não estão disponíveis através das funções definidas pelo sistema integradas fornecidas pelo Snowflake. Após criar uma UDF, é possível reutilizá-la diversas vezes. Para obter mais informações, consulte Visão geral das funções definidas pelo usuário.

Nota

Não há suporte para chamar UDFs usando a API no momento.

O Snowflake Python APIs representa UDFs com dois tipos separados:

UserDefinedFunction: Expõe as propriedades de uma UDF, como nome, lista de argumentos, tipo de retorno e definição de função.
UserDefinedFunctionResource: Expõe métodos que podem ser usados para buscar um objeto UserDefinedFunction correspondente, renomear a UDF e descartar a UDF.

Criação de uma UDF¶

Para criar uma UDF, primeiro crie um objeto UserDefinedFunction e depois crie um objeto UserDefinedFunctionCollection a partir do objeto Root da API. Usando UserDefinedFunctionCollection.create, adicione a nova UDF ao Snowflake.

Ao criar uma UDF, você especifica um manipulador cujo código é escrito em uma das seguintes linguagens compatíveis.

Python¶

O código no exemplo a seguir cria um objeto UserDefinedFunction que representa uma UDF nomeada my_python_function no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno, a linguagem e definição Python da UDF:

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.12", packages=[], handler="udf"),
    body="""
def udf():
    return {"key": "value"}
    """,
)

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

Java¶

O código no exemplo a seguir cria um objeto UserDefinedFunction que representa uma UDF nomeada my_java_function no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno, a linguagem e a definição Java da UDF:

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)

JavaScript¶

O código no exemplo a seguir cria um objeto UserDefinedFunction que representa uma UDF nomeada my_javascript_function no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno, a linguagem e a definição JavaScript da UDF:

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

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)

Scala¶

O código no exemplo a seguir cria um objeto UserDefinedFunction que representa uma UDF nomeada my_scala_function no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno, a linguagem e a definição Scala da UDF:

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

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)

SQL¶

O código no exemplo a seguir cria um objeto UserDefinedFunction que representa uma UDF nomeada my_sql_function no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno, a linguagem e a definição SQL da UDF:

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

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)

Como obter detalhes da UDF¶

É possível obter informações sobre a UDF chamando o método UserDefinedFunctionResource.fetch, que retorna um objeto UserDefinedFunction.

O código no exemplo a seguir busca informações sobre UDF my_javascript_function(DOUBLE) no banco de dados my_db e no esquema my_schema:

Nota

Ao obter um objeto de recurso da UDF, é necessário especificar a assinatura completa (o nome da UDF e seus tipos de dados de parâmetro), visto que as UDFs podem ser sobrecarregadas.

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

Listagem de UDFs¶

É possível listar UDFs usando o método UserDefinedFunctionCollection.iter, que retorna um iterador PagedIter de objetos UserDefinedFunction.

O código no exemplo a seguir lista aas UDFs cujo nome começa com my_java no banco de dados my_db e no esquema my_schema e, em seguida, imprime o nome de cada um:

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)

Como renomear uma UDF¶

É possível renomear uma UDF com um objeto UserDefinedFunctionResource.

O código no exemplo a seguir obtém o objeto de recurso da UDF my_javascript_function(DOUBLE) no banco de dados my_db e no esquema my_schema e, em seguida, renomeia a UDF para my_other_js_function, ao mesmo tempo em que a move para o banco de dados my_other_db e o esquema my_other_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"
)

Descarte de uma UDF¶

É possível descartar uma UDF com um objeto UserDefinedFunctionResource.

O código no exemplo a seguir obtém o objeto de recurso da UDF my_javascript_function(DOUBLE) e descarta a UDF:

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

Gerenciamento de procedimentos armazenados¶

É possível gerenciar procedimentos armazenados, que você pode escrever para estender o sistema com código procedural que executa SQL. Em um procedimento armazenado, você pode usar construções programáticas para realizar ramificações e looping. Após criar um procedimento armazenado, você pode reutilizá-lo diversas vezes. Para obter mais informações, consulte Visão geral dos procedimentos armazenados.

O Snowflake Python APIs representa procedimentos com dois tipos separados:

Procedure: Expõe as propriedades de um procedimento, como nome, lista de argumentos, tipo de retorno e definição do procedimento.
ProcedureResource: Expõe métodos que você pode usar para buscar um objeto Procedure correspondente, chamar o procedimento e descartar o procedimento.

Criação de um procedimento¶

Para criar um procedimento, primeiro crie um objeto Procedure e depois crie um objeto ProcedureCollection a partir do objeto Root da API. Usando ProcedureCollection.create, adicione o novo procedimento ao Snowflake.

O código no exemplo a seguir cria um objeto Procedure que representa um procedimento nomeado my_procedure no banco de dados my_db e no esquema my_schema, com os argumentos especificados, o tipo de retorno e a definição de procedimento SQL:

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)

Como chamar um procedimento¶

É possível chamar um procedimento com um objeto ProcedureResource.

O código no exemplo a seguir obtém o objeto de recurso do procedimento my_procedure(VARCHAR), cria um objeto CallArgumentList e, em seguida, chama o procedimento usando essa lista de argumentos.

Nota

Ao obter um objeto de recurso de procedimento, é necessário especificar a assinatura completa (o nome do procedimento e seus tipos de dados de parâmetro) porque os procedimentos podem ser sobrecarregados.

from snowflake.core.procedure import CallArgument, CallArgumentList

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

Como obter detalhes do procedimento¶

É possível obter informações sobre um procedimento chamando o método ProcedureResource.fetch, que retorna um objeto Procedure.

O código no exemplo a seguir busca informações sobre o procedimento my_procedure(VARCHAR) no banco de dados my_db e no esquema my_schema:

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

Procedimentos de listagem¶

É possível listar procedimentos usando o método ProcedureCollection.iter, que retorna um iterador PagedIter de objetos Procedure.

O código no exemplo a seguir lista procedimentos cujos nomes começam com my no banco de dados my_db e no esquema my_schema e, em seguida, imprime o nome de cada um:

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

Descarte de um procedimento¶

É possível descartar um procedimento com um objeto ProcedureResource.

O código no exemplo a seguir obtém o objeto de recurso do procedimento my_procedure(VARCHAR) no banco de dados my_db e no esquema my_schema e, em seguida, descarta o procedimento.

my_procedure_res = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(VARCHAR)"]
my_procedure_res.drop()