Python을 사용하여 Snowflake 데이터베이스, 스키마 및 테이블 관리하기¶

Python을 사용하여 Snowflake 데이터베이스, 스키마 및 테이블을 관리할 수 있습니다. Snowflake의 데이터 관리 및 작업에 대한 자세한 내용은 데이터베이스, 테이블, 뷰 - 개요 섹션을 참조하십시오.

이 항목의 예제에서는 Snowflake와 연결하고 Snowflake Python API를 사용할 Root 오브젝트를 생성하는 코드를 추가했다고 가정합니다. 자세한 내용은 Snowflake Python API를 사용하여 Snowflake에 연결하기 섹션을 참조하십시오.

다음 예제의 코드에서는 구성 파일에 정의된 연결 매개 변수를 사용하여 Snowflake에 대한 연결을 생성합니다. 해당 코드에서는 결과 Session 오브젝트를 사용하여 API의 유형과 메서드를 사용하기 위해 Root 오브젝트를 생성합니다.

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

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

Copy

데이터베이스 관리하기¶

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

데이터베이스 세부 정보 가져오기¶

Database 오브젝트를 반환하는 DatabaseResource.fetch 메서드를 호출하여 데이터베이스에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드로 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 오브젝트를 생성하십시오.

다음 예제의 코드는 my_schema 라는 스키마를 나타내는 Schema 오브젝트를 생성합니다. 그런 다음 Schema 오브젝트를 SchemaCollection.create 메서드에 전달하고 my_db 데이터베이스에 스키마를 생성하는 방법으로 스키마를 생성합니다.

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 오브젝트를 반환하는 SchemaResource.fetch 메서드를 호출하여 스키마에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 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 오브젝트를 생성하십시오.

다음 예제의 코드는 지정된 열이 있는 my_table 이라는 테이블을 나타내는 Table 오브젝트를 생성합니다. 그런 다음 코드는 Table 오브젝트를 TableCollection.create 메서드에 전달하고 my_db 데이터베이스와 my_schema 스키마에 테이블을 생성하는 방법으로 테이블을 생성합니다.

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 오브젝트를 반환하는 TableResource.fetch 메서드를 호출하여 테이블에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드로 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