使用 Python 管理 Snowflake 数据库、架构、表和视图

您可以使用 Python 来管理 Snowflake 数据库、架构、表和视图。有关在 Snowflake 中管理和使用数据的详细信息，请参阅 数据库、表和视图 – 概览。

先决条件

在本主题中的示例中，假设您已添加了用来连接 Snowflake 和创建 Root 对象以使用 Snowflake Python APIs 的代码。

例如，以下代码使用配置文件中定义的连接参数来创建与 Snowflake 的连接：

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

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

代码可通过生成的 Session 对象创建 Root 对象，从而使用 API 的类型和方法。有关更多信息，请参阅 使用 Snowflake Python APIs 连接到 Snowflake。

管理数据库

您可以在 Snowflake 中管理数据库。Snowflake Python APIs 使用两种不同类型来表示数据库：

  • Database：显示数据库的属性，例如其名称。

  • DatabaseResource：显示可用于提取相应 Database 对象和删除数据库的方法。

主题

创建数据库

您可以通过调用 DatabaseCollection.create 方法并传递一个表示要创建的数据库的 Database 对象来创建数据库。要创建数据库，请先创建一个指定数据库名称的 Database 对象。

以下示例中的代码会创建一个表示数据库 my_db 的 Database 对象，然后通过将该 Database 对象传递给 DatabaseCollection.create 方法来创建该数据库：

from snowflake.core.database import Database

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

获取数据库详细信息

您可以通过调用 DatabaseResource.fetch 方法来获取有关数据库的信息，该方法会返回一个:code:Database 对象。

以下示例中的代码会获取有关名为 my_db 的数据库的信息：

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

列出数据库

您可以使用 iter 方法列出数据库，该方法会返回一个 PagedIter 迭代器。

以下示例中的代码会列出名称以 my 开头的数据库：

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

删除或恢复数据库

您可以使用 DatabaseResource.drop 方法删除数据库或使用 DatabaseResource.undrop 方法恢复数据库。

要演示这些操作，请删除以下示例中的代码，然后恢复最新版本的 my_db 数据库：

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

管理架构

您可以在 Snowflake 中管理架构。架构是一个数据库级别的对象。当您创建或引用一个架构时，您是在其数据库的上下文中执行此操作的。

Snowflake Python APIs 使用两种不同类型来表示架构：

  • Schema：显示架构的属性，例如其名称。

  • SchemaResource：显示可用于提取相应 Schema 对象和删除架构的方法。

主题

创建架构

要创建架构，请先创建一个指定架构名称的 Schema 对象。

以下示例中的代码会创建一个 Schema 对象，用于表示一个名为 my_schema 架构：

from snowflake.core.schema import Schema

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

然后代码通过将该 Schema 对象传递到 SchemaCollection.create 方法在 my_db 数据库中创建该架构。

获取架构详细信息

您可以通过调用 SchemaResource.fetch 方法来获取关于该架构的信息，该方法会返回一个 Schema 对象。

以下示例中的代码会获取表示 my_schema 架构的 Schema 对象：

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

列出架构

您可以使用 iter 方法列出指定数据库中的架构。该方法会返回 Schema 对象的 PagedIter 迭代器。

以下示例中的代码会列出 my_db 数据库中的架构名称：

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

删除或恢复架构

您可以使用 SchemaResource.drop 方法删除架构或使用 SchemaResource.undrop 方法恢复架构。

要演示这些操作，请删除以下示例中的代码，然后恢复最新版本的 my_schema 架构：

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

管理标准表

您可以在 Snowflake 中管理标准表。表是一个架构级对象。当您创建或引用一个表时，您是在其架构的上下文中执行此操作的。

Snowflake Python APIs 使用两种不同类型来表示表：

  • Table：显示表的属性，例如其名称和列。

  • TableResource：显示可用于提取相应 Table 对象、更新表的属性，以及删除表的方法。

主题

创建表

要创建表，请先创建指定表名称、列名称和列数据类型的 Table 对象。

以下示例中的代码会创建一个 Table 对象，表示一个拥有指定列且名为 my_table 的表：

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)

然后代码通过将该 Table 对象传递到 TableCollection.create 方法在 my_db 数据库和 my_schema 架构中创建此表。

获取表详细信息

您可以通过调用 TableResource.fetch 方法来获取关于该表的信息，该方法会返回一个 Table 对象。

以下示例中的代码会获取有关名为 my_table 的表的信息：

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

创建或更改表

您可以设置 Table 对象的属性并将其传递给 TableResource.create_or_alter 方法以创建表（如果表不存在）或者根据表定义更改表（如果表存在）。create_or_alter 的行为旨在实现幂等性，这意味着无论在调用该方法之前表是否存在，生成的表对象都是一样的。

备注

对于您没有明确定义的任何 表 属性，create_or_alter 方法会使用默认值。例如，如果您没有设置 data_retention_time_in_days，则即使表以前以不同的值存在，其值也会默认为 None。

以下示例中的代码会将数据类型为 int 的新列 c3 追加到 my_table 表中，然后在 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)

列出表

您可以使用 iter 方法在指定架构中列出表，该方法会返回一个 Table 对象的 PagedIter 迭代器。

以下示例中的代码会列出名称以 my 开头的表：

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

交换表名称

您可以使用 TableResource.swap_with 方法在单个事务中将表名称与另一个表名称进行交换。有关更多信息，请参阅 ALTER TABLE 中的 SWAP WITH 参数说明。

以下示例中的代码可用于将同一数据库和架构中的 my_table 与 other_table 交换：

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

以下示例中的代码可用于将 my_table``（在 ``my_db 数据库和 my_schema 架构中）与 other_table``（在数据库和 ``other_schema 架构 other_db 中）交换：

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

执行表操作

您可以使用 TableResource 对象执行常见的表操作，例如管理表的 重聚类 以及删除或恢复表。

有关这些表操作的更多信息，请参阅 SQL 命令参考中的 表、视图、序列和用户定义类型命令。

为了演示可以对表资源执行的一些操作，以下示例中的代码会执行以下操作：

  1. 获取 my_db 数据库和 my_schema 架构中的 my_table 表资源对象。

  2. 暂停表的重聚类。

  3. 恢复表的重聚类。

  4. 删除表。

  5. 恢复已删除表的最新版本。

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

管理事件表

您可以管理 Snowflake 事件表，这是一种特殊的数据库表，具有一组预定义的列，Snowflake 可以在其中收集遥测数据。有关更多信息，请参阅 事件表概述。

Snowflake Python APIs 使用两种不同类型来表示事件表：

  • EventTable：显示事件表的属性，例如其名称、数据保留时间、最大数据扩展时间和更改跟踪选项。

  • EventTableResource：显示可用于提取相应 EventTable 对象、重命名事件表，以及删除事件表的方法。

主题

创建事件表

要创建事件表，请先创建一个 EventTable 对象，然后根据 API Root 对象创建 EventTableCollection 对象。使用 EventTableCollection.create 将新事件表添加到 Snowflake。

以下示例中的代码会创建一个 EventTable 对象，表示一个拥有指定参数且名为 my_event_table 的事件表：

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)

该代码会创建一个 EventTableCollection 变量 event_tables，并使用 EventTableCollection.create 在 Snowflake 中创建一个新事件表。

获取事件表详细信息

您可以通过调用 EventTableResource.fetch 方法来获取关于某个事件表的信息，该方法会返回一个 EventTable 对象。

以下示例中的代码会获取有关名为 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())

列出事件表

您可以使用 EventTableCollection.iter 方法列出事件表，该方法会返回 EventTable 对象的 PagedIter 迭代器。

以下示例中的代码会列出 my_db 数据库和 my_schema 架构中名称以 my 开头的事件表，并打印每个表的名称：

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)

以下示例中的代码也会列出名称以 my 开头的事件表，但它使用的是 starts_with 参数而不是 like。此示例还设置了可选参数 show_limit=10，将结果数限制为 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)

执行事件表操作

您可以使用 EventTableResource 对象执行常见的事件表操作 – 例如重命名事件表和删除事件表。

备注

目前仅支持 ALTER TABLE（事件表） 的 RENAME 功能。

默认事件表 SNOWFLAKE.TELEMETRY.EVENTS 不支持 RENAME。

为了演示您可以对事件表资源执行的操作，以下示例中的代码会执行以下操作：

  1. 获取 my_db 数据库和 my_schema 架构中的 my_event_table 事件表资源对象。

  2. 重命名事件表。

  3. 删除事件表。

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

管理视图

您可以在 Snowflake 中管理视图。视图是架构级对象，允许像访问表一样访问查询结果。当您创建或引用一个视图时，您是在其架构的上下文中执行此操作的。

备注

ALTER VIEW 目前不受支持。

Snowflake Python APIs 使用两种不同类型来表示视图：

  • View：显示视图的属性，例如其名称、列和 SQL 查询语句。

  • ViewResource：显示可用于提取相应 View 对象和删除视图的方法。

主题

创建视图

要创建视图，请先创建一个指定视图名称、列和 SQL 查询语句的 View 对象。

以下示例中的代码会创建一个 View 对象，表示一个拥有指定列和 SQL 查询且名为 my_view 的视图：

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)

然后代码通过将该 View 对象传递到 ViewCollection.create 方法在 my_db 数据库和 my_schema 架构中创建此视图。

获取视图详细信息

您可以通过调用 ViewResource.fetch 方法来获取关于某个视图的信息，该方法会返回一个 View 对象。

以下示例中的代码会获取表示 my_view 视图的 View 对象：

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

列表查看次数

您可以使用 iter 方法列出指定数据库中的视图。该方法会返回 View 对象的 PagedIter 迭代器。

以下示例中的代码会列出 my_db 数据库和 my_schema 架构中名称以 my 开头的视图：

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

以下示例中的代码也会列出名称以 my 开头的视图，但它使用的是 starts_with 参数而不是 like。此示例还设置了可选参数 show_limit=10，将结果数限制为 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)

删除视图

您可以使用 ViewResource.drop 方法删除视图。

以下示例中的代码会删除 my_view 视图：

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