Snowflake Connector para Python en marcha

El Snowflake Connector es un paquete nativo de Python, así que se instala igual que cualquier otro paquete de Python.

Con pip

La forma más sencilla de instalar la librería es con pip, el sistema oficial de gestión de paquetes de Python:

1pip install snowflake-connector-python

Con conda

Si usas conda como gestor de paquetes, puedes instalarlo así:

1conda install snowflake-connector-python

Aquí tienes un ejemplo de principio a fin de cómo usar el paquete Snowflake Connector para ejecutar una consulta y procesar los resultados. Este ejemplo muestra el consumo total de créditos por virtual warehouse en los últimos 30 días:

Los datos que se imprimirán en la salida se verán más o menos así:

('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'))

Para encontrar tu account identifier, sigue los pasos de este artículo

Objetos connection y cursor

Como viste en el ejemplo rápido anterior, hay dos objetos distintos con los que se interactúa: el objeto connection y el objeto cursor.

El objeto connection se encarga de autenticar y conectarse a Snowflake, además de devolver cursores. Los cursores, por su parte, se usan para ejecutar consultas contra Snowflake y obtener los resultados. Una sola connection puede tener varios cursores. Un mismo cursor sirve para ejecutar varias consultas; no hace falta crear uno nuevo cada vez. Lo mismo aplica al objeto connection: abrir una conexión a Snowflake toma unos segundos, así que conviene hacerlo una sola vez.

Puedes conocer más sobre los distintos métodos disponibles en cada objeto en la documentación oficial de la API del conector de Python de Snowflake.

Consultas síncronas vs. asíncronas

Hay dos formas de ejecutar consultas con el Snowflake Connector para Python: de manera síncrona o asíncrona.

Enviar una consulta síncrona

Cuando envías una consulta síncrona, tu proceso de Python queda en espera hasta que la consulta devuelve resultados. Por su simplicidad, es el enfoque más común al trabajar con el conector de Python:

cur = conn.cursor()
cur.execute('select * from table') # Python esperará hasta que la consulta termine antes de pasar a la siguiente línea
results = cur.fetchall()

Enviar una consulta asíncrona

Con una consulta asíncrona, el paquete de Python de Snowflake devuelve el control de inmediato a tu proceso de Python antes de que la consulta termine. Esto resulta especialmente útil cuando construyes aplicaciones multihilo en Python (como una aplicación web) o cuando usas Python para ejecutar muchas consultas distintas a la vez.

Para ejecutar una consulta async, usa el método execute_async del cursor en lugar del método execute habitual. Más adelante en tu código, tendrás que consultarle a Snowflake si la consulta ya terminó.

cur = conn.cursor()
cur.execute_async('select * from table') # Python pasará inmediatamente a la siguiente línea
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()

Control de transacciones

Por defecto, cuando ejecutas consultas en Snowflake mediante el paquete Snowflake Connector, las consultas se confirman automáticamente.

Para tener un control más detallado sobre tus transacciones SQL, puedes definir autocommit=False al abrir la conexión.

Usando el context manager with

Mira el siguiente código. Si alguna de las sentencias SQL falla, Snowflake las revertirá automáticamente. Si tienen éxito, se confirmarán de forma automática al salir del bloque 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(...)")

Control manual con try-except

Para un control aún más detallado, puedes confirmar o revertir manualmente las transacciones SQL con los métodos conn.commit() y conn.close() junto con el patrón 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()

Cerrar la conexión

Como buena práctica, conviene cerrar la conexión al final de tu script de Python con el método conn.close(). Para asegurarte de que la conexión se cierre incluso si tu script falla, puedes envolver el código en un bloque try - finally:

try:
   cur.execute('...')
   results = cur.fetchall()
   # ...
except:
    # manejar errores según se requiera
finally:
		conn.close()

Otro patrón útil es recurrir al with() de Python, que cierra la conexión automáticamente:

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

# el método conn.close() se ejecuta automáticamente después del bloque with
# Continúa con otras operaciones que no involucran a Snowflake
with open('filename.csv', 'w') as f:
		f.write(...)

Veamos algunos ejemplos adicionales que destacan otros patrones comunes con el Snowflake Python Connector.

Pasar parámetros a tu sentencia SQL

Una necesidad habitual entre desarrolladores es poder pasar distintos parámetros a las sentencias SQL.

Python tiene un excelente soporte para manipulación de strings con f-strings, que puedes usar justamente para esto:

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

Como alternativa, puedes apoyarte en las bind variables:

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

Obtener todos los resultados

En todos los ejemplos de este artículo hemos usado el método fetchall() para devolver todos los resultados de la consulta:

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

Procesar por lotes

Si quieres reducir al mínimo la memoria que consume tu aplicación de Python, puedes procesar los datos por lotes. Aquí tienes un ejemplo que procesa 100 mil registros a la vez:

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

Devolver una lista de diccionarios

Por defecto, Snowflake devuelve una lista de tuplas y no sabrás a qué columna corresponde cada entrada de la tupla. Aquí tienes un ejemplo que muestra cómo devolver una lista de diccionarios por cada fila, donde la clave es el nombre de la columna:

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)

Expandir código

Ejecutar varias sentencias SQL a la vez

Al usar el método cursor.execute, solo puedes pasar una única sentencia SQL. Para ejecutar varias sentencias SQL de una sola vez, puedes recurrir al método execute_string del objeto connection. Te devolverá una lista de cursores que puedes aprovechar para procesar los resultados si hace falta, tal como en los ejemplos anteriores.

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

# devolver los resultados de cada cursor si se requiere

Configurar parámetros de sesión

Al abrir tu conexión a Snowflake puedes configurar distintos parámetros de sesión de forma muy práctica. Por ejemplo, definir un query tag para que todas las consultas ejecutadas desde tu conexión queden etiquetadas, o un query timeout para que se cancelen automáticamente tras cierto tiempo.

conn = snowflake.connector.connect(
    user='XXXX',
    password='XXXX',
    account='XXXX',
    session_parameters={
        'QUERY_TAG': 'COST_ANALYTICS',
        'STATEMENT_TIMEOUT_IN_SECONDS': 3600, # cancela automáticamente la consulta después de 1 hora
    }
)

También puedes pasar otros parámetros como warehouse para controlar en qué virtual warehouse se ejecutarán tus consultas. Si no lo especificas, se ejecutarán en el warehouse predeterminado de ese usuario.

Puedes ver la lista completa de parámetros de conexión aquí.