Cómo conectar DuckDB a Google Sheets

Instala los 3 paquetes que necesita este pipeline: duckdb para el motor SQL, gspread para el wrapper de la API de Sheets, y google-auth para autenticación de cuenta de servicio. bash pip install duckdb gspread google-auth

Una cuenta de servicio permite que tu script se autentique como una identidad no humana que puede leer y escribir hojas de cálculo sin interacción del usuario - esencial para scripts que se ejecutan en horarios o en servidores.

La cuenta de servicio tiene una dirección de correo (visible en el archivo JSON bajo client_email, formato: [email protected]). Google Sheets la trata como cualquier otro usuario.

Google Sheets devuelve datos como un arreglo 2D de strings. El código a continuación obtiene una hoja de trabajo nombrada, la convierte en una lista de diccionarios usando la primera fila como encabezados, y luego la registra como una relación DuckDB. python import duckdb import gspread from google.oauth2.service_account import Credentials SCOPES = ["https://www.googleapis.com/auth/spreadsheets"] SPREADSHEET_ID = "your-spreadsheet-id-here" WORKSHEET_NAME = "Sheet1" # Autentica y abre la hoja de cálculo creds = Credentials.from_service_account_file("credentials.json", scopes=SCOPES) gc = gspread.authorize(creds) sheet = gc.open_by_key(SPREADSHEET_ID).worksheet(WORKSHEET_NAME) # Obtén todos los registros como lista de diccionarios records = sheet.get_all_records() # Usa fila 1 como encabezados de columna automáticamente # Carga en DuckDB con = duckdb.connect() con.register("sheet_data", duckdb.from_arrow( __import__("pyarrow").Table.from_pylist(records) ))

Una vez que los datos están registrados, escribes SQL estándar. DuckDB soporta funciones de ventana, CTEs, joins LATERAL y la mayoría de sintaxis compatible con PostgreSQL - ninguno de los cuales existe en fórmulas de Sheets. python # Ejemplo de agregación: ingresos totales por región result = con.execute(""" SELECT region, COUNT(*) AS deal_count, SUM(CAST(revenue AS DOUBLE)) AS total_revenue, AVG(CAST(revenue AS DOUBLE)) AS avg_deal_size FROM sheet_data WHERE status = 'Closed Won' GROUP BY region ORDER BY total_revenue DESC """).fetchdf() print(result)

Después de ejecutar tu análisis, devuelve el DataFrame de resultado a una hoja de trabajo nueva o existente. El patrón a continuación limpia la hoja de destino y la rescribe completamente - confiable para trabajos programados, aunque no ideal para hojas que las personas están editando activamente. python # Crea o limpia la hoja de trabajo de salida try: output_sheet = gc.open_by_key(SPREADSHEET_ID).worksheet("DuckDB Output") output_sheet.clear() except gspread.exceptions.WorksheetNotFound: output_sheet = gc.open_by_key(SPREADSHEET_ID).add_worksheet( title="DuckDB Output", rows=1000, cols=20 ) # Escribe encabezados + datos headers = result.columns.tolist() rows = result.fillna("").values.tolist() output_sheet.update([headers] + rows)

Un script de una sola vez es útil. Uno programado es un pipeline de datos. Envuelve la lógica en una función y ejecútala en un horario cron para mantener tu hoja de salida actualizada. python import schedule import time def refresh_analysis(): records = sheet.get_all_records() con.register("sheet_data", duckdb.from_arrow( __import__("pyarrow").Table.from_pylist(records) )) result = con.execute(""" SELECT region, SUM(revenue) as total FROM sheet_data GROUP BY region """).fetchdf() output_sheet.clear() output_sheet.update([result.columns.tolist()] + result.fillna("").values.tolist()) print(f"Actualizado a las {__import__('datetime').datetime.now()}") schedule.every(30).minutes.do(refresh_analysis) while True: schedule.run_pending() time.sleep(60)