Help Center/ Data Warehouse Service/ User Guide/ Connecting to a DWS Cluster/ Using a Third-Party Database Adapter for DWS Cluster Connection/ Using the Python Library psycopg2 to Connect to a DWS Cluster
Updated on 2026-10-16 GMT+08:00
View PDF
Share

Using the Python Library psycopg2 to Connect to a DWS Cluster

You can use the Python library psycopg2 to connect to a DWS cluster you have created, access DWS using Python, and perform operations on data tables, such as creating tables, inserting data, querying data, updating data, and deleting data.

Application Scenarios

psycopg2 applies to the following scenarios:

  • You want to use the web applications or data processing scripts developed using Python to operate data in DWS.
  • You want to connect existing Python applications to DWS and reuse the mature capabilities of the psycopg2 ecosystem.
  • You want to use Python to create data tables or add, delete, modify, or query data.

Example: For example, the backend service of a web application uses psycopg2 to query service data in DWS and return the data to the frontend, or write user operation records to DWS in batches.

Preparations Before Connecting to a Cluster

Before connecting to a cluster, ensure that you have completed the following preparations:

  • You have created a DWS cluster and bound an EIP to the cluster.
  • You have obtained the administrator username and password for logging in to the database in the DWS cluster.
    • The MD5 algorithm has collision risks. By default, MD5 password verification is not allowed by DWS. If an authentication failure is displayed when you use an open-source client to connect to the cluster, the most likely cause is that the client uses MD5 authentication. In this case, perform the following steps:
      1. Contact technical support to check whether the value of the database parameter password_encryption_type is 1. If the value is not 1, change it to 1.
      2. Change the password of the database user to be used or create another database user.
    • The database stores the hash digest of passwords instead of password text. During password verification, the system compares the hash digest with the password digest sent from the client (salt operations are involved). If the cryptographic algorithm policy changes, the database cannot automatically generate a digest of the password based on the new algorithm. In this case, you must change the password or create another user. A digest will then be generated for the new password based on the new algorithm for the next connection authentication.
  • You have obtained the public IP address and port number of the DWS cluster. For details, see Obtaining the Connection Address of a DWS Cluster.

Installing psycopg2

Download psycopg2 from https://pypi.org/project/psycopg2/ and install it by referring to https://www.psycopg.org/install/. The installation methods for different OSs are as follows:

  • In CentOS and Red Hat OS, run the following yum command: 
    1
    		 
    yum install python-psycopg2
  • psycopg2 depends on the libpq dynamic library of PostgreSQL (32-bit or 64-bit version, whichever matches the psycopg2 bit version). In Linux, you can run the yum command and do not need to install the library. In Windows, you can install libpq using either of the following methods:
    • Install PostgreSQL and add the directories of the libpq, SSL, and crypto dynamic libraries to the environment variable PATH.
    • Install psqlodbc and use the libpq, SSL, and crypto dynamic libraries provided by the PostgreSQL ODBC driver.

Version

There are many versions of DWS clusters, Python, and Psycopg2. The following table lists only the supported mainstream versions.

Table 1 Supported versions

Psycopg2 Version

Python Version

DWS Cluster Version

2.7.x

3.8.x

8.1.3 or later

3.9.x

8.1.3 or later

2.8.x

3.8.x

8.1.3 or later

3.9.x

8.1.3 or later

2.9.x

3.8.x

8.1.3 or later

3.9.x

8.1.3 or later

Constraints

Psycopg2 is a PostgreSQL-based client interface, and its functions are not fully supported by DWS. For details, see Table 2. The following APIs are supported based on Python 3.8.5 and Psycopg 2.9.1.

Table 2 Psycopg2 APIs supported by DWS

Class Name

Usage

Function/Member Variable

Support

Remarks

connections

basic

cursor(name=None, cursor_factory=None, scrollable=None, withhold=False)

Y

-

commit()

Y

-

rollback()

Y

-

close()

Y

-

Two-phase commit support methods

xid(format_id, gtrid, bqual)

Y

-

tpc_begin(xid)

Y

-

tpc_prepare()

N

The kernel does not support explicit PREPARE TRANSACTION.

tpc_commit([xid])

Y

-

tpc_rollback([xid])

Y

-

tpc_recover()

Y

-

closed

Y

-

cancel()

Y

-

reset()

N

DISCARD ALL is not supported.

dsn

Y

-

Transaction control methods and attributes.

set_session(isolation_level=None, readonly=None, deferrable=None, autocommit=None)

Y

The database does not support the setting of default_transaction_read_only in a session.

autocommit

Y

-

isolation_level

Y

-

readonly

N

The database does not support the setting of default_transaction_read_only in a session.

deferrable

Y

-

set_isolation_level(level)

Y

-

encoding

Y

-

set_client_encoding(enc)

Y

-

notices

N

The database does not support listen/notify.

notifies

Y

-

cursor_factory

Y

-

info

Y

-

status

Y

-

lobject

N

The database does not support operations related to large objects.

Methods related to asynchronous support

poll()

Y

-

fileno()

Y

-

isexecuting()

Y

-

Interoperation with other C API modules

pgconn_ptr

Y

-

get_native_connection()

Y

-

informative methods of the native connection

get_transaction_status()

Y

-

protocol_version

Y

-

server_version

Y

-

get_backend_pid()

Y

The obtained PID is not the background PID, but the ID of the logical connection.

get_parameter_status(parameter)

Y

-

get_dsn_parameters()

Y

-

cursor

basic

description

Y

-

close()

Y

-

closed

Y

-

connection

Y

-

name

Y

-

scrollable

N

The database does not support SCROLL CURSOR.

withhold

N

The withhold cursor needs to be closed before the commit operation.

Commands execution methods

execute(query, vars=None)

Y

-

executemany(query, vars_list)

Y

-

callproc(procname[, parameters])

Y

-

mogrify(operation[, parameters])

Y

-

setinputsizes(sizes)

Y

-

fetchone()

Y

-

fetchmany([size=cursor.arraysize])

Y

-

fetchall()

Y

-

scroll(value[, mode='relative'])

N

The database does not support SCROLL CURSOR.

arraysize

Y

-

itersize

Y

-

rowcount

Y

-

rownumber

Y

-

lastrowid

Y

-

query

Y

-

statusmessage

Y

-

cast(oid, s)

Y

-

tzinfo_factory

Y

-

nextset()

Y

-

setoutputsize(size[, column])

Y

-

COPY-related methods

copy_from(file, table, sep='\\t', null='\\\\N', size=8192, columns=None)

Y

-

copy_to(file, table, sep='\\t', null='\\\\N', columns=None)

Y

-

copy_expert(sql, file, size=8192)

Y

-

Interoperation with other C API modules

pgresult_ptr

Y

-

Using the Third-Party Function Library psycopg2 to Connect to a Cluster (Linux)

  1. Log in to the Linux environment as user root.
  2. Run the following command to create the python_dws.py file: 

    vi python_dws.py

    Copy and paste the following content to the python_dws.py file:

      1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
    		 
    #!/usr/bin/python
# -*- coding: UTF-8 -*-
 
from __future__ import print_function
 
import psycopg2
 
 
def create_table(connection):
    print("Begin to create table")
    try:
        cursor = connection.cursor()
        cursor.execute("drop table if exists test;"
                       "create table test(id int, name text);")
        connection.commit()
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("Table created successfully")
        cursor.close()
 
 
def insert_data(connection):
    print("Begin to insert data")
    try:
        cursor = connection.cursor()
        cursor.execute("insert into test values(1,'number1');")
        cursor.execute("insert into test values(2,'number2');")
        cursor.execute("insert into test values(3,'number3');")
        connection.commit()
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("Insert data successfully")
        cursor.close()
 
 
def update_data(connection):
    print("Begin to update data")
    try:
        cursor = connection.cursor()
        cursor.execute("update test set name = 'numberupdated' where id=1;")
        connection.commit()
        print("Total number of rows updated :", cursor.rowcount)
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("After Update, Operation done successfully")
 
 
def delete_data(connection):
    print("Begin to delete data")
    try:
        cursor = connection.cursor()
        cursor.execute("delete from test where id=3;")
        connection.commit()
        print("Total number of rows deleted :", cursor.rowcount)
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("After Delete,Operation done successfully")
 
 
def select_data(connection):
    print("Begin to select data")
    try:
        cursor = connection.cursor()
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
        print("select failed")
    else:
        print("Operation done successfully")
        cursor.close()
 
 
if __name__ == '__main__':
    try:
        conn = psycopg2.connect(host='10.154.70.231',
                                port='8000',
                                database='gaussdb',  # Database to be connected
                                user='dbadmin',
                                password='password')  # Database user password
    except psycopg2.DatabaseError as ex:
        print(ex)
        print("Connect database failed")
    else:
        print("Opened database successfully")
        create_table(conn)
        insert_data(conn)
        select_data(conn)
        update_data(conn)
        delete_data(conn)
        conn.close()

  3. Change the public network address, cluster port number, database name, database username, and database password in the python_dws.py file based on the actual cluster information.

    The psycopg2 API does not provide the connection retry capability. You need to implement the retry processing in the service code.

    1
2
3
4
5
    		 
    conn = psycopg2.connect(host='10.154.70.231',
                                port='8000',
                                database='gaussdb',  # Database to be connected
                                user='dbadmin',
                                password='password')  # Database user password
    Table 3 Parameters for connecting to a cluster

    Parameter

    Description

    host

    Public IP address (EIP) for accessing the DWS cluster. Replace it with the actual address. For how to obtain the address, see Table 4.

    port

    Database port number of the DWS cluster. The default port number is 8000.

    database

    Name of the database to connect to. The default database is gaussdb.

    user

    Database username. The default database administrator is dbadmin.

    password

    Password of the database user. Replace it with the actual password.

  4. Connect to the cluster using the third-party database adapter Psycopg. 

    python python_dws.py

  5. If the command output indicates that the database is successfully connected and messages indicating successful table creation, data insertion, query, update, and deletion are displayed in sequence, the database connection is normal and the operations on data tables have been successfully executed.

Using the Third-Party Function Library psycopg2 to Connect to a Cluster (Windows)

  1. In the Windows operating system, click Start, enter cmd in the search box, and click cmd.exe in the result list to open the command-line interface (CLI).
  2. In the CLI, run the following command to create the python_dws.py file: 

    type nul> python_dws.py

    Copy and paste the following content to the python_dws.py file:

      1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
    		 
    #!/usr/bin/python
# -*- coding:UTF-8 -*-

from __future__ import print_function

import psycopg2


def create_table(connection):
    print("Begin to create table")
    try:
        cursor = connection.cursor()
        cursor.execute("drop table if exists test;"
                       "create table test(id int, name text);")
        connection.commit()
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("Table created successfully")
        cursor.close()


def insert_data(connection):
    print("Begin to insert data")
    try:
        cursor = connection.cursor()
        cursor.execute("insert into test values(1,'number1');")
        cursor.execute("insert into test values(2,'number2');")
        cursor.execute("insert into test values(3,'number3');")
        connection.commit()
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("Insert data successfully")
        cursor.close()


def update_data(connection):
    print("Begin to update data")
    try:
        cursor = connection.cursor()
        cursor.execute("update test set name = 'numberupdated' where id=1;")
        connection.commit()
        print("Total number of rows updated :", cursor.rowcount)
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("After Update, Operation done successfully")


def delete_data(connection):
    print("Begin to delete data")
    try:
        cursor = connection.cursor()
        cursor.execute("delete from test where id=3;")
        connection.commit()
        print("Total number of rows deleted :", cursor.rowcount)
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
    else:
        print("After Delete,Operation done successfully")


def select_data(connection):
    print("Begin to select data")
    try:
        cursor = connection.cursor()
        cursor.execute("select * from test order by 1;")
        rows = cursor.fetchall()
        for row in rows:
            print("id = ", row[0])
            print("name = ", row[1], "\n")
    except psycopg2.ProgrammingError as e:
        print(e)
        print("select failed")
    else:
        print("Operation done successfully")
        cursor.close()


if __name__ == '__main__':
    try:
        conn = psycopg2.connect(host='10.154.70.231',
                                port='8000',
                                database='gaussdb',  # Database to be connected
                                user='dbadmin',
                                password='password')  # Database user password
    except psycopg2.DatabaseError as ex:
        print(ex)
        print("Connect database failed")
    else:
        print("Opened database successfully")
        create_table(conn)
        insert_data(conn)
        select_data(conn)
        update_data(conn)
        delete_data(conn)
        conn.close()

  3. Change the public IP address, port number, database name, database username, and database password of the cluster in the python_dws.py file based on the actual cluster information. For details about the parameters, see Table 3. 

    1
2
3
4
5
    		 
    conn = psycopg2.connect(host='10.154.70.231',
                        port='8000',
                                database='gaussdb'',  # Database to be connected
                        user='dbadmin',
                                password='password')  # Database user password

  4. On the CLI, run the following command to use Psycopg to connect to the cluster: 

    python python_dws.py

  5. If the command output indicates that the database is successfully connected and messages indicating successful table creation, data insertion, query, update, and deletion are displayed in sequence, the database connection is normal and the operations on data tables have been successfully executed.

Why Is CN Retry Not Supported When Psycopg2 Is Connected to a Cluster?

With the CN retry feature, DWS retries a statement that failed to be executed and identifies the failure type. For details, see Automatic Retry upon SQL Statement Execution Errors. However, in a session connected using Psycopg2, a failed SQL statement will report an error and stop being executed. In a primary/standby switchover, if a failed SQL statement is not retried, the following error will be reported. If the switchover is complete during an automatic retry, the correct result will be returned.

1
 
psycopg2.errors.ConnectionFailure: pooler: failed to create 1 connections, Error Message: remote node dn_6003_6004, detail: could not connect to server: Operation now in progress

Error causes:

  1. Psycopg2 sends the BEGIN statement to start a transaction before sending an SQL statement.
  2. CN retry does not support statements in transaction blocks.

Solution:

  • In synchronous connection mode, end the transaction started by the driver. 
    1
2
3
4
    		 
    cursor = conn.cursor()
# End the transaction started by the driver.
cursor.execute("end; select * from test order by 1;") 
rows = cursor.fetchall()
  • Start a transaction in an asynchronous connection. For details, visit the Psycopg official website at: https://www.psycopg.org/docs/advanced.html?highlight=async 
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
    		 
    #!/usr/bin/env python3
# _*_ encoding=utf-8 _*_
 
import psycopg2
import select
 
# Wait function provided by Psycopg2 in asynchronous connection mode
#For details, see https://www.psycopg.org/docs/advanced.html?highlight=async.
def wait(conn):
    while True:
        state = conn.poll()
        if state == psycopg2.extensions.POLL_OK:
            break
        elif state == psycopg2.extensions.POLL_WRITE:
            select.select([], [conn.fileno()], [])
        elif state == psycopg2.extensions.POLL_READ:
            select.select([conn.fileno()], [], [])
        else:
            raise psycopg2.OperationalError("poll() returned %s" % state)
 
def psycopg2_cnretry_async():
    # Create a connection.
    conn = psycopg2.connect(host='10.154.70.231',
                                port='8000',
                                database='gaussdb',  # Database to be connected
                                user='dbadmin',
                                password='password',  # Database user password
                                async=1) # Use the asynchronous connection mode.
    wait(conn)
 
    # Execute a query.
    cursor = conn.cursor()
    cursor.execute("select * from test order by 1;")
    wait(conn)
    rows = cursor.fetchall()
    for row in rows:
        print(row[0], row[1])
 
    # Close the connection.
    conn.close()
 
if __name__ == '__main__':
    psycopg2_cnretry_async()
Parent Topic: Using a Third-Party Database Adapter for DWS Cluster Connection