PythonによるSnowflakeデータベース、スキーマ、テーブル、ビューの管理¶

データベースの作成¶

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)

データベースの詳細の取得¶

Database オブジェクトを返す DatabaseResource.fetch メソッドを呼び出すことで、データベースに関する情報を取得できます。

次の例のコードは、 my_db というデータベースに関する情報を取得します。

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

データベースのリスト¶

PagedIter 反復子を返す iter メソッドを使用して、データベースをリストすることができます。

次の例のコードは、名前が 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()

スキーマの作成¶

スキーマを作成するには、まずスキーマ名を指定する Schema オブジェクトを作成します。

次の例のコードは、 my_schema という名前のスキーマを表す 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 データベースにスキーマを作成します。

スキーマの詳細を取得する¶

Schema オブジェクトを返す SchemaResource.fetch メソッドを呼び出すことで、スキーマの情報を取得できます。

次の例のコードは、 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()

テーブルの作成¶

テーブルを作成するには、まずテーブル名、列名、列のデータ型を指定する Table オブジェクトを作成します。

次の例のコードは、指定された列を持つ my_table という名前のテーブルを表す 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 スキーマにテーブルを作成します。

テーブルの詳細の取得¶

Table オブジェクトを返す TableResource.fetch メソッドを呼び出すことで、テーブルに関する情報を取得できます。

次の例のコードは、 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 の動作はべき等であることを意図しています。つまり、メソッドを呼び出す前にテーブルが存在していたかどうかにかかわらず、結果のテーブルオブジェクトは同じになります。

次の例のコードは、 my_table テーブルにデータ型 int の c3 という新しい列を追加してから、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 メソッドを使用すると、1回のトランザクションでテーブルの名前を別のテーブルと入れ替える（交換する）ことができます。詳細については、 SWAP WITH パラメーターの説明を ALTER TABLE でご参照ください。

次の例のコードは、同じデータベースとスキーマで、 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_db データベースおよび 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")

テーブル操作の実行¶

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

イベントテーブルの作成¶

ウェアハウスを作成するには、まず EventTable オブジェクトを作成して、 API Root オブジェクトから EventTableCollection オブジェクトを作成します。 EventTableCollection.create を使用して、新しいイベントテーブルをSnowflakeに追加します。

以下の例のコードは、 my_event_table という名前のイベント・テーブルを表す EventTable オブジェクトを、指定されたパラメーターで作成します。

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に新しいウェアハウスを作成します。

イベントテーブルの詳細を取得する¶

EventTable オブジェクトを返す EventTableResource.fetch メソッドを呼び出すことで、テーブルに関する情報を取得できます。

次の例のコードは、 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())

イベントテーブルの一覧表示¶

EventTable オブジェクトの PagedIter 反復子を返す EventTableCollection.iter メソッドを使用して、ウェアハウスを一覧表示することができます。

次の例のコードは、 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 で始まる名前のイベント・テーブルをリストしていますが、 like の代わりに starts_with パラメーターを使用しています。この例では、結果の数を 10 に制限するために、オプションのパラメーター show_limit=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 オブジェクトを使用して、イベント・テーブルの名前の変更やイベント・テーブルのドロップなど、一般的なイベント・テーブルの操作を行うことができます。

イベント・テーブル・リソースでできる操作を示すために、次の例のコードでは次のようにしています。

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

ビューの作成¶

ビューを作成するには、まずビュー名、列、 SQL クエリステートメントを指定する View オブジェクトを作成します。

次の例のコードは、指定された列と SQL クエリを持つ my_view という名前のビューを表す 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 スキーマにビューを作成します。

ビューの詳細を取得する¶

View オブジェクトを返す ViewResource.fetch メソッドを呼び出すことで、ビューに関する情報を取得できます。

次の例のコードは、 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 で始まるビューをリストアップしていますが、 like の代わりに starts_with パラメーターを使用しています。この例では、結果の数を 10 に制限するために、オプションのパラメーター show_limit=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()