Verwenden des Python-Konnektors¶

Erstellen einer Datenbank, eines Schemas und eines Warehouse¶

Erstellen Sie nach der Anmeldung mit den Befehlen CREATE DATABASE, CREATE SCHEMA und CREATE WAREHOUSE eine Datenbank, ein Schema und ein Warehouse, falls diese noch nicht vorhanden sind.

Das folgende Beispiel zeigt, wie sich ein Warehouse mit dem Namen tiny_warehouse, eine Datenbank mit dem Namen testdb und ein Schema mit dem Namen testschema erstellen lassen. Beachten Sie, dass Sie beim Erstellen des Schemas entweder den Namen der Datenbank angeben müssen, in der das Schema erstellt werden soll, oder bereits mit der Datenbank verbunden sein müssen, in der das Schema erstellt werden soll. Im folgenden Beispiel wird vor dem Befehl CREATE SCHEMA ein USE DATABASE-Befehl ausgeführt, um sicherzustellen, dass das Schema in der korrekten Datenbank erstellt wird.

CopyExpand

conn.cursor().execute("CREATE WAREHOUSE IF NOT EXISTS tiny_warehouse_mg")
conn.cursor().execute("CREATE DATABASE IF NOT EXISTS testdb_mg")
conn.cursor().execute("USE DATABASE testdb_mg")
conn.cursor().execute("CREATE SCHEMA IF NOT EXISTS testschema_mg")

Show lessSee more

Scroll to top

Verwenden von Datenbank, Schema und Warehouse¶

Geben Sie Datenbank und Schema an, in denen Sie Tabellen erstellen möchten. Geben Sie auch das Warehouse an, das Ressourcen für die Ausführung von DML-Anweisungen und -Abfragen bereitstellen soll.

Folgendes Beispiel zeigt die Verwendung von Datenbank testdb, Schema testschema und Warehouse tiny_warehouse (zuvor erstellt):

CopyExpand

conn.cursor().execute("USE WAREHOUSE tiny_warehouse_mg")
conn.cursor().execute("USE DATABASE testdb_mg")
conn.cursor().execute("USE SCHEMA testdb_mg.testschema_mg")

Show lessSee more

Scroll to top

Erstellen von Tabellen und Einfügen von Daten¶

Verwenden Sie den Befehl CREATE TABLE, um Tabellen zu erstellen, und den Befehl INSERT, um die Tabellen mit Daten zu füllen.

Erstellen Sie beispielsweise eine Tabelle mit dem Namen testtable, und fügen Sie zwei Zeilen in die Tabelle ein:

CopyExpand

conn.cursor().execute(
    "CREATE OR REPLACE TABLE "
    "test_table(col1 integer, col2 string)")

conn.cursor().execute(
    "INSERT INTO test_table(col1, col2) VALUES " + 
    "    (123, 'test string1'), " + 
    "    (456, 'test string2')")

Show lessSee more

Scroll to top

Laden von Daten¶

Anstatt Daten mit einzelnen INSERT-Befehlen in Tabellen einzufügen, können Sie Daten aus Dateien, die entweder an einem internen oder externen Speicherort bereitgestellt werden, per Massenladen hinzufügen.

Abfragen von Daten¶

Mit dem Snowflake-Konnektor für Python können Sie Folgendes übermitteln:

• Eine synchrone Abfrage, die die Kontrolle an Ihre Anwendung zurückgibt, nachdem die Abfrage abgeschlossen ist.

• Eine asynchrone Abfrage, die die Kontrolle an Ihre Anwendung zurückgibt, bevor die Abfrage abgeschlossen ist.

Nachdem die Abfrage abgeschlossen wurde, verwenden Sie das Cursor-Objekt, um die Werte in den Ergebnissen abzurufen. Standardmäßig konvertiert der Snowflake-Konnektor für Python die Werte von Snowflake-Datentypen in native Python-Datentypen. (Beachten Sie, dass Sie die Werte auch als Zeichenfolgen zurückgeben und die Typkonvertierungen in Ihrer Anwendung durchführen können. Siehe Verbessern der Abfrageleistung durch Umgehen der Datenkonvertierung.)

conn = snowflake.connector.connect( ... )
cur = conn.cursor()
cur.execute('select * from products')

Ausführen einer asynchronen Abfrage¶

Der Snowflake-Konnektor für Python unterstützt asynchrone Abfragen (d. h. Abfragen, die dem Benutzer die Kontrolle zurückgeben, bevor die Abfrage abgeschlossen ist). Sie können eine asynchrone Abfrage übermitteln und mithilfe von Abrufen (Polling) feststellen, wann die Abfrage abgeschlossen ist. Nachdem die Abfrage abgeschlossen ist, können Sie die Ergebnisse abrufen.

Bemerkung

Um asynchrone Abfragen auszuführen, müssen Sie sicherstellen, dass der Konfigurationsparameter ABORT_DETACHED_QUERY den Wert FALSE hat (Standardwert).

Wenn die Verbindung zum Client unterbrochen wird:

• Bei synchronen Abfragen werden alle laufenden synchronen Abfragen unabhängig vom Wert des Parameters sofort abgebrochen.

• Für asynchrone Abfragen:

  • Wenn ABORT_DETACHED_QUERY auf FALSE eingestellt ist, werden laufende asynchrone Abfragen weiter ausgeführt, bis sie normal beendet werden.

  • Wenn ABORT_DETACHED_QUERY auf TRUE eingestellt ist, bricht Snowflake automatisch alle laufenden asynchronen Abfragen ab, wenn die Verbindung zum Client nach fünf Minuten nicht wiederhergestellt ist.

    Sie können verhindern, dass die asynchrone Abfrage an der Fünf-Minuten-Marke abgebrochen wird, indem Sie cursor.query_result(queryId) aufrufen. Dieser Aufruf ruft zwar nicht das tatsächliche Abfrageergebnis ab, da die Abfrage noch läuft, verhindert jedoch, dass die Abfrage abgebrochen wird. Der Aufruf von query_result ist eine synchrone Operation, die für Ihren speziellen Anwendungsfall geeignet sein kann oder auch nicht.

Mit dieser Funktion können Sie mehrere Abfragen parallel übermitteln, ohne auf den Abschluss jeder Abfrage warten zu müssen. Sie können innerhalb derselben Sitzung auch Kombinationen von synchronen und asynchronen Abfragen ausführen.

Bemerkung

Die Ausführung mehrerer Anweisungen in einer einzigen Abfrage setzt voraus, dass ein gültiges Warehouse in einer Sitzung verfügbar ist.

Schließlich können Sie eine asynchrone Abfrage über eine Verbindung übermitteln und die Ergebnisse über eine andere Verbindung abrufen. Beispielsweise kann ein Benutzer eine Abfrage mit langer Laufzeit über Ihre Anwendung initiieren, die Anwendung beenden und die Anwendung zu einem späteren Zeitpunkt erneut starten, um die Ergebnisse zu überprüfen.

Zum besseren Verständnis der Hierarchie der Geschäftslogik der Treiber und der Interaktion des Parameters ABORT_DETACHED_QUERY betrachten Sie das folgende Flussdiagramm:

Übermitteln einer asynchronen Abfrage¶

Bemerkung

Asynchrone Abfragen unterstützen nicht PUT/GET-Anweisungen.

Wenn cursor.execute_async(query) verwendet wird, verfolgt der Snowflake-Python-Treiber automatisch die asynchron übermittelten Abfragen. Wenn die Verbindung explizit mit connection.close() geschlossen oder der Kontextmanager mit connect()... verwendet wird, wird die Liste der asynchronen Abfragen geprüft. Wenn eine Abfrage noch ausgeführt wird, wird die Snowflake-seitige Sitzung nicht gelöscht.

Wenn innerhalb der gleichen Verbindung keine asynchronen Abfragen laufen, wird die zur Verbindung gehörende Snowflake-Sitzung abgemeldet, wenn connection.close() aufgerufen wird, wodurch implizit alle anderen Abfragen in derselben Sitzung abgebrochen werden.

This behavior also depends on the SQL ABORT_DETACHED_QUERY parameter.

Als bewährte Methode sollten Sie alle asynchronen Aufgaben mit langer Ausführungszeit (insbesondere solche, die nach dem Schließen der Verbindung fortgesetzt werden sollen) in einer separaten Verbindung isolieren.

Sie können über den Verbindungsparameter server_session_keep_alive (Standard: False) dieses automatische Verhalten überschreiben. Standardmäßig wird die Snowflake-Sitzung abgemeldet, wenn connection.close() nur aufgerufen wird, wenn darin keine asynchronen Abfragen ausgeführt werden. Das Standardverhalten berücksichtigt keine Synchronisierungsabfragen und verfolgt diese auch nicht.

Bei der Einstellung server_session_keep_alive=True, meldet sich connection.close() unabhängig vom Status der Abfragen nicht bei der Snowflake-Sitzung ab. Bei Verbindungen, die für asynchrone Abfragen mit langer Ausführungszeit vorgesehen sind, können durch Aktivierung dieser Einstellung der CPU Overhead reduziert und der Verbindungsaufbau beschleunigt werden.

Wichtig

Das Aktivieren dieses Parameters kann unerwartete, abrechenbare Auswirkungen haben (z. B. kann es dazu kommen, dass Abfragen bis zum konfigurierten Wert von STATEMENT_TIMEOUT_IN_SECONDS) ausgeführt werden. Snowflake empfiehlt dringend, dass Sie sorgfältig entscheiden, ob der Wert von server_session_keep_alive unbedingt geändert werden muss. Testen Sie die Änderung, wenn möglich, gründlich in Nicht-Produktionsumgebungen, bevor Sie sie in der Produktion implementieren.

Um eine asynchrone Abfrage zu übermitteln, rufen Sie die Methode execute_async() im Cursor-Objekt auf. Beispiel:

CopyExpand

conn = snowflake.connector.connect( ... )
cur = conn.cursor()
# Submit an asynchronous query for execution.
cur.execute_async('select count(*) from table(generator(timeLimit => 25))')

Show lessSee more

Scroll to top

Nach dem Übermitteln der Abfrage:

• Um festzustellen, ob die Abfrage noch ausgeführt wird, siehe Überprüfen des Status einer Abfrage.

• Um die Ergebnisse der Abfrage abzurufen, siehe Verwenden der Abfrage-ID zum Abrufen von Abfrageergebnissen.

Beispiele für das Ausführen asynchroner Abfragen finden Sie unter Beispiele für asynchrone Abfragen.

Best Practices für asynchrone Abfragen¶

Beachten Sie beim Übermitteln von asynchronen Abfrage die folgenden bewährten Verfahren:

• Stellen Sie vor der parallelen Ausführung von Abfragen sicher, dass Sie die Abhängigkeiten der Abfragen von anderen Abfragen genau kennen. Einige Abfragen sind voneinander und von der Ausführungsreihenfolge abhängig und sind daher nicht für die Parallelisierung geeignet. Beispielsweise kann eine INSERT-Anweisung offensichtlich erst beginnen, nachdem die entsprechende CREATE TABLE-Anweisung abgeschlossen wurde.

• Stellen Sie sicher, dass die Anzahl der gestarteten Abfragen auf den verfügbaren Arbeitsspeicher abgestimmt ist. Das parallele Ausführen mehrerer Abfragen beansprucht normalerweise mehr Speicher, insbesondere wenn mehrere Ergebnis-Datasets gleichzeitig im Arbeitsspeicher gespeichert sind.

• Behandeln Sie beim Abrufen (Polling) die seltenen Fälle, in denen eine Abfrage nicht erfolgreich ist.

• Stellen Sie sicher, dass Anweisungen zur Transaktionssteuerung (BEGIN, COMMIT und ROLLBACK) nicht parallel zu anderen Anweisungen ausgeführt werden.

• Beachten Sie, dass nicht garantiert werden kann, dass asynchrone Abfragen geordnete Ergebnisse liefern, selbst wenn die SQL selbst über eine ORDER BY-Klausel verfügt. Folglich garantiert die Funktion result_scan keine geordneten Ergebnisse.

import time
...
# Execute a long-running query asynchronously.
cur.execute_async('select count(*) from table(generator(timeLimit => 25))')
...
# Wait for the query to finish running.
query_id = cur.sfqid
while conn.is_still_running(conn.get_query_status(query_id)):
  time.sleep(1)

from snowflake.connector import ProgrammingError
import time
...
# Wait for the query to finish running and raise an error
# if a problem occurred with the execution of the query.
try:
  query_id = cur.sfqid
  while conn.is_still_running(conn.get_query_status_throw_if_error(query_id)):
    time.sleep(1)
except ProgrammingError as err:
  print('Programming Error: {0}'.format(err))

# Get the results from a query.
cur.get_results_from_sfqid(query_id)
results = cur.fetchall()
print(f'{results[0]}')

from snowflake.connector import ProgrammingError
import time

conn = snowflake.connector.connect( ... )
cur = conn.cursor()

# Submit an asynchronous query for execution.
cur.execute_async('select count(*) from table(generator(timeLimit => 25))')

# Retrieve the results.
cur.get_results_from_sfqid(query_id)
results = cur.fetchall()
print(f'{results[0]}')

from snowflake.connector import ProgrammingError
import time

conn = snowflake.connector.connect( ... )
cur = conn.cursor()

# Submit an asynchronous query for execution.
cur.execute_async('select count(*) from table(generator(timeLimit => 25))')

# Get the query ID for the asynchronous query.
query_id = cur.sfqid

# Close the cursor and the connection.
cur.close()
conn.close()

# Open a new connection.
new_conn = snowflake.connector.connect( ... )

# Create a new cursor.
new_cur = new_conn.cursor()

# Retrieve the results.
new_cur.get_results_from_sfqid(query_id)
results = new_cur.fetchall()
print(f'{results[0]}')

Herunterladen von Daten¶

Snowflake Connector für Python Version 3.14.0 hat den Verbindungsparameter unsafe_file_write eingeführt, der angibt, wie der Konnektor die Dateiberechtigungen beim Herunterladen von Dateien für einen Stagingbereich von Snowflake mit dem Befehl GET setzen soll. Diese Dateien sind immer im Besitz desselben Benutzers, der den Python-Prozess ausführt.

Standardmäßig lautet der Parameter unsafe_file_writeFalse, da so eine sicherere und strengere 600-Dateiberechtigung möglich ist, was bedeutet, dass nur der Eigentümer die Lese- und Schreibberechtigung für die heruntergeladenen Dateien hat. Andere Gruppen und Benutzer haben keine Berechtigungen für die mit dem GET-Befehl heruntergeladenen Dateien.

Wenn Ihre Organisation weniger restriktive Dateiberechtigungen für die Dateien verlangt, können Sie den Parameter unsafe_file_write auf True setzen. Wenn Sie diesen Parameter aktivieren, werden die Dateiberechtigungen für die von einem Stagingbereich heruntergeladenen Dateien auf 644 gesetzt, sodass der Eigentümer die Dateien lesen und schreiben kann, andere jedoch nur lesen dürfen. Diese Einstellung kann z. B. für einige ETL-Tools notwendig sein, die unter einem anderen Systembenutzer ausgeführt werden, der die heruntergeladenen Dateien lesen und verarbeiten können muss.

Wenn Sie sich nicht sicher sind, welchen Wert Sie verwenden sollen, wenden Sie sich an das Team, das für die geltenden Sicherheitsrichtlinien Ihres Unternehmens zuständig ist.

Binden von Daten¶

Um Werte anzugeben, die in einer SQL-Anweisung verwendet werden sollen, können Sie Literale in die Anweisung einschließen oder Variablen binden. Wenn Sie Variablen binden, fügen Sie einen oder mehrere Platzhalter in den Text der SQL-Anweisung ein, und geben Sie dann für jeden Platzhalter die Variable (den zu verwendenden Wert) an.

Im folgenden Beispiel wird die Verwendung von Literalen und die Bindung gegenübergestellt:

Literale:

CopyExpand

con.cursor().execute("INSERT INTO testtable(col1, col2) VALUES(789, 'test string3')")

Show lessSee more

Scroll to top

Bindung:

CopyExpand

con.cursor().execute(
    "INSERT INTO testtable(col1, col2) "
    "VALUES(%s, %s)", (
        789,
        'test string3'
    ))

Show lessSee more

Scroll to top

Snowflake unterstützt die folgenden Bindungstypen:

• pyformat und format, die Daten auf dem Client binden.

• qmark und numeric, die Daten auf dem Server binden.

Jeder Typ wird unten erklärt.

CREATE TEMPORARY STAGE SYSTEM$BIND file_format=(type=csv field_optionally_enclosed_by='"')
Cannot perform CREATE STAGE. This session does not have a current schema. Call 'USE SCHEMA', or use a qualified name.

Abrufen von Spaltenmetadaten¶

Für das Abrufen der Metadaten jeder Spalte im Resultset abzurufen (z. B. Name, Typ, Genauigkeit, Skala usw. jeder Spalte) gibt es folgende Möglichkeiten:

• Nach dem Aufruf der Methode execute() zur Ausführung der Abfrage können Sie mit dem Attribut description des Cursor-Objekts auf die Metadaten zuzugreifen.

• Wenn Sie auf die Metadaten zugreifen möchten, ohne eine Abfrage ausführen zu müssen, rufen Sie die Methode describe() auf.

  Die Methode describe ist im Snowflake-Konnektor für Python ab Version 2.4.6 verfügbar.

Das Attribut description wird auf einen der folgenden Werte gesetzt:

• Version 2.4.5 und früher: Eine Liste von Tupeln.

• Versionen 2.4.6 und später: Eine Liste von ResultMetadata-Objekten. (Die Methode describe gibt ebenfalls diese Liste zurück.)

Jedes Tupel- und ResultMetadata-Objekt enthält die Metadaten zu einer Spalte (Spaltenname, Datentyp usw.). Sie können auf die Metadaten über den Index zugreifen oder ab Version 2.4.6 auch über das Attribut ResultMetadata.

Die folgenden Beispiele zeigen, wie Sie auf die Metadaten der zurückgegebenen Tupel und ResultMetadata-Objekte zugreifen können.

Beispiel: Abrufen der Spaltennamen-Metadaten nach Index (ab Version 2.4.5):

Im folgenden Beispiel wird das Attribut description verwendet, um die Liste der Spaltennamen nach der Ausführung einer Abfrage abzurufen. Das Attribut ist eine Liste von Tupeln. Das Beispiel greift auf den Spaltennamen des ersten Wertes in jedem Tupel zu.

CopyExpand

cur = conn.cursor()
cur.execute("SELECT * FROM test_table")
print(','.join([col[0] for col in cur.description]))

Show lessSee more

Scroll to top

Beispiel: Abrufen der Spaltennamen-Metadaten nach Attribut (ab Version 2.4.6):

Im folgenden Beispiel wird das Attribut description verwendet, um die Liste der Spaltennamen nach der Ausführung einer Abfrage abzurufen. Das Attribut ist eine Liste von ResultMetaData-Objekten. Das Beispiel greift auf den Spaltennamen aus dem Attribut name jedes ResultMetadata-Objekts zu.

CopyExpand

cur = conn.cursor()
cur.execute("SELECT * FROM test_table")
print(','.join([col.name for col in cur.description]))

Show lessSee more

Scroll to top

Beispiel: Abrufen der Spaltennamen-Metadaten ohne Ausführen der Abfrage (ab Version 2.4.6):

Das folgende Beispiel verwendet die Methode describe, um die Liste der Spaltennamen abzurufen, ohne eine Abfrage ausführen zu müssen. Die Methode describe() gibt eine Liste von ResultMetaData-Objekten zurück. Das Beispiel greift auf den Spaltennamen aus dem Attribut name jedes ResultMetadata-Objekts zu.

CopyExpand

cur = conn.cursor()
result_metadata_list = cur.describe("SELECT * FROM test_table")
print(','.join([col.name for col in result_metadata_list]))

Show lessSee more

Scroll to top

Fehlerbehandlung¶

Die Anwendung muss auf Ausnahmen, die vom Snowflake-Konnektor ausgelöst werden, korrekt reagieren und entscheiden, ob die Codeausführung fortgesetzt oder abgebrochen wird.

CopyExpand

# Catching the syntax error
cur = con.cursor()
try:
    cur.execute("SELECT * FROM testtable")
except snowflake.connector.errors.ProgrammingError as e:
    # default error message
    print(e)
    # customer error message
    print('Error {0} ({1}): {2} ({3})'.format(e.errno, e.sqlstate, e.msg, e.sfqid))
finally:
    cur.close()

Show lessSee more

Scroll to top

Verwenden von execute_stream zum Ausführen von SQL-Skripten¶

Mit der Funktion execute_stream können Sie ein oder mehrere SQL-Skripte in einem Stream ausführen:

CopyExpand

from codecs import open
with open(sqlfile, 'r', encoding='utf-8') as f:
    for cur in con.execute_stream(f):
        for ret in cur:
            print(ret)

Show lessSee more

Scroll to top

Schließen der Verbindung¶

Als Best Practice schließen Sie die Verbindung, indem Sie die Methode close aufrufen:

CopyExpand

connection.close()

Show lessSee more

Scroll to top

Dadurch wird sichergestellt, dass die gesammelten Clientmetriken an den Server übermittelt werden und die Sitzung gelöscht wird. Außerdem helfen try-finally Blöcke sicherzustellen, dass die Verbindung geschlossen wird, auch wenn mittendrin eine Ausnahme ausgelöst wird:

CopyExpand

# Connecting to Snowflake
con = snowflake.connector.connect(...)
try:
    # Running queries
    con.cursor().execute(...)
    ...
finally:
    # Closing the connection
    con.close()

Show lessSee more

Scroll to top

Verwenden eines Kontextmanagers zum Verbinden und Steuern von Transaktionen¶

Der Snowflake-Konnektor für Python unterstützt einen Kontextmanager, der bei Bedarf Ressourcen zuweist und freigibt. Der Kontextmanager ist nützlich, um für Transaktionen auf Basis des Anweisungsstatus ein Commit oder ein Rollback auszuführen, wenn autocommit deaktiviert ist.

CopyExpand

# Connecting to Snowflake using the context manager
with snowflake.connector.connect(
  user=USER,
  password=PASSWORD,
  account=ACCOUNT,
  autocommit=False,
) as con:
    con.cursor().execute("INSERT INTO a VALUES(1, 'test1')")
    con.cursor().execute("INSERT INTO a VALUES(2, 'test2')")
    con.cursor().execute("INSERT INTO a VALUES(not numeric value, 'test3')") # fail

Show lessSee more

Scroll to top

Im obigen Beispiel führt der Kontextmanager nach Fehlschlagen der dritten Anweisung ein Rollback der Änderungen in der Transaktion aus und schließt die Verbindung. Wenn alle Anweisungen erfolgreich waren, würde der Kontextmanager die Änderungen committen und die Verbindung schließen.

Der äquivalente Code mit try- und except-Blöcken ist wie folgt:

CopyExpand

# Connecting to Snowflake using try and except blocks
con = snowflake.connector.connect(
  user=USER,
  password=PASSWORD,
  account=ACCOUNT,
  autocommit=False)
try:
    con.cursor().execute("INSERT INTO a VALUES(1, 'test1')")
    con.cursor().execute("INSERT INTO a VALUES(2, 'test2')")
    con.cursor().execute("INSERT INTO a VALUES(not numeric value, 'test3')") # fail
    con.commit()
except Exception as e:
    con.rollback()
    raise e
finally:
    con.close()

Show lessSee more

Scroll to top

Verwenden des Datentyps VECTOR¶

Unterstützung für den Datentyp VECTOR Datentyp wurde in Version 3.6.0 des Snowflake Connector für Python eingeführt. Sie können den Datentyp VECTOR mit den Vektorähnlichkeitsfunktionen verwenden, um Anwendungen zu implementieren, die auf Vektorsuche oder Retrieval-Augmented-Generation basieren (RAG).

Das folgende Codebeispiel zeigt, wie Sie den Python-Konnektor verwenden, um Tabellen mit VECTOR-Spalten zu erstellen und die Funktion VECTOR_INNER_PRODUCT aufzurufen:

import snowflake.connector

conn = ... # Set up connection
cur = conn.cursor()

# Create a table and insert some vectors
cur.execute("CREATE OR REPLACE TABLE vectors (a VECTOR(FLOAT, 3), b VECTOR(FLOAT, 3))")
values = [([1.1, 2.2, 3], [1, 1, 1]), ([1, 2.2, 3], [4, 6, 8])]
for row in values:
    cur.execute(f"""
        INSERT INTO vectors(a, b)
          SELECT {row[0]}::VECTOR(FLOAT,3), {row[1]}::VECTOR(FLOAT,3)
    """)

# Compute the pairwise inner product between columns a and b
cur.execute("SELECT VECTOR_INNER_PRODUCT(a, b) FROM vectors")
print(cur.fetchall())

[(6.30...,), (41.2...,)]

Das folgende Codebeispiel zeigt, wie Sie den Python-Konnektor verwenden, um die Funktion VECTOR_COSINE_SIMILARITY aufzurufen, um die nächstgelegenen Vektoren zu [1,2,3] zu finden:

cur.execute(f"""
    SELECT a, VECTOR_COSINE_SIMILARITY(a, {[1,2,3]}::VECTOR(FLOAT, 3))
      AS similarity
      FROM vectors
      ORDER BY similarity DESC
      LIMIT 1;
""")
print(cur.fetchall())

[([1.0, 2.2..., 3.0], 0.9990...)]

Protokollieren¶

Der Snowflake-Konnektor für Python nutzt das logging-Standardmodul von Python, um den Status in regelmäßigen Abständen zu protokollieren, sodass die Anwendung ihre Aktivitäten im Hintergrund verfolgen kann. Der einfachste Weg, die Protokollierung zu aktivieren, ist der Aufruf von logging.basicConfig() am Anfang der Anwendung.

Im folgenden Beispiel wird der Protokolliergrad auf INFO eingestellt, und die Protokolle werden in einer Datei namens /tmp/snowflake_python_connector.log gespeichert.

CopyExpand

logging.basicConfig(
    filename=file_name,
    level=logging.INFO)

Show lessSee more

Scroll to top

Eine umfassendere Protokollierung kann durch Einstellen des Protokolliergrads auf DEBUG wie folgt aktiviert werden:

CopyExpand

# Logging including the timestamp, thread and the source code location
import logging
for logger_name in ['snowflake.connector', 'botocore', 'boto3']:
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)
    ch = logging.FileHandler('/tmp/python_connector.log')
    ch.setLevel(logging.DEBUG)
    ch.setFormatter(logging.Formatter('%(asctime)s - %(threadName)s %(filename)s:%(lineno)d - %(funcName)s() - %(levelname)s - %(message)s'))
    logger.addHandler(ch)

Show lessSee more

Scroll to top

Die optionale, aber empfohlene Formatierungsklasse SecretDetector stellt sicher, dass ein bestimmter Satz bekannter vertraulicher Informationen maskiert wird, bevor diese in die Protokolldateien des Python-Konnektors von Snowflake geschrieben werden. Verwenden Sie für SecretDetector folgenden Code:

CopyExpand

# Logging including the timestamp, thread and the source code location
import logging
from snowflake.connector.secret_detector import SecretDetector
for logger_name in ['snowflake.connector', 'botocore', 'boto3']:
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)
    ch = logging.FileHandler('/tmp/python_connector.log')
    ch.setLevel(logging.DEBUG)
    ch.setFormatter(SecretDetector('%(asctime)s - %(threadName)s %(filename)s:%(lineno)d - %(funcName)s() - %(levelname)s - %(message)s'))
    logger.addHandler(ch)

Show lessSee more

Scroll to top

Bemerkung

botocore und boto3 sind über das AWS (Amazon Web Services) SDK für Python verfügbar.

[log]
save_logs = true
level = "INFO"
path = "<directory to store logs>"

Beispielprogramm¶

Der folgende Beispielcode kombiniert die meisten der in den vorangegangenen Abschnitten beschriebenen Beispiele zu einem funktionierenden Python-Programm: Dieses Beispiel enthält zwei Teile:

• Eine übergeordnete Klasse („python_veritas_base“) enthält den Code für viele gängige Operationen, z. B. das Herstellen einer Verbindung zum Server.

• Eine untergeordnete Klasse („python_connector_example“) repräsentiert die kundenspezifischen Teile eines bestimmten Clients, z. B. zum Abfragen einer Tabelle.

Dieser Beispielcode wird direkt aus einem unserer Tests importiert, um sicherzustellen, dass er in einem neueren Build des Produkts ausgeführt wurde.

Da dies aus einem Test stammt, enthält es eine kleine Menge Code, um einen alternativen Port und ein alternatives Protokoll festzulegen, die in einigen Tests verwendet werden. Benutzer sollten Protokoll oder Portnummer nicht festlegen. Lassen Sie diese Werte stattdessen aus, und verwenden Sie die Standardeinstellungen.

Dies enthält auch einige Abschnittsmarkierungen (manchmal als „Snippet-Tags“ bezeichnet), um Code zu identifizieren, der unabhängig in die Dokumentation importiert werden kann. Abschnittsmarkierungen sehen normalerweise ähnlich aus wie:

# -- (> ---------------------- SECTION=import_connector ---------------------
...
# -- <) ---------------------------- END_SECTION ----------------------------

Diese Abschnittsmarkierungen sind im Benutzercode nicht erforderlich.

Der erste Teil des Codebeispiels enthält die allgemeinen Unterroutinen für:

• Lesen Sie Befehlszeilenargumente (z. B. „–warehouse MyWarehouse“), die Verbindungsinformationen enthalten.

• Stellen Sie eine Verbindung zum Server her.

• Erstellen und verwenden Sie ein Warehouse, eine Datenbank und ein Schema.

• Löschen Sie das Schema, die Datenbank und das Warehouse, wenn Sie damit fertig sind.

import logging
import os
import sys


# -- (> ---------------------- SECTION=import_connector ---------------------
import snowflake.connector
# -- <) ---------------------------- END_SECTION ----------------------------


class python_veritas_base:

    """
    PURPOSE:
        This is the Base/Parent class for programs that use the Snowflake
        Connector for Python.
        This class is intended primarily for:
            * Sample programs, e.g. in the documentation.
            * Tests.
    """


    def __init__(self, p_log_file_name = None):

        """
        PURPOSE:
            This does any required initialization steps, which in this class is
            basically just turning on logging.
        """

        file_name = p_log_file_name
        if file_name is None:
            file_name = '/tmp/snowflake_python_connector.log'

        # -- (> ---------- SECTION=begin_logging -----------------------------
        logging.basicConfig(
            filename=file_name,
            level=logging.INFO)
        # -- <) ---------- END_SECTION ---------------------------------------


    # -- (> ---------------------------- SECTION=main ------------------------
    def main(self, argv):

        """
        PURPOSE:
            Most tests follow the same basic pattern in this main() method:
               * Create a connection.
               * Set up, e.g. use (or create and use) the warehouse, database,
                 and schema.
               * Run the queries (or do the other tasks, e.g. load data).
               * Clean up. In this test/demo, we drop the warehouse, database,
                 and schema. In a customer scenario, you'd typically clean up
                 temporary tables, etc., but wouldn't drop your database.
               * Close the connection.
        """

        # Read the connection parameters (e.g. user ID) from the command line
        # and environment variables, then connect to Snowflake.
        connection = self.create_connection(argv)

        # Set up anything we need (e.g. a separate schema for the test/demo).
        self.set_up(connection)

        # Do the "real work", for example, create a table, insert rows, SELECT
        # from the table, etc.
        self.do_the_real_work(connection)

        # Clean up. In this case, we drop the temporary warehouse, database, and
        # schema.
        self.clean_up(connection)

        print("\nClosing connection...")
        # -- (> ------------------- SECTION=close_connection -----------------
        connection.close()
        # -- <) ---------------------------- END_SECTION ---------------------

    # -- <) ---------------------------- END_SECTION=main --------------------


    def args_to_properties(self, args):

        """
        PURPOSE:
            Read the command-line arguments and store them in a dictionary.
            Command-line arguments should come in pairs, e.g.:
                "--user MyUser"
        INPUTS:
            The command line arguments (sys.argv).
        RETURNS:
            Returns the dictionary.
        DESIRABLE ENHANCEMENTS:
            Improve error detection and handling.
        """

        connection_parameters = {}

        i = 1
        while i < len(args) - 1:
            property_name = args[i]
            # Strip off the leading "--" from the tag, e.g. from "--user".
            property_name = property_name[2:]
            property_value = args[i + 1]
            connection_parameters[property_name] = property_value
            i += 2

        return connection_parameters


    def create_connection(self, argv):

        """
        PURPOSE:
            This gets account identifier and login information from the
            environment variables and command-line parameters, connects to the
            server, and returns the connection object.
        INPUTS:
            argv: This is usually sys.argv, which contains the command-line
                  parameters. It could be an equivalent substitute if you get
                  the parameter information from another source.
        RETURNS:
            A connection.
        """

        # Get account identifier and login information from environment variables and command-line parameters.
        # For information about account identifiers, see
        # https://docs.snowflake.com/en/user-guide/admin-account-identifier.html .
        # -- (> ----------------------- SECTION=set_login_info ---------------

        # Get the password from an appropriate environment variable, if
        # available.
        PASSWORD = os.getenv('SNOWSQL_PWD')

        # Get the other login info etc. from the command line.
        if len(argv) < 11:
            msg = "ERROR: Please pass the following command-line parameters:\n"
            msg += "--warehouse <warehouse> --database <db> --schema <schema> "
            msg += "--user <user> --account <account_identifier> "
            print(msg)
            sys.exit(-1)
        else:
            connection_parameters = self.args_to_properties(argv)
            USER = connection_parameters["user"]
            ACCOUNT = connection_parameters["account"]
            WAREHOUSE = connection_parameters["warehouse"]
            DATABASE = connection_parameters["database"]
            SCHEMA = connection_parameters["schema"]
            # Optional: for internal testing only.
            try:
                PORT = connection_parameters["port"]
            except:
                PORT = ""
            try:
                PROTOCOL = connection_parameters["protocol"]
            except:
                PROTOCOL = ""

        # If the password is set by both command line and env var, the
        # command-line value takes precedence over (is written over) the
        # env var value.

        # If the password wasn't set either in the environment var or on
        # the command line...
        if PASSWORD is None or PASSWORD == '':
            print("ERROR: Set password, e.g. with SNOWSQL_PWD environment variable")
            sys.exit(-2)
        # -- <) ---------------------------- END_SECTION ---------------------

        # Optional diagnostic:
        #print("USER:", USER)
        #print("ACCOUNT:", ACCOUNT)
        #print("WAREHOUSE:", WAREHOUSE)
        #print("DATABASE:", DATABASE)
        #print("SCHEMA:", SCHEMA)
        #print("PASSWORD:", PASSWORD)
        #print("PROTOCOL:" "'" + PROTOCOL + "'")
        #print("PORT:" + "'" + PORT + "'")

        print("Connecting...")
        # If the PORT is set but the protocol is not, we ignore the PORT (bug!!).
        if PROTOCOL is None or PROTOCOL == "" or PORT is None or PORT == "":
            # -- (> ------------------- SECTION=connect_to_snowflake ---------
            conn = snowflake.connector.connect(
                user=USER,
                password=PASSWORD,
                account=ACCOUNT,
                warehouse=WAREHOUSE,
                database=DATABASE,
                schema=SCHEMA
                )
            # -- <) ---------------------------- END_SECTION -----------------
        else:

            conn = snowflake.connector.connect(
                user=USER,
                password=PASSWORD,
                account=ACCOUNT,
                warehouse=WAREHOUSE,
                database=DATABASE,
                schema=SCHEMA,
                # Optional: for internal testing only.
                protocol=PROTOCOL,
                port=PORT
                )

        return conn


    def set_up(self, connection):

        """
        PURPOSE:
            Set up to run a test. You can override this method with one
            appropriate to your test/demo.
        """

        # Create a temporary warehouse, database, and schema.
        self.create_warehouse_database_and_schema(connection)


    def do_the_real_work(self, conn):

        """
        PURPOSE:
            Your sub-class should override this to include the code required for
            your documentation sample or your test case.
            This default method does a very simple self-test that shows that the
            connection was successful.
        """

        # Create a cursor for this connection.
        cursor1 = conn.cursor()
        # This is an example of an SQL statement we might want to run.
        command = "SELECT PI()"
        # Run the statement.
        cursor1.execute(command)
        # Get the results (should be only one):
        for row in cursor1:
            print(row[0])
        # Close this cursor.
        cursor1.close()


    def clean_up(self, connection):

        """
        PURPOSE:
            Clean up after a test. You can override this method with one
            appropriate to your test/demo.
        """

        # Create a temporary warehouse, database, and schema.
        self.drop_warehouse_database_and_schema(connection)


    def create_warehouse_database_and_schema(self, conn):

        """
        PURPOSE:
            Create the temporary schema, database, and warehouse that we use
            for most tests/demos.
        """

        # Create a database, schema, and warehouse if they don't already exist.
        print("\nCreating warehouse, database, schema...")
        # -- (> ------------- SECTION=create_warehouse_database_schema -------
        conn.cursor().execute("CREATE WAREHOUSE IF NOT EXISTS tiny_warehouse_mg")
        conn.cursor().execute("CREATE DATABASE IF NOT EXISTS testdb_mg")
        conn.cursor().execute("USE DATABASE testdb_mg")
        conn.cursor().execute("CREATE SCHEMA IF NOT EXISTS testschema_mg")
        # -- <) ---------------------------- END_SECTION ---------------------

        # -- (> --------------- SECTION=use_warehouse_database_schema --------
        conn.cursor().execute("USE WAREHOUSE tiny_warehouse_mg")
        conn.cursor().execute("USE DATABASE testdb_mg")
        conn.cursor().execute("USE SCHEMA testdb_mg.testschema_mg")
        # -- <) ---------------------------- END_SECTION ---------------------


    def drop_warehouse_database_and_schema(self, conn):

        """
        PURPOSE:
            Drop the temporary schema, database, and warehouse that we create
            for most tests/demos.
        """

        # -- (> ------------- SECTION=drop_warehouse_database_schema ---------
        conn.cursor().execute("DROP SCHEMA IF EXISTS testschema_mg")
        conn.cursor().execute("DROP DATABASE IF EXISTS testdb_mg")
        conn.cursor().execute("DROP WAREHOUSE IF EXISTS tiny_warehouse_mg")
        # -- <) ---------------------------- END_SECTION ---------------------


# ----------------------------------------------------------------------------

if __name__ == '__main__':
    pvb = python_veritas_base()
    pvb.main(sys.argv)

Der zweite Teil des Codebeispiels erstellt eine Tabelle, fügt Zeilen ein usw.:

import sys

# -- (> ---------------------- SECTION=import_connector ---------------------
import snowflake.connector
# -- <) ---------------------------- END_SECTION ----------------------------


# Import the base class that contains methods used in many tests and code 
# examples.
from python_veritas_base import python_veritas_base


class python_connector_example (python_veritas_base):

  """
  PURPOSE:
      This is a simple example program that shows how to use the Snowflake 
      Python Connector to create and query a table.
  """

  def __init__(self):
    pass


  def do_the_real_work(self, conn):

    """
    INPUTS:
        conn is a Connection object returned from snowflake.connector.connect().
    """

    print("\nCreating table test_table...")
    # -- (> ----------------------- SECTION=create_table ---------------------
    conn.cursor().execute(
        "CREATE OR REPLACE TABLE "
        "test_table(col1 integer, col2 string)")

    conn.cursor().execute(
        "INSERT INTO test_table(col1, col2) VALUES " + 
        "    (123, 'test string1'), " + 
        "    (456, 'test string2')")
    # -- <) ---------------------------- END_SECTION -------------------------


    print("\nSelecting from test_table...")
    # -- (> ----------------------- SECTION=querying_data --------------------
    cur = conn.cursor()
    try:
        cur.execute("SELECT col1, col2 FROM test_table ORDER BY col1")
        for (col1, col2) in cur:
            print('{0}, {1}'.format(col1, col2))
    finally:
        cur.close()
    # -- <) ---------------------------- END_SECTION -------------------------




# ============================================================================

if __name__ == '__main__':

    test_case = python_connector_example()
    test_case.main(sys.argv)

Gehen Sie folgendermaßen vor, um dieses Beispiel auszuführen:

1. Kopieren Sie den ersten Teil des Codes in eine Datei mit dem Namen „python_veritas_base.py“.

2. Kopieren Sie den zweiten Teil des Codes in eine Datei mit dem Namen „python_connector_example.py“.

3. Setzen Sie die Umgebungsvariable SNOWSQL_PWD auf Ihr Kennwort, zum Beispiel:

   CopyExpand

   export SNOWSQL_PWD='MyPassword'
   

   Show lessSee more

   Scroll to top

4. Führen Sie das Programm über eine Befehlszeile ähnlich der folgenden aus (ersetzen Sie die Benutzer- und Kontoinformationen natürlich durch Ihre eigenen Benutzer- und Kontoinformationen).

   Warnung

   Dies löscht das Warehouse, die Datenbank und das Schema am Ende des Programms! Verwenden Sie nicht den Namen einer vorhandenen Datenbank, da Sie diese verlieren werden!

   CopyExpand

   python3 python_connector_example.py --warehouse <unique_warehouse_name> --database <new_warehouse_zzz_test> --schema <new_schema_zzz_test> --account myorganization-myaccount --user MyUserName
   

   Show lessSee more

   Scroll to top

Und hier ist die Ausgabe:

Connecting...

Creating warehouse, database, schema...

Creating table test_table...

Selecting from test_table...
123, test string1
456, test string2

Closing connection...

Hier ist ein längeres Beispiel:

Bemerkung

Achten Sie darauf, dass Sie in dem Abschnitt, in dem Sie Ihre Konto- und Anmeldeinformationen festlegen, die Variablen bei Bedarf so ersetzen, dass sie Ihren Snowflake-Anmeldeinformationen (Name, Kennwort usw.) entsprechen.

In diesem Beispiel wird die Anweisung mit der Funktion format() zusammengestellt. Wenn in Ihrer Umgebung das Risiko von Angriffen durch Einschleusung von SQL-Befehlen besteht, sollten Sie möglicherweise das Binden von Werten vorziehen, anstatt format() zu verwenden.

#!/usr/bin/env python
#
# Snowflake Connector for Python Sample Program
#

# Logging
import logging
logging.basicConfig(
    filename='/tmp/snowflake_python_connector.log',
    level=logging.INFO)

import snowflake.connector

# Set ACCOUNT to your account identifier.
# See https://docs.snowflake.com/en/user-guide/gen-conn-config.
ACCOUNT = '<my_organization>-<my_account>'
# Set your login information.
USER = '<login_name>'
PASSWORD = '<password>'

import os

# Only required if you copy data from your S3 bucket
AWS_ACCESS_KEY_ID = os.getenv('AWS_ACCESS_KEY_ID')
AWS_SECRET_ACCESS_KEY = os.getenv('AWS_SECRET_ACCESS_KEY')

# Connecting to Snowflake
con = snowflake.connector.connect(
  user=USER,
  password=PASSWORD,
  account=ACCOUNT,
)

# Creating a database, schema, and warehouse if none exists
con.cursor().execute("CREATE WAREHOUSE IF NOT EXISTS tiny_warehouse")
con.cursor().execute("CREATE DATABASE IF NOT EXISTS testdb")
con.cursor().execute("USE DATABASE testdb")
con.cursor().execute("CREATE SCHEMA IF NOT EXISTS testschema")

# Using the database, schema and warehouse
con.cursor().execute("USE WAREHOUSE tiny_warehouse")
con.cursor().execute("USE SCHEMA testdb.testschema")

# Creating a table and inserting data
con.cursor().execute(
    "CREATE OR REPLACE TABLE "
    "testtable(col1 integer, col2 string)")
con.cursor().execute(
    "INSERT INTO testtable(col1, col2) "
    "VALUES(123, 'test string1'),(456, 'test string2')")

# Copying data from internal stage (for testtable table)
con.cursor().execute("PUT file:///tmp/data0/file* @%testtable")
con.cursor().execute("COPY INTO testtable")

# Copying data from external stage (S3 bucket -
# replace <s3_bucket> with the name of your bucket)
con.cursor().execute("""
COPY INTO testtable FROM s3://<s3_bucket>/data/
     STORAGE_INTEGRATION = myint
     FILE_FORMAT=(field_delimiter=',')
""".format(
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY))

# Querying data
cur = con.cursor()
try:
    cur.execute("SELECT col1, col2 FROM testtable")
    for (col1, col2) in cur:
        print('{0}, {1}'.format(col1, col2))
finally:
    cur.close()

# Binding data
con.cursor().execute(
    "INSERT INTO testtable(col1, col2) "
    "VALUES(%(col1)s, %(col2)s)", {
        'col1': 789,
        'col2': 'test string3',
        })

# Retrieving column names
cur = con.cursor()
cur.execute("SELECT * FROM testtable")
print(','.join([col[0] for col in cur.description]))

# Catching syntax errors
cur = con.cursor()
try:
    cur.execute("SELECT * FROM testtable")
except snowflake.connector.errors.ProgrammingError as e:
    # default error message
    print(e)
    # user error message
    print('Error {0} ({1}): {2} ({3})'.format(e.errno, e.sqlstate, e.msg, e.sfqid))
finally:
    cur.close()

# Retrieving the Snowflake query ID
cur = con.cursor()
cur.execute("SELECT * FROM testtable")
print(cur.sfqid)

# Closing the connection
con.close()