Snowflake in Python
Snowflake account connections, queries and warehouse options from Python services.
Zato lets you work with Snowflake directly from your Python services. You set up connection pools in the Dashboard, and Zato manages the connections, credentials and pooling.
Creating a connection
Open the Dashboard and go to Connections → Outgoing → SQL. Create a new connection, choose Snowflake, and fill in your account details.
The fields have Snowflake-specific meanings:
| Field | Description |
|---|---|
| Host | Your account identifier, e.g. myorg-myaccount - not a hostname |
| Port | Always 443 - Snowflake connections go over HTTPS |
| Database | Name of the Snowflake database, e.g. MYDB |
| User | Snowflake username |
Once saved, the connection becomes available to all your services by its configured name.
Executing queries
Call conn.execute with your SQL and a dictionary of parameters. Each row comes back as a dictionary in a list.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetRecentOrders(Service):
def handle(self):
# Get the connection by the name configured in Dashboard
conn = self.out.sql['My Snowflake']
# Query with named parameter
query = 'SELECT * FROM orders WHERE status = :status'
params = {'status': 'shipped'}
# Execute returns a list of dicts
result = conn.execute(query, params)
self.response.payload = result
Response:
[
{"order_id": 1, "status": "shipped", "total": 149.99},
{"order_id": 2, "status": "shipped", "total": 89.50}
]
Querying a single row
If your query should return exactly one row, conn.one gives you that row directly as a dictionary. It raises an exception when zero or multiple rows come back.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class GetOrderById(Service):
""" Returns a single order by its ID.
"""
input = 'order_id'
def handle(self):
# Get the connection from the pool
conn = self.out.sql['My Snowflake']
# Build the query
query = 'SELECT * FROM orders WHERE order_id = :order_id'
# Assign input data to query parameters
params = {'order_id': self.request.input.order_id}
# Run the query - returns a dict directly, raises if not found
order = conn.one(query, params)
# Return the result to the caller
self.response.payload = order
For cases where the row might not exist, conn.one_or_none returns None instead of raising an exception.
Extra options
When creating a connection in the Dashboard, you can provide extra options as a list of key=value pairs. Each option goes on its own line.
| Option | Description |
|---|---|
warehouse | Name of the virtual warehouse to use, e.g. COMPUTE_WH |
role | Name of the role to assume, e.g. ANALYST |
schema | Name of the schema within the database, e.g. PUBLIC |
login_timeout | Login timeout in seconds |
network_timeout | Network timeout in seconds |
Example in Dashboard:
TLS is always on with Snowflake - all connections go over HTTPS to your account's endpoint and there is nothing to configure to enable it.
How to avoid SQL injection attacks
Always use parameterized queries with named parameters (:param_name) rather than string formatting. This prevents SQL injection attacks and makes your queries more readable.
# Correct - parameterized query
query = 'SELECT * FROM users WHERE user_id = :user_id AND status = :status'
params = {'user_id': 123, 'status': 'active'}
result = conn.execute(query, params)
# Wrong - string formatting (vulnerable to SQL injection)
# query = f'SELECT * FROM users WHERE user_id = {user_id}'