Python을 사용하여 Snowflake 통합 관리¶

Snowflake에서는 Python을 사용하여 다양한 유형의 통합을 관리할 수 있습니다.

전제 조건¶

이 항목의 예제에서는 Snowflake와 연결하고 Snowflake Python APIs 을 사용할 수 있는 Root 오브젝트를 생성하는 코드를 추가했다고 가정합니다.

예를 들어, 다음 코드는 구성 파일에 정의된 연결 매개 변수를 사용하여 Snowflake에 대한 연결을 생성합니다.

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

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

Copy

해당 코드에서는 결과 Session 오브젝트를 사용하여 API의 유형과 메서드를 사용하기 위해 Root 오브젝트를 생성합니다. 자세한 내용은 Snowflake Python APIs 을 사용하여 Snowflake에 연결 섹션을 참조하십시오.

카탈로그 통합 관리¶

계정에서 Apache Iceberg™ 테이블에 대한 카탈로그 통합을 관리할 수 있습니다. 카탈로그 통합은 이름이 지정된 계정 수준의 Snowflake 오브젝트로, Snowflake를 Iceberg 카탈로그로 사용하지 않거나 Snowflake Open Catalog 와 통합하려는 경우 시나리오에 대한 Iceberg 테이블 메타데이터가 어떻게 구성되는지에 대한 정보를 저장합니다. 자세한 내용은 Apache Iceberg™ 테이블 의 카탈로그 통합 섹션을 참조하십시오.

참고

ALTER CATALOG INTEGRATION 는 현재 지원되지 않습니다.

그만큼 Snowflake Python APIs 두 가지 별도 유형의 카탈로그 통합을 나타냅니다.

CatalogIntegration: 카탈로그 통합의 이름, 테이블 형식, 카탈로그 설정 등 속성을 표시합니다.
CatalogIntegrationResource: 해당 CatalogIntegration 오브젝트를 가져오고 카탈로그 통합을 삭제하는 데 사용할 수 있는 메서드를 노출합니다.

카탈로그 통합 만들기¶

카탈로그 통합을 생성하려면 먼저 CatalogIntegration 오브젝트를 생성한 다음 API Root 오브젝트에서 CatalogIntegrationCollection 오브젝트를 생성합니다. CatalogIntegrationCollection.create 를 사용하여 Snowflake에 새로운 카탈로그 통합을 추가합니다.

사용자의 계정에서 다음 유형의 외부 Iceberg 카탈로그에 대한 카탈로그 통합을 생성할 수 있습니다.

AWS Glue¶

다음 예제의 코드는 지정된 속성을 가진 AWS Glue를 사용하는 Iceberg 테이블에 대해 my_catalog_integration 라는 이름의 카탈로그 통합을 나타내는 CatalogIntegration 오브젝트를 생성합니다.

from snowflake.core.catalog_integration import CatalogIntegration, Glue

root.catalog_integrations.create(CatalogIntegration(
    name="my_catalog_integration",
    catalog = Glue(
        catalog_namespace="abcd-ns",
        glue_aws_role_arn="arn:aws:iam::123456789012:role/sqsAccess",
        glue_catalog_id="1234567",
    ),
    table_format="ICEBERG",
    enabled=True,
))

Copy

오브젝트 저장소¶

다음 예제의 코드는 지정된 속성을 가진 오브젝트 스토어를 사용하는 Iceberg 테이블에 대해 my_catalog_integration 라는 이름의 카탈로그 통합을 나타내는 CatalogIntegration 오브젝트를 생성합니다.

from snowflake.core.catalog_integration import CatalogIntegration, ObjectStore

root.catalog_integrations.create(CatalogIntegration(
    name="my_catalog_integration",
    catalog = ObjectStore(),
    table_format="ICEBERG",
    enabled=True,
))

Copy

Snowflake Open Catalog¶

다음 예제의 코드는 지정된 속성을 가진 Open Catalog 를 사용하는 Iceberg 테이블에 대해 my_catalog_integration 라는 이름의 카탈로그 통합을 나타내는 CatalogIntegration 오브젝트를 생성합니다.

from snowflake.core.catalog_integration import CatalogIntegration, OAuth, Polaris, RestConfig

root.catalog_integrations.create(CatalogIntegration(
    name="my_catalog_integration",
    catalog = Polaris(
        catalog_namespace="abcd-ns",
        rest_config=RestConfig(
            catalog_uri="https://my_account.snowflakecomputing.com/polaris/api/catalog",
            warehouse="my-warehouse",
        ),
        rest_authentication=OAuth(
            type="OAUTH",
            oauth_client_id="my_client_id",
            oauth_client_secret="my_client_secret",
            oauth_allowed_scopes=["PRINCIPAL_ROLE:ALL"],
        ),
    ),
    table_format="ICEBERG",
    enabled=True,
))

Copy

카탈로그 통합 세부 정보 가져오기¶

CatalogIntegration 오브젝트를 반환하는 CatalogIntegrationResource.fetch 메서드를 호출하여 카탈로그 통합에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 my_catalog_integration 카탈로그 통합에 대한 정보를 가져옵니다.

my_catalog_integration = root.catalog_integrations["my_catalog_integration"].fetch()
print(my_catalog_integration.to_dict())

Copy

카탈로그 통합 목록¶

CatalogIntegration 오브젝트의 PagedIter 반복기를 반환하는 CatalogIntegrationCollection.iter 메서드를 사용하여 카탈로그 통합을 나열할 수 있습니다.

다음 예제의 코드는 이름이 my 로 시작하는 카탈로그 통합을 나열하고 각 계정의 이름을 출력합니다.

catalog_integration_iter = root.catalog_integrations.iter(like="my%")
for catalog_integration_obj in catalog_integration_iter:
  print(catalog_integration_obj.name)

Copy

카탈로그 통합 삭제하기¶

카탈로그 통합을 CatalogIntegrationResource 오브젝트로 삭제할 수 있습니다.

다음 예제의 코드는 my_catalog_integration 카탈로그 통합 리소스 오브젝트를 가져온 다음 카탈로그 통합을 삭제합니다.

my_catalog_integration_res = root.catalog_integrations["my_catalog_integration"]
my_catalog_integration_res.drop()

Copy

알림 통합 관리하기¶

Snowflake 오브젝트인 알림 통합을 관리할 수 있습니다. 알림 통합은 Snowflake와 서드 파티 메시징 서비스(서드 파티 클라우드 메시지 큐 서비스, 이메일 서비스, 웹훅 등) 간의 인터페이스를 제공합니다. 자세한 내용은 Snowflake의 알림 섹션을 참조하십시오.

참고

ALTER NOTIFICATION INTEGRATION 는 현재 지원되지 않습니다.

Snowflake Python APIs 은 두 가지 유형의 알림 통합을 나타냅니다.

NotificationIntegration: 알림 통합의 속성(이름, 알림 후크 설정 등)을 표시합니다.
NotificationIntegrationResource: 해당 NotificationIntegration 오브젝트를 가져오고 알림 통합을 삭제하는 데 사용할 수 있는 메서드를 노출합니다.

알림 통합 만들기¶

알림 통합을 생성하려면 먼저 NotificationIntegration 오브젝트를 생성한 다음 API Root 오브젝트에서 NotificationIntegrationCollection 오브젝트를 생성합니다. NotificationIntegrationCollection.create 를 사용하여 Snowflake에 새 알림 통합을 추가합니다.

다음 유형의 메시징 서비스에 대해 알림 통합을 생성할 수 있습니다.

이메일¶

다음 예제의 코드는 NotificationEmail 속성이 지정되고 이름이 my_email_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationEmail, NotificationIntegration

my_notification_integration = NotificationIntegration(
  name="my_email_notification_integration",
  notification_hook=NotificationEmail(
      allowed_recipients=["test1@snowflake.com", "test2@snowflake.com"],
      default_recipients=["test1@snowflake.com"],
      default_subject="test default subject",
  ),
  enabled=True,
)

root.notification_integrations.create(my_notification_integration)

Copy

웹후크¶

다음 예제의 코드는 NotificationWebhook 속성이 지정되고 이름이 my_webhook_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationIntegration, NotificationWebhook

my_notification_integration = NotificationIntegration(
  name="my_webhook_notification_integration",
  enabled=False,
  notification_hook=NotificationWebhook(
      webhook_url=webhook_url,
      webhook_secret=WebhookSecret(
          # This example assumes that this secret already exists
          name="mySecret".upper(), database_name=database, schema_name=schema
      ),
      webhook_body_template=webhook_template,
      webhook_headers=webhook_headers,
  ),
)

root.notification_integrations.create(my_notification_integration)

Copy

Azure Event Grid 항목(아웃바운드)¶

다음 예제의 코드는 NotificationQueueAzureEventGridOutbound 속성이 지정되고 이름이 my_azure_outbound_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationIntegration, NotificationQueueAzureEventGridOutbound

my_notification_integration = NotificationIntegration(
  name="my_azure_outbound_notification_integration",
  enabled=False,
  notification_hook=NotificationQueueAzureEventGridOutbound(
      azure_event_grid_topic_endpoint="https://fake.queue.core.windows.net/api/events",
      azure_tenant_id="fake.onmicrosoft.com",
  )
)

root.notification_integrations.create(my_notification_integration)

Copy

Azure Event Grid 항목(인바운드)¶

다음 예제의 코드는 NotificationQueueAzureEventGridInbound 속성이 지정되고 이름이 my_azure_inbound_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationIntegration, NotificationQueueAzureEventGridInbound

my_notification_integration = NotificationIntegration(
  name="my_azure_inbound_notification_integration",
  enabled=False,
  notification_hook=NotificationQueueAzureEventGridInbound(
      azure_storage_queue_primary_uri="https://fake.queue.core.windows.net/snowapi_queue",
      azure_tenant_id="fake.onmicrosoft.com",
  ),
)

root.notification_integrations.create(my_notification_integration)

Copy

Google Pub/Sub 항목(아웃바운드)¶

다음 예제의 코드는 NotificationQueueGcpPubsubOutbound 속성이 지정되고 이름이 my_gcp_outbound_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationIntegration, NotificationQueueGcpPubsubOutbound

my_notification_integration = NotificationIntegration(
  name="my_gcp_outbound_notification_integration",
  enabled=False,
  notification_hook=NotificationQueueGcpPubsubOutbound(
      gcp_pubsub_topic_name="projects/fake-project-name/topics/pythonapi-test",
  )
)

root.notification_integrations.create(my_notification_integration)

Copy

Google Pub/Sub 항목(인바운드)¶

다음 예제의 코드는 NotificationQueueGcpPubsubInbound 속성이 지정되고 이름이 my_gcp_inbound_notification_integration 인 알림 통합을 나타내는 NotificationIntegration 오브젝트를 생성합니다

from snowflake.core.notification_integration import NotificationIntegration, NotificationQueueGcpPubsubInbound

my_notification_integration = NotificationIntegration(
  name="my_gcp_inbound_notification_integration",
  enabled=True,
  notification_hook=NotificationQueueGcpPubsubInbound(
      gcp_pubsub_subscription_name="projects/fake-project-name/subscriptions/sub-test",
  )
)

root.notification_integrations.create(my_notification_integration)

Copy

알림 통합 세부 정보 가져오기¶

NotificationIntegration 오브젝트를 반환하는 NotificationIntegrationResource.fetch 메서드를 호출하여 알림 통합에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 my_notification_integration 알림 통합에 대한 정보를 가져옵니다.

my_notification_integration = root.notification_integrations["my_notification_integration"].fetch()
print(my_notification_integration.to_dict())

Copy

알림 통합 나열하기¶

NotificationIntegration 오브젝트의 PagedIter 반복기를 반환하는 NotificationIntegrationCollection.iter 메서드를 사용하여 알림 통합을 나열할 수 있습니다.

다음 예제의 코드는 이름이 my 로 시작하는 알림 통합을 나열하고 각 계정의 이름을 출력합니다.

notification_integration_iter = root.notification_integrations.iter(like="my%")
for notification_integration_obj in notification_integration_iter:
  print(notification_integration_obj.name)

Copy

알림 통합 삭제하기¶

알림 통합을 NotificationIntegrationResource 오브젝트로 삭제할 수 있습니다.

다음 예제의 코드는 my_notification_integration 알림 통합 리소스 오브젝트를 가져온 다음 알림 통합을 삭제합니다.

my_notification_integration_res = root.notification_integrations["my_notification_integration"]
my_notification_integration_res.drop()

Copy

Python을 사용하여 Snowflake 통합 관리¶

전제 조건¶

카탈로그 통합 관리¶

카탈로그 통합 만들기¶

AWS Glue¶

오브젝트 저장소¶

Snowflake Open Catalog¶

카탈로그 통합 세부 정보 가져오기¶

카탈로그 통합 목록¶

카탈로그 통합 삭제하기¶

알림 통합 관리하기¶

알림 통합 만들기¶

이메일¶

웹후크¶

Amazon SNS 항목(아웃바운드)¶

Azure Event Grid 항목(아웃바운드)¶

Azure Event Grid 항목(인바운드)¶

Google Pub/Sub 항목(아웃바운드)¶

Google Pub/Sub 항목(인바운드)¶

알림 통합 세부 정보 가져오기¶

알림 통합 나열하기¶

알림 통합 삭제하기¶