Amazon Redshift Python Driver

The paramstyle determines how parameters are passed to SQL queries. It can be set globally at the module level or locally for a specific cursor.

• Module level: Setting redshift_connector.paramstyle = 'value' applies to all subsequent cursors unless overridden.
• Cursor level: Setting cursor.paramstyle = 'value' applies only to that specific cursor object.

Supported values: qmark, numeric, named, format, pyformat. (The default module-level value is format).

Parameter passing rules:

• For named and pyformat, parameters must be passed as a Python dictionary.
• For qmark, numeric, and format, parameters must be passed as a Python tuple or list.

# qmark
cursor.paramstyle = 'qmark'
sql = 'insert into foo(bar, jar) VALUES(?, ?)'
cursor.execute(sql, (1, "hello world"))

# numeric
cursor.paramstyle = 'numeric'
sql = 'insert into foo(bar, jar) VALUES(:1, :2)'
cursor.execute(sql, (1, "hello world"))

# named
cursor.paramstyle = 'named'
sql = 'insert into foo(bar, jar) VALUES(:p1, :p2)'
cursor.execute(sql, {"p1":1, "p2":"hello world"})

# format
cursor.paramstyle = 'format'
sql = 'insert into foo(bar, jar) VALUES(%s, %s)'
cursor.execute(sql, (1, "hello world"))

# pyformat
cursor.paramstyle = 'pyformat'
sql = 'insert into foo(bar, jar) VALUES(%(bar)s, %(jar)s)'
cursor.execute(sql, {"bar": 1, "jar": "hello world"})

The IdpTokenAuthPlugin allows for Identity Center authentication. You can use it in three ways:

1. Automatic Resolution: Use the default credential chain (environment variables, ~/.aws/credentials, etc.) to resolve credentials automatically.
2. Identity-Enhanced Credentials: Provide explicit AWS credentials (e.g., from AssumeRole) to the plugin.
3. Direct Token: Pass a pre-obtained subject token directly.

Note: For clusters using Identity Center, the token must be a subject token obtained from GetIdentityCenterAuthToken. Raw external IdP tokens (like Azure AD JWTs) are not supported.

import redshift_connector

# 1. Automatic Resolution
conn = redshift_connector.connect(
    host='my-cluster.abc123.us-east-1.redshift.amazonaws.com',
    database='dev',
    credentials_provider='IdpTokenAuthPlugin',
)

# 2. Identity-Enhanced Credentials
conn = redshift_connector.connect(
        host='my-cluster.abc123.us-east-1.redshift.amazonaws.com',
        database='dev',
        credentials_provider='IdpTokenAuthPlugin',
        access_key_id='my_access_key_id',
        secret_access_key='my_secret_access_key',
        session_token='my_session_token',
    )

# 3. Direct Token
conn = redshift_connector.connect(
        host='my-cluster.abc123.us-east-1.redshift.amazonaws.com',
        database='dev',
        credentials_provider='IdpTokenAuthPlugin',
        token='<subject_token_from_GetIdentityCenterAuthToken>',
        token_type='SUBJECT_TOKEN',
    )

IAM authentication allows for the use of temporary credentials and improved security. You can provide IAM credentials in two ways:

1. Using an AWS Profile: Specify the profile name (e.g., 'default') to pull credentials from your ~/.aws/credentials and ~/.aws/config files. When using this method, redshift_connector requires a region to be specified. If a region is provided in both the config file and the connect() call, the value in connect() takes precedence.

2. Directly providing credentials: Pass access_key_id, secret_access_key, and session_token directly to the connect() method.

In both cases, you must set iam=True.

import redshift_connector

# Option 1: Using an AWS Profile
conn = redshift_connector.connect(
    iam=True,
    database='dev',
    db_user='awsuser',
    password='',
    user='',
    cluster_identifier='examplecluster',
    profile='default'
)

# Option 2: Providing credentials directly
conn = redshift_connector.connect(
    iam=True,
    database='dev',
    db_user='awsuser',
    password='',
    user='',
    cluster_identifier='examplecluster',
    access_key_id="my_aws_access_key_id",
    secret_access_key="my_aws_secret_access_key",
    session_token="my_aws_session_token",
    region="us-east-2"
)

The redshift_connector supports the Amazon Redshift SUPER datatype for handling semistructured JSON data.

Sending Data

To send data to a SUPER column, pass a string containing JSON data. It is recommended to use the Redshift json_parse(%s) function within your SQL statement to convert the string into the SUPER type.

Receiving Data

SUPER data is received as a string containing the JSON data.

import redshift_connector 
import json

# Example: Creating a table with SUPER data using json_parse
with redshift_connector.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute(
            "CREATE TABLE foo AS SELECT json_parse(%s) AS multi_level_array;",
            ('[[1.1, 1.2], [2.1, 2.2], [3.1, 3.2]]',
        )
        cursor.execute("SELECT array, element FROM foo AS f, f.multi_level_array AS array, array AS element;")
        result = cursor.fetchall()
        print(result)

# Example: Retrieving specific elements from a JSON object stored in SUPER
with redshift_connector.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute(
            "CREATE TABLE foo AS SELECT json_parse(%s) AS vals;",
            (json.dumps({"x": [1,2,3,4], "y": [5,6,7,8], "z": [9,10,11,12]}),)
        )
        cursor.execute("SELECT vals.x FROM foo;")
        result = cursor.fetchall()
        print(result)

The redshift_connector supports Amazon Redshift geometry datatypes.

Sending Data

You can send geometry data in two primary ways:

1. WKB (Well-Known Binary) format: Pass a string representing the binary data directly into the query.
2. WKT (Well-Known Text) format: Use Redshift spatial functions like ST_GeomFromText(%s) to convert a WKT string into a geometry type.

Receiving Data

Geometry data is received as a string holding the data in WKB format.

import redshift_connector

# Example: Sending data in WKB format
with redshift_connector.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute("create table datatype_test (c1 geometry);")
        cursor.execute(
            "insert into datatype_test (c1) values (%s);",
            ('0103000020E61000000100000005000000000000000000000000000000000000000000000000000000000000000000F03F000000000000F03F000000000000F03F000000000000F03F000000000000000000000000000000000000000000000000',
        )
        cursor.execute("select c1 from datatype_test;")
        result = cursor.fetchone()
        print("c1={}\n".format(result[0]))

# Example: Sending data in WKT format
with redshift_connector.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute("create table datatype_test (c1 geometry);")
        cursor.execute(
            "insert into datatype_test (c1) values (ST_GeomFromText(%s));",
            ('LINESTRING(1 2,3 4,5 6,7 8,9 10,11 12,13 14,15 16,17 18,19 20)',)
        )
        cursor.execute("select c1 from datatype_test;")
        result = cursor.fetchone()
        print("c1={}".format(result[0]))