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

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

先决条件¶

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

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

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

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

Copy

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

管理数据库¶

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

Database：显示数据库的属性，例如其名称。
DatabaseResource：显示可用于提取相应 Database 对象和删除数据库的方法。

创建数据库¶

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

以下示例中的代码会创建一个表示数据库 my_db 的 Database 对象，然后通过将该 Database 对象传递给 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

获取数据库详细信息¶

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

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

列出数据库¶

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

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

from snowflake.core import Root

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

Copy

删除数据库¶

您可以使用 DatabaseResource.delete 方法删除数据库。

以下示例中的代码会删除 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

管理架构¶

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

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

Schema：显示架构的属性，例如其名称。
SchemaResource：显示可用于提取相应 Schema 对象和删除架构的方法。

创建架构¶

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

以下示例中的代码会创建一个 Schema 对象，用于表示一个名为 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

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

获取架构详细信息¶

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

以下示例中的代码会获取表示 my_schema 架构的 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

列出架构¶

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

以下示例中的代码会列出 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

删除架构¶

您可以使用 SchemaResource.delete 方法删除架构。

以下示例中的代码会删除 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

管理表¶

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

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

Table：显示表的属性，例如其名称和列。
TableResource：显示可用于提取相应 Table 对象、更新表的属性，以及删除表的方法。

创建表¶

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

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

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

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

获取表详细信息¶

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

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

创建或更新表¶

您可以设置表示现有表的 Table 对象的属性，然后使用 TableResource.create_or_update 方法将已更新的对象传递给 Snowflake，从而更新现有表的特征。

当您想要创建表时，您还可以传递一个描述新表的 Table 对象。

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

列出表¶

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

以下示例中的代码会列出名称以 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

删除表¶

您可以使用 TableResource.delete 方法删除表。

以下示例中的代码会删除 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