Gerenciamento de integrações do Snowflake com Python

É possível usar Python para gerenciar diferentes tipos de integrações no Snowflake.

Pré-requisitos

Os exemplos neste tópico pressupõem que você tenha adicionado código para se conectar ao Snowflake e criar um objeto Root a partir do qual usar o Snowflake Python APIs.

Por exemplo, o seguinte código usa parâmetros de conexão definidos em um arquivo de configuração para criar uma conexão com o Snowflake:

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

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

Usando o objeto Session resultante, o código cria um objeto Root para usar os tipos e métodos de API. Para obter mais informações, consulte Conexão ao Snowflake com o Snowflake Python APIs.

Gerenciamento de integrações de catálogo

É possível gerenciar integrações de catálogo para tabelas Apache Iceberg™ em sua conta. Uma integração de catálogo é um objeto Snowflake nomeado em nível de conta que armazena informações sobre como os metadados da tabela Iceberg são organizados para cenários em que você não usa o Snowflake como o catálogo Iceberg ou quando deseja integrar com o Snowflake Open Catalog. Para obter mais informações, consulte a seção Integração de catálogo em Tabelas Apache Iceberg™.

Nota

ALTER CATALOG INTEGRATION não é compatível atualmente.

O Snowflake Python APIs representa integrações de catálogo com dois tipos separados:

  • CatalogIntegration: Expõe as propriedades de uma integração de catálogo, como seu nome, formato de tabela e configurações de catálogo.

  • CatalogIntegrationResource: Expõe métodos que você pode usar para buscar um objeto CatalogIntegration correspondente e descartar a integração do catálogo.

Criação de uma integração de catálogo

Para criar uma integração de catálogo, primeiro crie um objeto CatalogIntegration e, em seguida, crie um objeto CatalogIntegrationCollection a partir do objeto Root da API. Usando CatalogIntegrationCollection.create, adicione a nova integração de catálogo ao Snowflake.

É possível criar integrações de catálogo em sua conta para os seguintes tipos de catálogos externos do Iceberg.

AWS Glue

O código no exemplo a seguir cria um objeto CatalogIntegration que representa uma integração de catálogo nomeada my_catalog_integration para tabelas Iceberg que usam o AWS Glue com as propriedades especificadas:

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

Loja de objetos

O código no exemplo a seguir cria um objeto CatalogIntegration que representa uma integração de catálogo nomeada my_catalog_integration para tabelas Iceberg que usam um armazenamento de objeto:

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

O código no exemplo a seguir cria um objeto CatalogIntegration que representa uma integração de catálogo nomeada my_catalog_integration para tabelas Iceberg que usam o Open Catalog com as propriedades especificadas:

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

Como obter detalhes de integração de catálogo

É possível obter informações sobre uma integração de catálogo chamando o método CatalogIntegrationResource.fetch, que retorna um objeto CatalogIntegration.

O código no exemplo a seguir obtém informações sobre uma integração de catálogo nomeada my_catalog_integration:

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

Integrações de catálogo de listagem

É possível listar integrações de catálogo usando o método CatalogIntegrationCollection.iter, que retorna um iterador PagedIter de objetos CatalogIntegration.

O código no exemplo a seguir lista integrações de catálogo cujos nomes começam com my e imprime o nome de cada uma:

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

Descarte de uma integração de catálogo

É possível descartar uma integração de catálogo com um objeto CatalogIntegrationResource.

O código no exemplo a seguir busca o objeto de recurso de integração de catálogo my_catalog_integration e, em seguida, descarta a integração de catálogo.

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

Gerenciamento de integrações de notificação

É possível gerenciar integrações de notificação, que são objetos Snowflake que fornecem uma interface entre o Snowflake e serviços de mensagens de terceiros, como serviços de enfileiramento de mensagens em nuvem de terceiros, serviços de e-mail e webhooks. Para obter mais informações, consulte Notificações no Snowflake.

Nota

ALTER NOTIFICATION INTEGRATION não é compatível atualmente.

O Snowflake Python APIs representa integrações de notificação com dois tipos separados:

  • NotificationIntegration: Expõe as propriedades de uma integração de notificação, como nome e configurações de gancho de notificação.

  • NotificationIntegrationResource: Expõe métodos que podem ser usados para buscar um objeto NotificationIntegration correspondente e descartar a integração de notificação.

Criação da integração de notificação

Para criar uma integração de notificação, primeiro crie um objeto NotificationIntegration e, em seguida, crie um objeto NotificationIntegrationCollection a partir do objeto Root da API. Usando NotificationIntegrationCollection.create, adicione a nova integração de notificação ao Snowflake.

É possível criar uma integração de notificação para os seguintes tipos de serviço de mensagem.

E-mail

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_email_notification_integration com as propriedades NotificationEmail especificadas:

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

Webhooks

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_webhook_notification_integration com as propriedades NotificationWebhook especificadas:

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

Tópicos do Amazon SNS (saída)

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_aws_sns_outbound_notification_integration com as propriedades NotificationQueueAwsSnsOutbound especificadas:

from snowflake.core.notification_integration import NotificationIntegration, NotificationQueueAwsSnsOutbound

my_notification_integration = NotificationIntegration(
  name="my_aws_sns_outbound_notification_integration",
  enabled=False,
  notification_hook=NotificationQueueAwsSnsOutbound(
      aws_sns_topic_arn="arn:aws:sns:us-west-1:123456789012:sns-test-topic",
      aws_sns_role_arn="arn:aws:iam::123456789012:role/sns-test-topic",
  )
)

root.notification_integrations.create(my_notification_integration)
Copy

Tópicos do Azure Event Grid (saída)

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_azure_outbound_notification_integration com as propriedades NotificationQueueAzureEventGridOutbound especificadas:

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

Tópicos do Azure Event Grid (entrada)

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_azure_inbound_notification_integration com as propriedades NotificationQueueAzureEventGridInbound especificadas:

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

Tópicos do Google Pub/Sub (saída)

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_gcp_outbound_notification_integration com as propriedades NotificationQueueGcpPubsubOutbound especificadas:

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

Tópicos do Google Pub/Sub (entrada)

O código no exemplo a seguir cria um objeto NotificationIntegration que representa uma integração de notificação nomeada my_gcp_inbound_notification_integration com as propriedades NotificationQueueGcpPubsubInbound especificadas:

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

Como obter detalhes de integração de notificação

É possível obter informações sobre uma integração de notificação chamando o método NotificationIntegrationResource.fetch, que retorna um objeto NotificationIntegration.

O código no exemplo a seguir obtém informações sobre uma integração de notificação chamada my_notification_integration:

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

Integrações de notificação de listagem

É possível listar integrações de notificação usando o método NotificationIntegrationCollection.iter, que retorna um iterador PagedIter de objetos NotificationIntegration.

O código no exemplo a seguir lista integrações de notificação cujos nomes começam com my e imprime o nome de cada uma:

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

Descarte de uma integração de notificação

É possível descartar uma integração de notificação com um objeto NotificationIntegrationResource.

O código no exemplo a seguir busca o objeto de recurso de integração de notificação my_notification_integration e, em seguida, descarta a integração de notificação.

my_notification_integration_res = root.notification_integrations["my_notification_integration"]
my_notification_integration_res.drop()
Copy
Linguagem: Português