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 메서드를 사용하여 데이터베이스를 삭제할 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스를 삭제합니다.

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

스키마 만들기¶

스키마를 생성하려면 먼저 스키마 이름을 지정하는 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 메서드를 사용하여 스키마를 삭제할 수 있습니다.

다음 예제의 코드는 my_schema 스키마를 삭제합니다.

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

테이블 만들기¶

테이블을 생성하려면 먼저 테이블 이름, 열 이름, 열 데이터 타입을 지정하는 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 의 동작은 멱등성을 갖도록 설계되었으므로, 메서드를 실행하기 전에 테이블이 존재하는지 여부에 관계없이 결과 테이블 오브젝트는 동일합니다.

다음 예제의 코드는 데이터 타입이 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)

테이블 나열하기¶

Table 오브젝트의 PagedIter 반복기를 반환하는 iter 메서드를 사용하여 지정된 스키마의 테이블을 나열할 수 있습니다.

다음 예제의 코드는 이름이 my 로 시작하는 테이블을 나열합니다.

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

테이블 삭제하기¶

TableResource.drop 메서드를 사용하여 테이블을 삭제할 수 있습니다.

다음 예제의 코드는 my_table 테이블을 삭제합니다.

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

이벤트 테이블 만들기¶

이벤트 테이블을 생성하려면 먼저 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 매개 변수를 사용합니다. 이 예제에서는 선택적 매개 변수 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 오브젝트를 사용하여 수행할 수 있습니다.

이벤트 테이블 리소스로 할 수 있는 작업을 보여주기 위해 다음 예제의 코드는 다음을 수행합니다.

1. 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)

그런 다음 코드는 ViewCollection.create 메서드에 View 오브젝트를 전달하여 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 매개 변수를 사용합니다. 이 예제에서는 선택적 매개 변수 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()