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 objetoUserDefinedFunctioncorrespondente, 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.9", 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 objetoProcedurecorrespondente, 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(NUMBER, NUMBER), 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(NUMBER, NUMBER)"]
call_argument_list = CallArgumentList(call_arguments=[
CallArgument(name="id", datatype="NUMBER", value=1),
])
procedure_reference.call(call_argument_list)
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(NUMBER, NUMBER) no banco de dados my_db e no esquema my_schema:
my_procedure = root.databases["my_db"].schemas["my_schema"].procedures["my_procedure(NUMBER, NUMBER)"].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(NUMBER, NUMBER) 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(NUMBER, NUMBER)"]
my_procedure_res.drop()