Tutorial: Snowflake Native SDK for Connectors Java-Konnektor-Vorlage¶

Einführung¶

Willkommen bei unserem Tutorial zur Verwendung einer Konnektorvorlage mit Snowflake Native SDK for Connectors. Diese Anleitung hilft Ihnen bei der Einrichtung einer einfachen nativen Konnektoranwendung.

In diesem Tutorial lernen Sie Folgendes:

• Bereitstellen einer nativen Konnektoranwendung

• Konfigurieren einer Konnektorvorlage für die Datenaufnahme (Ingestion)

• Anpassen einer Konnektorvorlage an Ihre Anforderungen

Die Vorlage enthält verschiedene hilfreiche Kommentare im Code, die es Ihnen erleichtern, bestimmte Dateien zu finden, die geändert werden müssen. Achten Sie auf Kommentare mit den folgenden Stichwörtern, diese werden Sie leiten und Ihnen helfen, Ihren eigenen Konnektor zu implementieren:

• TODO

• TODO: HINT

• TODO: IMPLEMENT ME

Bevor Sie mit diesem Tutorial beginnen, sollten Sie sich mit den folgenden empfohlenen Inhalten vertraut machen:

• Snowflake Native SDK für Konnektoren

• Tutorial: Snowflake Native SDK for Connectors Beispiel Java-Konnektor

Bereiten Sie Ihre lokale Umgebung vor¶

Bevor Sie fortfahren, müssen Sie sicherstellen, dass alle erforderlichen Programme auf Ihrem Computer installiert sind, und die Konnektorvorlage klonen.

snow init <project_dir> \
  --template-source https://github.com/snowflakedb/connectors-native-sdk \
  --template templates/connectors-native-sdk-template

$ snow init my_connector \
    --template-source https://github.com/snowflakedb/connectors-native-sdk \
    --template templates/connectors-native-sdk-template

Name of the application instance which will be created in Snowflake [connectors-native-sdk-template]: MY_CONNECTOR
Name of the schema in which the connector files stage will be created [TEST_SCHEMA]:
Name of the stage used to store connector files in the application package [TEST_STAGE]: CUSTOM_STAGE_NAME
Initialized the new project in my_connector

Erstellen, Bereitstellen und Bereinigen des Konnektors¶

Die Vorlage kann sofort eingesetzt werden, noch vor jeglicher Änderung. In den folgenden Abschnitten erfahren Sie, wie Sie den Konnektor erstellen, bereitstellen und installieren.

Schritt „Voraussetzungen“¶

Direkt nach der Installation befindet sich der Konnektor in der Assistentenphase. Diese Phase besteht aus einigen Schritten, die den Endbenutzer durch alle notwendigen Konfigurationen führen.

Der erste Schritt ist der Schritt „Voraussetzungen“. Dieser ist optional und möglicherweise nicht für jeden Konnektor erforderlich. Bei den Voraussetzungen handelt es sich in der Regel um Aktionen, die der Benutzer außerhalb der Anwendung durchführen muss, z. B. Ausführen von Abfragen im SQL-Arbeitsblatt, Durchführen von Konfigurationen auf der Seite des Quellsystems usw.

Weitere Informationen zu den Voraussetzungen finden Sie unter: Voraussetzungen

Der Inhalt jeder Voraussetzung wird direkt aus der Tabelle STATE.PREREQUISITES abgerufen, die sich innerhalb des Konnektors befindet. Sie können über das Skript setup.sql angepasst werden. Beachten Sie jedoch, dass das Skript setup.sql bei jeder Installation, bei jedem Upgrade und bei jedem Downgrade der Anwendung ausgeführt wird. Die Einfügungen müssen idempotent sein, daher empfiehlt sich die Verwendung einer Merge-Abfrage wie im folgenden Beispiel:

MERGE INTO STATE.PREREQUISITES AS dest
USING (SELECT * FROM VALUES
           ('1',
            'Sample prerequisite',
            'Prerequisites can be used to notice the end user of the connector about external configurations. Read more in the SDK documentation below. This content can be modified inside `setup.sql` script',
            'https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/prerequisites',
            NULL,
            NULL,
            1
           )
) AS src (id, title, description, documentation_url, learnmore_url, guide_url, position)
ON dest.id = src.id
WHEN NOT MATCHED THEN
    INSERT (id, title, description, documentation_url, learnmore_url, guide_url, position)
    VALUES (src.id, src.title, src.description, src.documentation_url, src.learnmore_url, src.guide_url, src.position);

Schritt „Konnektorkonfiguration“¶

Der nächste Schritt der Assistentenphase ist der Konnektorkonfigurationsschritt. In diesem Schritt können Sie die für den Konnektor erforderlichen Datenbankobjekte und Berechtigungen konfigurieren. In diesem Schritt können Sie die folgenden Konfigurationseigenschaften festlegen:

• warehouse

• operational_warehouse

• cortex_warehouse

• destination_database

• destination_schema

• global_schedule

• data_owner_role

• cortex_user_role

• agent_username

• agent_role

Wenn Sie weitere kundenspezifische Eigenschaften benötigen, können Sie diese in einem der nächsten Schritte der Assistentenphase konfigurieren. Weitere Informationen zu den einzelnen Eigenschaften finden Sie unter: Konnektorkonfiguration

Darüber hinaus zeigt die in der Vorlage enthaltene Streamlit-Komponente (streamlit/wizard/connector_config.py), wie Sie Native Apps Permission SDK triggern und Berechtigungen beim Endbenutzer anfordern können. Solange die verfügbaren Eigenschaften den Anforderungen des Konnektors genügen, müssen Sie keine der Backend-Klassen überschreiben, obwohl dies in den weiteren Schritten der Konfiguration auf die gleiche Weise wie bei den Komponenten möglich ist.

Weitere Informationen zu interne Prozeduren und Java-Objekten finden Sie unter: Referenz zur Konnektorkonfiguration

Das bereitgestellte Streamlit-Beispiel ermöglicht das Anfordern von Berechtigungen auf Kontoebene, die in der manifest.yml-Datei – CREATE DATABASE und EXECUTE TASKS konfiguriert sind. Außerdem wird dem Benutzer die Möglichkeit gegeben, über das Berechtigungs-SDK-Popup eine Warehouse-Referenz anzugeben.

In der Vorlage wird der Benutzer lediglich aufgefordert, Werte für destination_database und destination_schema anzugeben. Ein TODO-Kommentar in streamlit/wizard/connector_configuration.py enthält jedoch kommentierten Code, der wiederverwendet werden kann, um weitere Eingabefelder im Streamlit-UI anzuzeigen.

# TODO: Here you can add additional fields in connector configuration.
# For example:
st.subheader("Operational warehouse")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="operational_warehouse", label_visibility="collapsed")
st.caption("Name of the operational warehouse to be used")

Schritt „Verbindungskonfiguration“¶

Der nächste Schritt der Assistentenphase ist der Verbindungskonfigurationsschritt. In diesem Schritt kann der Benutzer die Parameter für die externe Konnektivität des Konnektors konfigurieren. Diese Konfiguration kann Bezeichner von Objekten wie Geheimnissen, Integrationen usw. enthalten.

Da diese Informationen je nach Quellsystem für die vom Konnektor aufgenommenen Daten unterschiedlich ist, ist dies die erste Stelle, an der größere Anpassungen im Quellcode vorgenommen werden müssen.

Weitere Informationen zur Verbindungskonfiguration finden Sie unter:

• Verbindungskonfiguration

• Referenz zur Verbindungskonfiguration

Beginnend mit der Streamlit-UI-Seite (streamlit/wizard/connection_config.py) müssen Sie Texteingaben für alle benötigten Parameter hinzufügen. Eine Beispiel-Texteingabe ist für Sie implementiert, und wenn Sie den Code in dieser Datei durchsuchen, finden Sie eine TODO mit kommentiertem Code für ein neues Feld.

# TODO: Additional configuration properties can be added to the UI like this:
st.subheader("Additional connection parameter")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="additional_connection_property", label_visibility="collapsed")
st.caption("Some description of the additional property")

Nachdem die Eigenschaften zum Formular hinzugefügt wurden, müssen sie an die Backend-Schicht des Konnektors übergeben werden. Dazu müssen zwei zusätzliche Stellen in den Streamlit-Dateien geändert werden. Die erste ist die Funktion finish_config in der Datei streamlit/wizard/connection_config.py. Der Status der neu hinzugefügten Texteingaben muss hier gelesen werden. Außerdem kann er bei Bedarf validiert und dann an die Funktion set_connection_configuration übergeben werden.

Wenn zum Beispiel additional_connection_property hinzugefügt wurde, würde die Stelle nach den Änderungen wie folgt aussehen:

def finish_config():
try:
    # TODO: If some additional properties were specified they need to be passed to the set_connection_configuration function.
    # The properties can also be validated, for example, check whether they are not blank strings etc.
    response = set_connection_configuration(
        custom_connection_property=st.session_state["custom_connection_property"],
        additional_connection_property=st.session_state["additional_connection_property"],
    )

# rest of the method without changes

Dann muss die Funktion set_connection_configuration bearbeitet werden, die Sie in der Datei streamlit/native_sdk_api/connection_config.py finden. Diese Funktion ist ein Proxy zwischen Streamlit-UI und der zugrunde liegenden SQL-Prozedur, die ein Einstiegspunkt in das Backend des Konnektors ist.

def set_connection_configuration(custom_connection_property: str, additional_connection_property: str):
    # TODO: this part of the code sends the config to the backend so all custom properties need to be added here
    config = {
        "custom_connection_property": escape_identifier(custom_connection_property),
        "additional_connection_property": escape_identifier(additional_connection_property),
    }

    return call_procedure(
        "PUBLIC.SET_CONNECTION_CONFIGURATION",
        [variant_argument(config)]
    )

Danach wird die neue Eigenschaft in der internen Konnektortabelle gespeichert, die die Konfiguration enthält. Dies ist jedoch nicht das Ende der möglichen Anpassungen. Einige Backend-Komponenten können ebenfalls angepasst werden. Achten Sie auf die folgenden Kommentare im Code, um sie zu finden:

• TODO: IMPLEMENT ME connection configuration validate

• TODO: IMPLEMENT ME connection callback

• TODO: IMPLEMENT ME test connection

Der Validierungsteil ermöglicht eine zusätzliche Validierung der Daten, die über die UI erhaltenen wurden. Er kann die Daten auch umwandeln, z. B. die Groß- und Kleinschreibung ändern, die bereitgestellten Daten kürzen, oder prüfen, ob Objekte mit den angegebenen Namen tatsächlich in Snowflake existieren.

Der Verbindungs-Callback ist ein Teil, mit dem Sie zusätzliche Operationen auf Basis der Konfiguration ausführen können, z. B. Änderungsprozeduren, bei denen Integrationen für den externen Zugriff verwendet werden müssen, unter Verwendung einer Lösung, die unter Referenz für die Einrichtung der externen Integration beschrieben wird.

Die Testverbindung ist eine letzte Komponente der Verbindungskonfiguration. Sie prüft, ob die Verbindung zwischen dem Konnektor und dem Quellsystem hergestellt werden kann.

Weitere Informationen zu diesen internen Komponenten finden Sie unter:

• Verbindungskonfiguration

• Referenz zur Verbindungskonfiguration

Beispielimplementierungen könnten wie folgt aussehen:

public class TemplateConfigurationInputValidator implements ConnectionConfigurationInputValidator {

    private static final String ERROR_CODE = "INVALID_CONNECTION_CONFIGURATION";

    @Override
    public ConnectorResponse validate(Variant config) {
      // TODO: IMPLEMENT ME connection configuration validate: If the connection configuration input
      // requires some additional validation this is the place to implement this logic.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      var integrationCheck = checkParameter(config, INTEGRATION_PARAM, false);
      if (!integrationCheck.isOk()) {
        return integrationCheck;
      }

      var secretCheck = checkParameter(config, SECRET_PARAM, true);
      if (!secretCheck.isOk()) {
        return ConnectorResponse.error(ERROR_CODE);
      }

      return ConnectorResponse.success();
    }
}

public class TemplateConnectionConfigurationCallback implements ConnectionConfigurationCallback {

    private static final String[] EXTERNAL_SOURCE_PROCEDURE_SIGNATURES = {
        asVarchar(format("%s.%s()", PUBLIC_SCHEMA, TEST_CONNECTION_PROCEDURE)),
        asVarchar(format("%s.%s(VARIANT)", PUBLIC_SCHEMA, FINALIZE_CONNECTOR_CONFIGURATION_PROCEDURE)),
        asVarchar(format("%s.%s(NUMBER, STRING)", PUBLIC_SCHEMA, WORKER_PROCEDURE))
      };

    private final Session session;

    public TemplateConnectionConfigurationCallback(Session session) {
      this.session = session;
    }

    @Override
    public ConnectorResponse execute(Variant config) {
      // TODO: If you need to alter some procedures with external access you can use
      // configureProcedure method or implement a similar method on your own.
      // TODO: IMPLEMENT ME connection callback: Implement the custom logic of changes in application
      // to be done after connection configuration, like altering procedures with external access.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      var response = configureProceduresWithReferences();
      if (response.isNotOk()) {
         return response;
      }
      return ConnectorResponse.success();
    }

    private ConnectorResponse configureProceduresWithReferences() {
      return callProcedure(
        session,
        PUBLIC_SCHEMA,
        SETUP_EXTERNAL_INTEGRATION_WITH_NAMES_PROCEDURE,
        EXTERNAL_SOURCE_PROCEDURE_SIGNATURES);
    }
}

public class TemplateConnectionValidator {

    private static final String ERROR_CODE = "TEST_CONNECTION_FAILED";

    public static Variant testConnection(Session session) {
      // TODO: IMPLEMENT ME test connection: Implement the custom logic of testing the connection to
      // the source system here. This usually requires connection to some webservice or other external
      // system. It is suggested to perform only the basic connectivity validation here.
      // If that's the case then this procedure must be altered in TemplateConnectionConfigurationCallback first.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/connection_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/connection_configuration
      return test().toVariant();
    }

    private static ConnectorResponse test() {
      try {
        var response = SourceSystemHttpHelper.testEndpoint();

        if (isSuccessful(response.statusCode())) {
          return ConnectorResponse.success();
        } else {
          return ConnectorResponse.error(ERROR_CODE, "Connection to source system failed");
        }
      } catch (Exception exception) {
        return ConnectorResponse.error(ERROR_CODE, "Test connection failed");
      }
    }
}

Schritt „Konfiguration abschließen“¶

Der letzte Schritt der Assistentenphase ist die Fertigstellung der Verbindungskonfiguration. Dieser Schritt umfasst mehrere Teilschritte:

1. Ermöglicht dem Benutzer die Angabe zusätzlicher Konfigurationen, die der Konnektor benötigt

2. Erstellt die Senkendatenbank, ein Schema und bei Bedarf zusätzliche Tabellen und Ansichten für die aufgenommenen Daten

3. Initialisiert interne Komponenten wie den Scheduler und den Task Reactor

Weitere Informationen zum Abschluss der Konfiguration finden Sie unter:

• Konfiguration abschließen

• Referenz zum Konfigurationsabschluss

Weitere Informationen über den Task Reactor und die Zeitplanung finden Sie unter:

• Task Reactor

• SQL-Referenz zum Task Reactor

• Datenaufnahmeplan

• Referenz zu Datenaufnahmeplan

Ähnlich wie bei der Verbindungskonfiguration kann die Anpassung mit dem Streamlit-UI gestartet werden. Die Datei streamlit/wizard/finalize_config.py enthält ein Formular mit einer Beispieleigenschaft. Weitere Eigenschaften können je nach Bedarf des Konnektor hinzugefügt werden. Um eine weitere Eigenschaft hinzuzufügen, suchen Sie nach einem TODO-Kommentar, der Beispielcode für das Hinzufügen einer neuen Eigenschaft in der genannten Datei enthält.

# TODO: Here you can add additional fields in finalize connector configuration.
# For example:
st.subheader("Some additional property")
input_col, _ = st.columns([2, 1])
with input_col:
    st.text_input("", key="some_additional_property", label_visibility="collapsed")
st.caption("Description of some new additional property")

Nachdem Sie die Texteingabe für eine neue Eigenschaft hinzugefügt haben, muss diese an die Backend-Seite übergeben werden. Ändern Sie dazu die Funktion finalize_configuration in derselben Datei:

def finalize_configuration():
    try:
        st.session_state["show_main_error"] = False
        # TODO: If some additional properties were introduced, they need to be passed to the finalize_connector_configuration function.
        response = finalize_connector_configuration(
            st.session_state.get("custom_property"),
            st.session_state.get("some_additional_property")
        )

Öffnen Sie danach die Datei streamlit/native_sdk_api/finalize_config.py und fügen Sie die neue Eigenschaft zur folgenden Funktion hinzu:

def finalize_connector_configuration(custom_property: str, some_additional_property: str):
    # TODO: If some custom properties were configured, then they need to be specified here and passed to the FINALIZE_CONNECTOR_CONFIGURATION procedure.
    config = {
        "custom_property": custom_property,
        "some_additional_property": some_additional_property,
    }
    return call_procedure(
        "PUBLIC.FINALIZE_CONNECTOR_CONFIGURATION",
        [variant_argument(config)]
    )

Ähnlich wie bei der Verbindungskonfiguration ermöglicht auch dieser Schritt die Anpassung verschiedener Backend-Komponenten, die Sie über die folgenden Kommentare im Quellcode finden:

• TODO: IMPLEMENT ME validate source

• TODO: IMPLEMENT ME finalize internal

Im Teilschritt zur Validierung der Quelle geht es um die Durchführung anspruchsvollerer Validierungen der Quellsysteme. Wenn bei der vorherigen Testverbindung nur geprüft wurde, ob eine Verbindung hergestellt werden kann, könnte die Quellenvalidierung den Zugriff auf bestimmte Daten im System prüfen, z. B. durch Extraktion eines einzelnen Datensatzes.

Die interne Finalisierung ist eine interne Prozedur, die für die Initialisierung von Task Reactor und Scheduler und das Erstellen einer Senkendatenbank und der notwendigen verschachtelten Objekte verantwortlich ist. Sie kann auch verwendet werden, um die während des Finalisierungsschritts bereitgestellte Konfiguration zu speichern (diese Konfiguration wird standardmäßig nicht gespeichert).

Weitere Informationen zu den internen Komponenten finden Sie unter:

• Konfiguration abschließen

• Referenz zum Konfigurationsabschluss

Zusätzlich können Eingaben über die Schnittstelle FinalizeConnectorInputValidator validiert und an den Finalisierungs-Handler übergeben werden (siehe TemplateFinalizeConnectorConfigurationCustomHandler). Weitere Informationen zur Verwendung von Buildern finden Sie unter: Anpassung von gespeicherten Prozeduren und Handlern.

Die Implementierung der Quellenvalidierung könnte beispielsweise wie folgt aussehen:

public class SourceSystemAccessValidator implements SourceValidator {

    @Override
    public ConnectorResponse validate(Variant variant) {
      // TODO: IMPLEMENT ME validate source: Implement the custom logic of validating the source
      // system. In some cases this can be the same validation that happened in
      // TemplateConnectionValidator.
      // However, it is suggested to perform more complex validations, like specific access rights to
      // some specific resources here.
      // See more in docs:
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/reference/finalize_configuration_reference
      // https://docs.snowflake.com/developer-guide/native-apps/connector-sdk/flow/finalize_configuration
      var finalizeProperties = Configuration.fromCustomConfig(variant);

      var httpResponse = SourceSystemHttpHelper.validateSource(finalizeProperties.get("custom_property"));
      return prepareConnectorResponse(httpResponse.statusCode());
    }

    private ConnectorResponse prepareConnectorResponse(int statusCode) {
      switch (statusCode) {
        case 200:
          return ConnectorResponse.success();
        case 401:
          return ConnectorResponse.error("Unauthorized error");
        case 404:
          return ConnectorResponse.error("Not found error");
        default:
          return ConnectorResponse.error("Unknown error");
      }
    }
}

Ressourcen erstellen¶

Nachdem die Assistentenphase abgeschlossen ist, ist der Konnektor bereit, mit der Aufnahme von Daten zu beginnen. Aber zuerst müssen die Ressourcen implementiert und konfiguriert werden. Eine Ressource ist eine Abstraktion, die eine bestimmte Menge von Daten im Quellsystem beschreibt, z. B. eine Tabelle, einen Endpunkt oder eine Datei.

Verschiedene Quellsysteme benötigen möglicherweise unterschiedliche Informationen zu einer Ressource. Aus diesem Grund muss eine Ressourcendefinition entsprechend den spezifischen Anforderungen angepasst werden. Um dies zu tun, rufen Sie die Datei streamlit/daily_use/data_sync_page.py auf. Dort finden Sie einen TODO-Kommentar zum Hinzufügen von Texteingaben für Ressourcenparameter. Die Ressourcenparameter sollten die Identifizierung und das Abrufen von Daten aus dem Quellsystem ermöglichen. Diese Parameter können dann während der Datenaufnahme extrahiert werden:

# TODO: specify all the properties needed to define a resource in the source system. A subset of those properties should allow for a identification of a single resource, be it a table, endpoint, repository or some other data storage abstraction
st.text_input(
    "Resource name",
    key="resource_name",
)
st.text_input(
    "Some resource parameter",
    key="some_resource_parameter"
)

Sobald alle erforderlichen Eigenschaften zum Formular hinzugefügt wurden, können sie an die Backend-Seite übergeben werden. Zunächst muss der Zustand der Textfelder extrahiert und auf API-Ebene an die Methode queue_resource in der streamlit/daily_use/data_sync_page.py-Datei übergeben werden:

def queue_resource():
    # TODO: add additional properties here and pass them to create_resource function
    resource_name = st.session_state.get("resource_name")
    some_resource_parameter = st.session_state.get("some_resource_parameter")

    if not resource_name:
        st.error("Resource name cannot be empty")
        return

    result = create_resource(resource_name, some_resource_parameter)
    if result.is_ok():
        st.success("Resource created")
    else:
        st.error(result.get_message())

Dann muss die Funktion create_resource aus der Datei streamlit/native_sdk_api/resource_management.py aktualisiert werden:

def create_resource(resource_name, some_resource_parameter):
    ingestion_config = [{
        "id": "ingestionConfig",
        "ingestionStrategy": "INCREMENTAL",
        # TODO: HINT: scheduleType and scheduleDefinition are currently not supported out of the box, due to globalSchedule being used. However, a custom implementation of the scheduler can use those fields. They need to be provided becuase they are mandatory in the resourceDefinition.
        "scheduleType": "INTERVAL",
        "scheduleDefinition": "60m"
    }]
    # TODO: HINT: resource_id should allow identification of a table, endpoint etc. in the source system. It should be unique.
    resource_id = {
        "resource_name": resource_name,
    }
    id = f"{resource_name}_{random_suffix()}"

    # TODO: if you specified some additional resource parameters then you need to put them inside resource metadata:
    resource_metadata = {
        "some_resource_parameter": some_resource_parameter
    }

    return call_procedure("PUBLIC.CREATE_RESOURCE",
                          [
                              varchar_argument(id),
                              variant_argument(resource_id),
                              variant_list_argument(ingestion_config),
                              varchar_argument(id),
                              "true",
                              variant_argument(resource_metadata)
                          ])

Datenaufnahme (Ingestion)¶

Um die Aufnahme von Daten durchzuführen, müssen Sie eine Klasse implementieren, die die Verbindung mit dem Quellsystem herstellt und die Daten auf der Grundlage der Ressourcenkonfiguration abruft. Die Module Scheduler und Task Reactor sorgen dafür, dass Datenaufnahmeaufgaben ausgelöst und zur Warteschlange hinzugefügt werden.

Die Datenaufnahmelogik wird von der Klasse TemplateIngestion aufgerufen. Suchen Sie den Kommentar TODO: IMPLEMENT ME ingestion im Code und ersetzen Sie die zufällige Datengenerierung durch den Datenabruf aus dem Quellsystem. Wenn Sie der Ressourcendefinition kundenspezifische Eigenschaften hinzugefügt haben, können diese aus den internen Konnektortabellen mithilfe von ResourceIngestionDefinitionRepository und den in TemplateWorkItem verfügbaren Eigenschaften abgerufen werden:

• resourceIngestionDefinitionId

• ingestionConfigurationId

Ein Beispiel für den Abruf von Daten von einem Webservice könnte folgendermaßen aussehen:

public final class SourceSystemHttpHelper {

  private static final String DATA_URL = "https://source_system.com/data/%s";
  private static final SourceSystemHttpClient sourceSystemClient = new SourceSystemHttpClient();
  private static final ObjectMapper objectMapper = new ObjectMapper();

  private static List<Variant> fetchData(String resourceId) {
    var response = sourceSystemClient.get(String.format(url, resourceId));
    var body = response.body();

    try {
        return Arrays.stream(objectMapper.readValue(body, Map[].class))
              .map(Variant::new)
              .collect(Collectors.toList());
    } catch (JsonProcessingException e) {
      throw new RuntimeException("Cannot parse json", e);
    }
  }
}

public class SourceSystemHttpClient {

  private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(15);

  private final HttpClient client;
  private final String secret;

  public SourceSystemHttpClient() {
    this.client = HttpClient.newHttpClient();
    this.secret =
        SnowflakeSecrets.newInstance()
            .getGenericSecretString(ConnectionConfiguration.TOKEN_NAME);
  }

  public HttpResponse<String> get(String url) {
    var request =
        HttpRequest.newBuilder()
            .uri(URI.create(url))
            .GET()
            .header("Authorization", format("Bearer %s", secret))
            .header("Content-Type", "application/json")
            .timeout(REQUEST_TIMEOUT)
            .build();

    try {
      return client.send(request, HttpResponse.BodyHandlers.ofString());
    } catch (IOException | InterruptedException ex) {
      throw new RuntimeException(format("HttpRequest failed: %s", ex.getMessage()), ex);
    }
  }
}

Lebenszyklus von Ressourcen verwalten¶

Sobald die Logik zur Erstellung von Ressourcen und deren Aufnahme implementiert ist, können Sie deren Lebenszyklus verwalten, indem Sie die folgenden Prozeduren aufrufen:

1. PUBLIC.ENABLE_RESOURCE() aktiviert eine bestimmte Ressource, was bedeutet, dass sie für die Datenaufnahme eingeplant wird

2. PUBLIC.DISABLE_RESOURCE() deaktiviert eine bestimmte Ressource, was bedeutet, dass ihre Aufnahmeplanung gestoppt wird

3. PUBLIC.UPDATE_RESOURCE() ermöglicht es Ihnen, die Datenaufnahmekonfigurationen einer bestimmten Ressource zu aktualisieren. Diese Funktion ist nicht standardmäßig in der Streamlit-UI implementiert, da es manchmal unerwünscht sein kann, dass der Benutzer des Konnektors die Konfiguration der Datenaufnahme (Ingestion) anpassen kann (widerrufen Sie die Berechtigungen für diese Prozedur für die ADMIN-Anwendungsrolle, um ihre Verwendung vollständig zu unterbinden).

Alle diese Prozeduren verfügen über Java-Handler und werden durch Rückrufe erweitert, mit denen Sie ihre Ausführung anpassen können. Sie können benutzerdefinierte Implementierungen von Rückrufen einfügen, indem Sie die Builder für diese Handler verwenden. Standardmäßig stellt die Vorlage simulierte Java-Implementierungen von Callback-Handlern bereit. Diese nachgebildeten Implementierungen tun nichts anderes, als eine erfolgreiche Antwort zurückzugeben. Sie können entweder die benutzerdefinierte Implementierung für die von der Vorlage vorbereiteten Callback-Klassen bereitstellen oder diese Callbacks von Grund auf neu erstellen und sie in den Hauptablauf der Ausführung der Prozedur in den Handler-Builders einfügen.

Einstellungen¶

Die Vorlage enthält eine Registerkarte für Einstellungen, auf der Sie alle zuvor vorgenommenen Konfigurationen einsehen können. Wenn jedoch die Konfigurationseigenschaften angepasst wurden, dann muss auch diese Ansicht angepasst werden. Den Code für die Registerkarte mit den Einstellungen finden Sie in der Datei streamlit/daily_use/settings_page.py.

Zum Anpassen extrahieren Sie einfach die Werte aus der Konfiguration für die Schlüssel, die in den jeweiligen Konfigurationen hinzugefügt wurden. Wenn beispielsweise zuvor im Schritt Verbindungskonfiguration additional_connection_property hinzugefügt wurde, könnte es in der Einstellungsansicht wie folgt hinzugefügt werden:

def connection_config_page():
    current_config = get_connection_configuration()

    # TODO: implement the display for all the custom properties defined in the connection configuration step
    custom_property = current_config.get("custom_connection_property", "")
    additional_connection_property = current_config.get("additional_connection_property", "")


    st.header("Connector configuration")
    st.caption("Here you can see the connector connection configuration saved during the connection configuration step "
               "of the Wizard. If some new property was introduced it has to be added here to display.")
    st.divider()

    st.text_input(
        "Custom connection property:",
        value=custom_property,
        disabled=True
    )
    st.text_input(
        "Additional connection property:",
        value=additional_connection_property,
        disabled=True
    )
    st.divider()