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

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 de um banco de dados¶

É possível excluir um banco de dados usando o método DatabaseResource.drop.

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

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

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 de um esquema¶

É possível descartar um esquema usando o método SchemaResource.drop.

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

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

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 de create_or_alter é projetado para ser idempotente, o que significa que o objeto de tabela resultante será o mesmo, independentemente de a tabela existir ou não antes de você executar o método.

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)

Descarte de uma tabela¶

É possível descartar uma tabela usando o método TableResource.drop.

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

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

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.

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. Busca o objeto de recurso da tabela de evento my_event_table.

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

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