Démarrer avec le Snowflake Connector pour Python

Le Snowflake Connector est un package Python natif : il s'installe comme n'importe quel autre package Python.

Avec pip

Le plus simple pour installer la bibliothèque est de passer par pip, le gestionnaire de packages officiel de Python :

1pip install snowflake-connector-python

Avec conda

Si vous utilisez conda comme gestionnaire de packages, installez le package de la manière suivante :

1conda install snowflake-connector-python

Voici un exemple complet illustrant l'utilisation du package Snowflake Connector pour exécuter une requête et traiter les résultats. Cet exemple affiche la consommation totale de crédits par virtual warehouse sur les 30 derniers jours :

Les données affichées en sortie ressembleront à ceci :

('REPLICATOR_LARGE', Decimal('669.080833333'))
('REPLICATOR_XSMALL', Decimal('613.355000002'))
('REPLICATOR_XLARGE', Decimal('416.437500000'))
('BACKEND_LARGE', Decimal('162.702500031'))
('BACKEND', Decimal('105.263888856'))
('BACKEND_MEDIUM', Decimal('47.033055570'))
('REPLICATOR_MEDIUM', Decimal('40.775833333'))
('REPLICATOR_SMALL', Decimal('31.046388890'))
('BACKEND_STAGING', Decimal('26.985555532'))
('BACKEND_SMALL', Decimal('12.798333345'))
('SELECT_DOGFOOD', Decimal('11.283055542'))
('ADMIN', Decimal('10.899444408'))
('BACKEND_STAGING_MEDIUM', Decimal('6.822222224'))
('BI_XSMALL', Decimal('1.814999976'))
('CLOUD_SERVICES_ONLY', Decimal('0E-9'))

Pour récupérer votre identifiant de compte, suivez les étapes détaillées dans cet article

Les objets Connection et Cursor

Comme l'illustre l'exemple de démarrage rapide ci-dessus, nous manipulons deux objets distincts : l'objet connection et l'objet cursor.

L'objet connection se charge de l'authentification, de la connexion à Snowflake et de la fourniture des curseurs. Les curseurs servent quant à eux à exécuter des requêtes sur Snowflake et à en récupérer les résultats. Une même connexion peut comporter plusieurs curseurs, et un même curseur peut exécuter plusieurs requêtes : inutile d'en créer un nouveau à chaque fois. Il en va de même pour l'objet connection. Établir une connexion à Snowflake prend quelques secondes : autant ne le faire qu'une seule fois.

Pour en savoir plus sur les méthodes disponibles pour chaque objet, consultez la documentation officielle de l'API du Snowflake Python connector.

Requêtes synchrones et asynchrones

Il existe deux façons d'exécuter des requêtes avec le Snowflake Connector pour Python : de façon synchrone ou asynchrone.

Lancer une requête synchrone

Lorsque vous lancez une requête synchrone, votre processus Python attend que la requête se termine avant de poursuivre. C'est l'approche la plus courante avec le Python connector, en raison de sa simplicité :

cur = conn.cursor()
cur.execute('select * from table') # Python attendra la fin de la requête avant de passer à la ligne suivante
results = cur.fetchall()

Lancer une requête asynchrone

Avec une requête asynchrone, le package Snowflake Python rend immédiatement la main à votre processus Python, sans attendre la fin de la requête. C'est particulièrement utile lorsque vous développez des applications Python multi-threadées (une application web, par exemple) ou que vous utilisez Python pour lancer de nombreuses requêtes en parallèle.

Pour lancer une requête asynchrone, utilisez la méthode execute_async du curseur à la place de la méthode execute classique. Plus loin dans votre code, il faudra interroger Snowflake pour vérifier si la requête est terminée.

cur = conn.cursor()
cur.execute_async('select * from table') # Python passe immédiatement à la ligne suivante
query_id = cur.sfqid

while True:
		if conn.is_still_running(conn.get_query_status(query_id)):
				break
    time.sleep(1)

cur.get_results_from_sfqid(query_id)
results = cur.fetchall()

Contrôle des transactions

Par défaut, lorsque vous exécutez des requêtes dans Snowflake via le package Snowflake Connector, celles-ci sont automatiquement validées (commit).

Pour un contrôle plus fin de vos transactions SQL, définissez autocommit=False au moment d'établir votre connexion.

Avec le gestionnaire de contexte with

Examinons le code suivant. Si l'une des instructions SQL échoue, Snowflake effectue automatiquement un rollback. Si elles réussissent, elles sont automatiquement validées à la sortie du bloc with.

with snowflake.connector.connect(
    user=USER,
    password=PASSWORD,
    account=ACCOUNT,
    autocommit=False,
) as conn:
    cur = conn.cursor()
    cur.execute("insert into table values(...)")
    cur.execute("insert into table values(...)")

Contrôle manuel avec try-except

Pour un contrôle plus précis, vous pouvez valider ou annuler manuellement les transactions SQL grâce aux méthodes conn.commit() et conn.close(), combinées au pattern try-except-finally :

con = snowflake.connector.connect(
  user=USER,
  password=PASSWORD,
  account=ACCOUNT,
  autocommit=False)
try:
    cur = conn.cursor()
    cur.execute("insert into table values(...)")
    cur.execute("insert into table values(...)")
    conn.commit()
except Exception as e:
    conn.rollback()
    raise e
finally:
    conn.close()

Fermer la connexion

En bonne pratique, fermez votre connexion à la fin de votre script Python avec la méthode conn.close(). Pour garantir que la connexion sera bien fermée même en cas d'erreur dans votre script, encapsulez votre code dans un bloc try - finally :

try:
   cur.execute('...')
   results = cur.fetchall()
   # ...
except:
    # gérer les erreurs au besoin
finally:
		conn.close()

Autre pattern utile : utiliser la syntaxe Python with(), qui ferme automatiquement la connexion :

with snowflake.connector.connect(...) as conn:
		cur = conn.cursor()
		cur.execute('...')
		results = cur.fetchall()

# la méthode conn.close() est automatiquement appelée à la sortie du bloc with
# On peut alors enchaîner sur d'autres opérations sans rapport avec Snowflake
with open('filename.csv', 'w') as f:
		f.write(...)

Passons en revue d'autres exemples mettant en lumière des patterns courants avec le Snowflake Python Connector.

Passer des paramètres à votre instruction SQL

Les développeurs ont fréquemment besoin de transmettre différents paramètres à leurs instructions SQL.

Python offre une excellente prise en charge de la manipulation de chaînes via les f-strings, parfaitement adaptées à cet usage :

user_id = 123456
sql = f"select * from users where user_id={user_id}"
cur.execute(sql)

Vous pouvez également vous appuyer sur les bind variables :

warehouse_name='ADMIN'
sql="""
select *
from snowflake.account_usage.warehouse_metering_history
where warehouse_name=%s
"""
cur.execute(sql, (warehouse_name))

Récupérer tous les résultats

Dans tous les exemples de cet article, nous nous sommes appuyés sur la méthode fetchall() pour récupérer l'ensemble des résultats de la requête :

cur.execute('select * from table')
results=cur.fetchall()

Traiter les données par lots

Pour limiter la mémoire consommée par votre application Python, vous pouvez traiter les données par lots. Voici un exemple qui traite 100 000 enregistrements à la fois :

cur.execute('select * from table')
while True:
    rows = cur.fetchmany(100000)
    if not rows:
        break
    for row in rows:
        # Traiter chaque ligne
        print(row)

Renvoyer une liste de dictionnaires

Par défaut, Snowflake renvoie une liste de tuples, sans préciser à quelle colonne correspond chaque entrée du tuple. Voici comment obtenir une liste de dictionnaires (un par ligne), où la clé est le nom de la colonne :

1sql = """

2	select

3		warehouse_name,

4		credits_used_compute

5	from snowflake.account_usage.warehouse_metering_history

6	limit 1

7"""

8cur.execute(sql)

9results = cur.fetchall()

10result_meta = cur.description

11results = [\
\
12    dict(zip([col.name for col in result_meta], row)) for row in results\
\
13]

14

15# >>> print(results)

Déplier le code

Exécuter plusieurs instructions SQL en une seule fois

La méthode cursor.execute n'accepte qu'une seule instruction SQL à la fois. Pour en exécuter plusieurs d'un coup, utilisez la méthode execute_string sur l'objet connection. Une liste de curseurs est alors renvoyée, que vous pouvez exploiter pour traiter les résultats si besoin, comme dans les exemples précédents.

sql="""
delete from table where user_id=123;
insert into table values(...);
"""
cursors = conn.execute_string(sql)

# renvoyer les résultats de chaque curseur si besoin

Définir les paramètres de session

Au moment d'établir votre connexion Snowflake, vous pouvez configurer aisément différents paramètres de session. Vous pouvez par exemple définir un query tag pour que toutes les requêtes lancées depuis votre connexion soient taguées, ou un query timeout pour que vos requêtes soient automatiquement annulées au-delà d'un certain délai.

conn = snowflake.connector.connect(
    user='XXXX',
    password='XXXX',
    account='XXXX',
    session_parameters={
        'QUERY_TAG': 'COST_ANALYTICS',
        'STATEMENT_TIMEOUT_IN_SECONDS': 3600, # annule automatiquement la requête au bout d'1 heure
    }
)

Vous pouvez également transmettre d'autres paramètres, par exemple warehouse, pour préciser sur quel virtual warehouse vos requêtes s'exécuteront. À défaut, elles s'exécuteront sur le warehouse par défaut associé à l'utilisateur.

La liste complète des paramètres de connexion est disponible ici.