Amazon Redshift in Python
Redshift cluster connections, queries and TLS options from Python services.
Zato lets you work with Amazon Redshift 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 Amazon Redshift, and fill in your cluster details.
| Field | Description |
|---|---|
| Host | Your cluster endpoint, e.g. examplecluster.abc123xyz789.us-west-2.redshift.amazonaws.com |
| Port | Cluster port, 5439 by default |
| Database | Name of the database within the cluster |
| User | Redshift 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 GetDailySales(Service):
def handle(self):
# Get the connection by the name configured in Dashboard
conn = self.out.sql['My Redshift']
# Query with named parameter
query = 'SELECT * FROM sales WHERE region = :region'
params = {'region': 'EMEA'}
# Execute returns a list of dicts
result = conn.execute(query, params)
self.response.payload = result
Response:
[
{"sale_id": 1, "region": "EMEA", "amount": 1200.00},
{"sale_id": 2, "region": "EMEA", "amount": 950.00}
]
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 GetSaleById(Service):
""" Returns a single sale by its ID.
"""
input = 'sale_id'
def handle(self):
# Get the connection from the pool
conn = self.out.sql['My Redshift']
# Build the query
query = 'SELECT * FROM sales WHERE sale_id = :sale_id'
# Assign input data to query parameters
params = {'sale_id': self.request.input.sale_id}
# Run the query - returns a dict directly, raises if not found
sale = conn.one(query, params)
# Return the result to the caller
self.response.payload = sale
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 |
|---|---|
ssl | Whether to use TLS, True by default |
sslmode | TLS mode - verify-ca (default) or verify-full |
timeout | Connection timeout in seconds |
tcp_keepalive | Whether to use TCP keepalives, e.g. True |
Example in Dashboard:
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}'