Gerenciamento de bancos de dados, esquemas e tabelas Snowflake com Python

Você pode usar Python para gerenciar bancos de dados, esquemas e tabelas 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ê adicionou código para se conectar ao Snowflake e para criar um objeto Root a partir do qual usar a Snowflake Python API.

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)
Copy

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 a Snowflake Python API.

Gerenciamento de bancos de dados

Você pode gerenciar bancos de dados no Snowflake. A Snowflake Python API 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 excluir o banco de dados.

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 chamado my_db e, em seguida, cria o banco de dados passando o objeto Database para o método DatabaseCollection.create:

from snowflake.core import Root
from snowflake.core.database import Database

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

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 chamado my_db:

from snowflake.core import Root
from snowflake.core.database import Database

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

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 cujo nome começa com my:

from snowflake.core import Root

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

Exclusão de um banco de dados

Você pode excluir um banco de dados usando o método DatabaseResource.delete.

O código no exemplo a seguir exclui o banco de dados my_db:

from snowflake.core import Root
from snowflake.core.database import Database

my_db_res = root.databases["my_db"]
my_db_res.delete()
Copy

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.

A Snowflake Python API 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 excluir o esquema.

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 chamado my_schema:

from snowflake.core import Root
from snowflake.core.schema import Schema

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

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:

from snowflake.core import Root
from snowflake.core.schema import Schema

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

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:

from snowflake.core import Root

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

Exclusão de um esquema

Você pode excluir um esquema usando o método SchemaResource.delete.

O código no exemplo a seguir exclui o esquema my_schema:

from snowflake.core import Root
from snowflake.core.schema import Schema

my_schema_res = root.databases["my_db"].schemas["my_schema"]
my_schema_res.delete()
Copy

Gerenciamento de tabelas

Você pode gerenciar tabelas 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.

A Snowflake Python API representa tabelas com dois tipos separados:

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

  • TableResource: expõe métodos que você pode usar para buscar um objeto Table correspondente, atualizar as propriedades da tabela e excluir a tabela.

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 chamada my_table com as colunas especificadas:

from snowflake.core import Root
from snowflake.core.table import Table, TableColumn

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

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 chamada my_table:

from snowflake.core import Root
from snowflake.core.table import Table

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

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

Você pode atualizar as características de uma tabela existente definindo propriedades de um objeto Table que representa uma tabela existente e, em seguida, passar o objeto atualizado para Snowflake com o método TableResource.create_or_update.

Você também pode passar um objeto Table que descreve uma nova tabela quando quiser criá-la.

O código no exemplo a seguir anexa uma nova coluna chamada c3 do tipo de dados int à tabela my_table e, em seguida, atualiza a tabela no Snowflake:

from snowflake.core import Root
from snowflake.core.table import PrimaryKey, Table, TableColumn

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch(deep=True)
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_update(my_table)
Copy

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 cujo nome começa com my:

from snowflake.core import Root

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

Exclusão de uma tabela

Você pode excluir uma tabela usando o método TableResource.delete.

O código no exemplo a seguir exclui a tabela my_table:

from snowflake.core import Root
from snowflake.core.table import Table

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