Verwalten von Snowflake-Integrationen mit Python

Sie können Python verwenden, um verschiedene Typen von Integrationen in Snowflake zu verwalten.

Voraussetzungen

Die Beispiele in diesem Thema gehen davon aus, dass Sie Code hinzugefügt haben, um eine Verbindung zu Snowflake herzustellen und ein Root-Objekt zu erstellen, von dem aus Sie die Snowflake Python APIs verwenden können.

Beispielsweise verwendet der folgende Code Verbindungsparameter, die in einer Konfigurationsdatei definiert sind, um eine Verbindung zu Snowflake zu erstellen:

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

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

Unter Verwendung des resultierenden Session-Objekts erstellt der Code ein Root-Objekt, das die Typen und Methoden der API verwendet. Weitere Informationen dazu finden Sie unter Verbinden mit Snowflake mit dem Snowflake Python APIs.

Verwalten von Katalogintegrationen

Sie können Katalogintegrationen für Apache Iceberg™-Tabellen in Ihrem Konto verwalten. Eine Katalogintegration ist ein benanntes Snowflake-Objekt auf Kontoebene, das Informationen darüber speichert, wie die Metadaten Ihrer Iceberg-Tabellen für Szenarien organisiert sind, in denen Sie Snowflake nicht als Iceberg-Katalog verwenden, oder wenn Sie eine Integration mit Snowflake Open Catalog wünschen. Weitere Informationen finden Sie im Abschnitt Katalogintegration auf Apache Iceberg™-Tabellen.

Bemerkung

ALTER CATALOG INTEGRATION wird derzeit nicht unterstützt.

Die Snowflake Python APIs stellt Katalogintegrationen mit zwei verschiedenen Typen dar:

  • CatalogIntegration: Zeigt die Eigenschaften einer Katalogintegration an, z. B. den Namen, das Tabellenformat und die Katalogeinstellungen.

  • CatalogIntegrationResource: Zeigt Methoden an, mit denen Sie ein entsprechendes CatalogIntegration Objekt abrufen und die Katalogintegration löschen können.

Erstellen einer Katalogintegration

Um eine Katalogintegration zu erstellen, erstellen Sie zunächst ein CatalogIntegration-Objekt und dann ein CatalogIntegrationCollection-Objekt aus dem API Root-Objekt. Fügen Sie mit CatalogIntegrationCollection.create die neue Katalogintegration zu Snowflake hinzu.

Sie können in Ihrem Konto Katalogintegrationen für die folgenden Typen von externen Iceberg-Katalogen erstellen.

AWS Glue

Der Code im folgenden Beispiel erstellt ein CatalogIntegration-Objekt, das eine Katalogintegration namens my_catalog_integration für Iceberg-Tabellen darstellt, die AWS Glue mit den angegebenen Eigenschaften verwenden:

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

Objektspeicher

Der Code im folgenden Beispiel erstellt ein CatalogIntegration-Objekt, das eine Katalogintegration namens my_catalog_integration für Iceberg-Tabellen darstellt, die einen Objektspeicher verwenden:

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

Der Code im folgenden Beispiel erstellt ein CatalogIntegration-Objekt, das eine Katalogintegration namens my_catalog_integration für Iceberg-Tabellen darstellt, die Open Catalog mit den angegebenen Eigenschaften verwenden:

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

Abrufen von Details zur Katalogintegration

Sie können Informationen über eine Katalogintegration erhalten, indem Sie die CatalogIntegrationResource.fetch-Methode aufrufen, die ein CatalogIntegration-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft Informationen über eine Katalogintegration namens my_catalog_integration ab:

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

Auflistung von Katalogintegrationen

Sie können Katalogintegrationen mit der CatalogIntegrationCollection.iter-Methode auflisten, die einen PagedIter-Iterator von CatalogIntegration-Objekten zurückgibt.

Der Code im folgenden Beispiel listet Katalogintegrationen auf, deren Name mit my beginnt, und gibt den Namen jeder Integration aus:

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

Eine Katalogintegration löschen

Sie können eine Katalogintegration mit einem CatalogIntegrationResource-Objekt löschen.

Der Code im folgenden Beispiel ruft das my_catalog_integration-Ressourcenobjekt für die Katalogintegration ab und löscht dann die Katalogintegration.

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

Verwalten von Benachrichtigungsintegrationen

Sie können die Integration von Benachrichtigungen verwalten. Dabei handelt es sich um Snowflake-Objekte, die eine Schnittstelle zwischen Snowflake und Messagingdiensten von Drittanbietern wie Cloud-Meldungswarteschlangendienst, E-Mail-Diensten und Webhooks bilden. Weitere Informationen dazu finden Sie unter Benachrichtigungen in Snowflake.

Bemerkung

ALTER NOTIFICATION INTEGRATION wird derzeit nicht unterstützt.

Die Snowflake Python APIs stellt Benachrichtigungsintegrationen mit zwei verschiedenen Typen dar:

  • NotificationIntegration: Zeigt die Eigenschaften einer Integration für Benachrichtigungen an, z. B. den Namen und die Hook-Einstellungen für Benachrichtigungen.

  • NotificationIntegrationResource: Stellt Methoden zur Verfügung, mit denen Sie ein entsprechendes NotificationIntegration-Objekt abrufen und die Benachrichtigungintegration löschen können.

Erstellen einer Benachrichtigungsintegration

Um eine Benachrichtigungsintegration zu erstellen, erstellen Sie zunächst ein NotificationIntegration-Objekt und dann ein NotificationIntegrationCollection-Objekt aus dem API Root-Objekt. Fügen Sie mit NotificationIntegrationCollection.create die neue Benachrichtigungsintegration zu Snowflake hinzu.

Sie können eine Benachrichtigungsintegration für die folgenden Typen von Messagingdiensten erstellen.

Email (E-Mail)

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_email_notification_integration und den angegebenen Eigenschaften NotificationEmail darstellt:

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

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_webhook_notification_integration und den angegebenen Eigenschaften NotificationWebhook darstellt:

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

Amazon SNS-Themen (ausgehend)

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_aws_sns_outbound_notification_integration und den angegebenen Eigenschaften NotificationQueueAwsSnsOutbound darstellt:

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

Azure Event Grid-Themen (ausgehend)

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_azure_outbound_notification_integration und den angegebenen Eigenschaften NotificationQueueAzureEventGridOutbound darstellt:

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

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_azure_inbound_notification_integration und den angegebenen Eigenschaften NotificationQueueAzureEventGridInbound darstellt:

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

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_gcp_outbound_notification_integration und den angegebenen Eigenschaften NotificationQueueGcpPubsubOutbound darstellt:

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

Der Code im folgenden Beispiel erstellt ein NotificationIntegration-Objekt, das eine Benachrichtigungsintegration mit dem Namen my_gcp_inbound_notification_integration und den angegebenen Eigenschaften NotificationQueueGcpPubsubInbound darstellt:

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

Abrufen von Details zur Benachrichtigungsintegration

Sie können Informationen über eine Benachrichtigungsintegration erhalten, indem Sie die NotificationIntegrationResource.fetch-Methode aufrufen, die ein NotificationIntegration-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft Informationen über eine Benachrichtigungsintegration namens my_notification_integration ab:

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

Auflisten von Benachrichtigungsintegrationen

Sie können die Benachrichtigungsintegration mit der NotificationIntegrationCollection.iter-Methode auflisten, die einen PagedIter-Iterator von NotificationIntegration-Objekten zurückgibt.

Der Code im folgenden Beispiel listet Benachrichtigungsintegrationeb auf, deren Name mit my beginnt, und gibt den Namen jeder Integration aus:

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

Löschen einer Benachrichtigungsintegration

Sie können eine Benachrichtigungsintegration mit einem NotificationIntegrationResource-Objekt löschen.

Der Code im folgenden Beispiel ruft das my_notification_integration-Ressourcenobjekt für die BenachrichtigungsIntegration ab und löscht die Benachrichtigungsintegration dann.

my_notification_integration_res = root.notification_integrations["my_notification_integration"]
my_notification_integration_res.drop()
Copy
Sprache: Deutsch