Gerenciamento de bancos de dados, esquemas, tabelas e exibições Snowflake com Python

É possível usar Python para gerenciar bancos de dados, esquemas, tabelas e exibições Snowflake. Para obter mais informações sobre como gerenciar e trabalhar com dados no Snowflake, consulte Bancos de dados, tabelas e exibições - Visão geral.

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 bancos de dados

Você pode gerenciar bancos de dados no Snowflake. O Snowflake Python APIs representa bancos de dados com dois tipos separados:

  • Database: expõe as propriedades de um banco de dados, como seu nome.

  • DatabaseResource: Expõe métodos que você pode usar para buscar um objeto Database correspondente e descartar o banco de dados.

Tópicos

Como criar um banco de dados

Você pode criar um banco de dados chamando o método DatabaseCollection.create e passando um objeto Database que representa o banco de dados que você deseja criar. Para criar um banco de dados, primeiro crie um objeto Database que especifique o nome do banco de dados.

O código no exemplo a seguir cria um objeto Database que representa um banco de dados nomeado my_db e, em seguida, cria o banco de dados passando o objeto Database ao método DatabaseCollection.create:

from snowflake.core.database import Database

my_db = Database(name="my_db")
root.databases.create(my_db)

Obtenção dos detalhes do banco de dados

Você pode obter informações sobre um banco de dados chamando o método DatabaseResource.fetch, que retorna um objeto Database.

O código no exemplo a seguir obtém informações sobre um banco de dados nomeado my_db:

my_db = root.databases["my_db"].fetch()
print(my_db.to_dict())

Listagem dos bancos de dados

Você pode listar bancos de dados usando o método iter, que retorna um iterador PagedIter.

O código no exemplo a seguir lista bancos de dados cujos nomes começam com my:

databases = root.databases.iter(like="my%")
for database in databases:
  print(database.name)

Descarte ou restauração de um banco de dados

Você pode descartar um banco de dados usando o método DatabaseResource.drop ou restaurar um banco de dados usando o método DatabaseResource.undrop.

Para demonstrar essas operações, o código do exemplo a seguir descarta e, em seguida, restaura a versão mais recente do banco de dados my_db:

my_db_res = root.databases["my_db"]
my_db_res.drop()
my_db_res.undrop()

Gerenciamento de esquemas

Você pode gerenciar esquemas no Snowflake. Um esquema é um objeto no nível do banco de dados. Ao criar ou referenciar um esquema, você faz isso no contexto de seu banco de dados.

O Snowflake Python APIs representa esquemas com dois tipos separados:

  • Schema: expõe as propriedades de um esquema, como seu nome.

  • SchemaResource: Expõe métodos que você pode usar para buscar um objeto Schema correspondente e descartar o esquema.

Tópicos

Criação de um esquema

Para criar um esquema, primeiro crie um objeto Schema que especifique o nome do esquema.

O código no exemplo a seguir cria um objeto Schema que representa um esquema nomeado my_schema:

from snowflake.core.schema import Schema

my_schema = Schema(name="my_schema")
root.databases["my_db"].schemas.create(my_schema)

O código então cria o esquema no banco de dados my_db passando o objeto Schema para o método SchemaCollection.create.

Obtenção de detalhes do esquema

Você pode obter informações sobre um esquema chamando o método SchemaResource.fetch, que retorna um objeto Schema.

O código no exemplo a seguir obtém um objeto Schema que representa o esquema my_schema:

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

Listagem de esquemas

Você pode listar os esquemas em um banco de dados especificado usando o método iter. O método retorna um iterador PagedIter de objetos Schema.

O código no exemplo a seguir lista os nomes dos esquemas no banco de dados my_db:

schema_list = root.databases["my_db"].schemas.iter()
for schema_obj in schema_list:
  print(schema_obj.name)

Descarte ou restauração de um esquema

Você pode descartar um esquema usando o método SchemaResource.drop ou restaurar um esquema usando o método SchemaResource.undrop.

Para demonstrar essas operações, o código do exemplo a seguir descarta e, em seguida, restaura a versão mais recente do esquema my_schema:

my_schema_res = root.databases["my_db"].schemas["my_schema"]
my_schema_res.drop()
my_schema_res.undrop()

Gerenciamento de tabelas padrão

É possível gerenciar tabelas padrão no Snowflake. Uma tabela é um objeto no nível do esquema. Ao criar ou referenciar uma tabela, você faz isso no contexto de seu esquema.

O Snowflake Python APIs representa tabelas com dois tipos separados:

  • Table: expõe as propriedades de uma tabela, como nome e colunas.

  • TableResource: Expõe métodos que podem ser usados para buscar um objeto Table correspondente, atualizar as propriedades da tabela e excluir a tabela.

Tópicos

Como criar uma tabela

Para criar uma tabela, primeiro crie um objeto Table que especifique o nome da tabela, os nomes das colunas e os tipos de dados das colunas.

O código no exemplo a seguir cria um objeto Table que representa uma tabela nomeada my_table com as colunas especificadas:

from snowflake.core.table import Table, TableColumn

my_table = Table(
  name="my_table",
  columns=[TableColumn(name="c1", datatype="int", nullable=False),
           TableColumn(name="c2", datatype="string")]
)
root.databases["my_db"].schemas["my_schema"].tables.create(my_table)

O código cria então a tabela no banco de dados my_db e esquema my_schema passando o objeto Table para o método TableCollection.create.

Obtenção de detalhes da tabela

Você pode obter informações sobre uma tabela chamando o método TableResource.fetch, que retorna um objeto Table.

O código no exemplo a seguir obtém informações sobre uma tabela nomeada my_table:

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
print(my_table.to_dict())

Criação ou alteração de uma tabela

É possível definir as propriedades de um objeto Table e passá-las ao método TableResource.create_or_alter para criar uma tabela, caso ela não exista, ou alterá-la de acordo com a definição da tabela, caso ela exista. O comportamento do create_or_alter deve ser idempotente, o que significa que o objeto da tabela resultante será o mesmo, independentemente de a tabela existir ou não antes de você chamar o método.

Nota

O método create_or_alter usa valores padrão para quaisquer propriedades da tabela que você não define explicitamente. Por exemplo, se você não definir data_retention_time_in_days, seu valor padrão será None mesmo se a tabela já existisse com um valor diferente.

O código no exemplo a seguir acrescenta uma nova coluna nomeada c3 de tipo de dados int à tabela my_table e, em seguida, altera a tabela no Snowflake:

from snowflake.core.table import PrimaryKey, TableColumn

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
my_table.columns.append(TableColumn(name="c3", datatype="int", nullable=False, constraints=[PrimaryKey()]))

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.create_or_alter(my_table)

Listagem de tabelas

Você pode listar as tabelas em um esquema especificado usando o método iter, que retorna um iterador PagedIter de objetos Table.

O código no exemplo a seguir lista tabelas cujos nomes começam com my:

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

Troca de nomes de tabelas

Você pode trocar o nome de uma tabela pelo nome de outra tabela em uma única transação usando o método TableResource.swap_with. Para obter mais informações, consulte a descrição do parâmetro SWAP WITH em ALTER TABLE.

O código no exemplo a seguir troca my_table por other_table no mesmo banco de dados e esquema:

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.swap_with("other_table")

O código no exemplo a seguir troca my_table (no banco de dados my_db e no esquema my_schema) por other_table (no banco de dados other_db e no esquema other_schema):

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.swap_with(to_swap_table_name="other_table", target_database="other_db", target_schema="other_schema")

Execução de operações de tabela

Você pode executar operações comuns de tabela — como gerenciar reclustering para uma tabela e descartar ou restaurar uma tabela — com um objeto TableResource.

Para obter mais informações sobre essas operações de tabela, consulte Comandos de tabela, exibição, sequência e tipo definido pelo usuário na referência do comando SQL.

Para demonstrar algumas operações que você pode fazer com um recurso de tabela, o código do exemplo a seguir faz o seguinte:

  1. Obtém o objeto de recurso da tabela my_table no banco de dados my_db e no esquema my_schema.

  2. Suspende o reclustering da tabela.

  3. Retoma o reclustering da tabela.

  4. Descarta a tabela.

  5. Restaura a versão mais recente da tabela descartada.

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]

my_table_res.suspend_recluster()
my_table_res.resume_recluster()
my_table_res.drop()
my_table_res.undrop()

Gerenciamento de tabelas de evento

É possível gerenciar tabelas de evento Snowflake, que são um tipo especial de tabela do banco de dados com um conjunto predefinido de colunas onde o Snowflake pode coletar dados de telemetria. Para obter mais informações, consulte Visão geral da tabela de evento.

O Snowflake Python APIs representa tabelas de evento com dois tipos separados:

  • EventTable: Expõe as propriedades de uma tabela de evento, como nome, tempo de retenção de dados, tempo máximo de extensão de dados e opção de rastreamento de alterações.

  • EventTableResource: Expõe métodos que você pode usar para buscar um objeto EventTable correspondente, renomear a tabela de evento e descartar a tabela de evento.

Tópicos

Criação de uma tabela de evento

Para criar uma tabela de evento, primeiro crie um objeto EventTable e, em seguida, crie um objeto EventTableCollection a partir do objeto Root da API. Usando EventTableCollection.create, adicione a nova tabela de evento ao Snowflake.

O código no exemplo a seguir cria um objeto EventTable que representa uma tabela de evento nomeada my_event_table com os parâmetros especificados:

from snowflake.core.event_table import EventTable

event_table = EventTable(
  name="my_event_table",
  data_retention_time_in_days = 3,
  max_data_extension_time_in_days = 5,
  change_tracking = True,
  default_ddl_collation = 'EN-CI',
  comment = 'CREATE EVENT TABLE'
)

event_tables = root.databases["my_db"].schemas["my_schema"].event_tables
event_tables.create(my_event_table)

O código cria uma variável EventTableCollection chamada event_tables e usa EventTableCollection.create para criar uma nova tabela de evento no Snowflake.

Como obter detalhes da tabela de evento

É possível obter informações sobre uma tabela de evento chamando o método EventTableResource.fetch, que retorna um objeto EventTable.

O código no exemplo a seguir obtém informações sobre uma tabela de evento nomeada my_event_table:

my_event_table = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"].fetch()
print(my_event_table.to_dict())

Listagem de tabelas de evento

É possível listar tabelas de evento usando o método EventTableCollection.iter, que retorna um iterador PagedIter de objetos EventTable.

O código no exemplo a seguir lista tabelas de evento cujos nomes começam com my no banco de dados my_db e no esquema my_schema e imprime o nome de cada uma:

from snowflake.core.event_table import EventTableCollection

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(like="my%")  # returns a PagedIter[EventTable]
for event_table_obj in event_table_iter:
  print(event_table_obj.name)

O código no exemplo a seguir também lista tabelas de evento cujos nomes começam com my, mas usa o parâmetro starts_with em vez de like. Este exemplo também define o parâmetro opcional show_limit=10 para limitar o número de resultados para 10:

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(starts_with="my", show_limit=10)
for event_table_obj in event_table_iter:
  print(event_table_obj.name)

Execução de operações de tabela de evento

É possível executar operações comuns de tabela de evento – como renomear uma tabela de evento e descartar uma tabela de evento – com um objeto EventTableResource.

Nota

Atualmente, somente a funcionalidade RENAME de ALTER TABLE (tabelas de eventos) é compatível.

RENAME não é compatível na tabela de evento padrão, SNOWFLAKE.TELEMETRY.EVENTS.

Para demonstrar as operações que você pode fazer com um recurso de tabela de evento, o código no exemplo a seguir faz o seguinte:

  1. Obtém o objeto de recurso da tabela de eventos my_event_table no banco de dados my_db e no esquema my_schema.

  2. Renomeia a tabela de evento.

  3. Descarta a tabela de evento.

my_event_table_res = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"]

my_event_table_res.rename("my_other_event_table")
my_event_table_res.drop()

Gerenciamento de exibições

É possível gerenciar exibições no Snowflake. Uma exibição é um objeto ao nível de esquema e permite que o resultado de uma consulta seja acessado como se fosse uma tabela. Ao criar ou referenciar uma exibição, você faz isso no contexto de seu esquema.

Nota

ALTER VIEW não é compatível atualmente.

O Snowflake Python APIs representa exibições com dois tipos separados:

  • View: Expõe as propriedades de uma exibição, como nome, colunas e instrução de consulta SQL.

  • ViewResource: Expõe métodos que você pode usar para buscar um objeto View correspondente e descartar a exibição.

Tópicos

Criação de uma exibição

Para criar uma exibir, primeiro crie um objeto View que especifique o nome da exibição, as colunas e a instrução de consulta SQL.

O código no exemplo a seguir cria um objeto View que representa uma exibição nomeada my_view com as colunas e consulta SQL especificadas:

from snowflake.core.view import View, ViewColumn

my_view = View(
  name="my_view",
  columns=[
      ViewColumn(name="c1"), ViewColumn(name="c2"), ViewColumn(name="c3"),
  ],
  query="SELECT * FROM my_table",
)

root.databases["my_db"].schemas["my_schema"].views.create(my_view)

O código então cria a exibição no banco de dados my_db e no esquema my_schema passando o objeto View ao método ViewCollection.create.

Como obter os detalhes de exibição

É possível obter informações sobre uma exibição chamando o método ViewResource.fetch, que retorna um objeto View.

O código no exemplo a seguir obtém um objeto View que representa a exibição my_view:

my_view = root.databases["my_db"].schemas["my_schema"].views["my_view"].fetch()
print(my_view.to_dict())

Exibições de listagem

É possível listar as exibições em um banco de dados especificado usando o método iter. O método retorna um iterador PagedIter de objetos View.

O código no exemplo a seguir lista as exibições cujo nome começa com my no banco de dados my_db e no esquema my_schema:

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(like="my%")
for view_obj in view_list:
  print(view_obj.name)

O código no exemplo a seguir também lista as exibições cujos nomes começam com my, mas usa o parâmetro starts_with em vez de like. Este exemplo também define o parâmetro opcional show_limit=10 para limitar o número de resultados para 10:

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(starts_with="my", show_limit=10)
for view_obj in view_list:
  print(view_obj.name)

Descarte de uma exibição

É possível descartar uma exibição usando o método ViewResource.drop.

O código no exemplo a seguir descarta a exibição my_view:

my_view_res = root.databases["my_db"].schemas["my_schema"].views["my_view"]
my_view_res.drop()