用户在创建好数据仓库集群后使用psycopg2第三方库连接到集群，则可以使用Python访问DWS ，并进行数据表的各类操作。

连接集群前的准备

• DWS 集群已绑定弹性IP。
• 已获取DWS 集群的数据库管理员用户名和密码。

使用约束

由于psycopg2是基于PostgreSQL的客户端接口，它的功能DWS并不能完全支持。具体支持情况请见下表。

DWS对psycopg2主要接口支持情况

在Linux环境使用psycopg2第三方库连接集群

1.以root用户登录Linux环境。

2.执行以下命令创建python_dws.py文件。

vi python_dws.py复制

请复制粘贴以下内容放入python_dws.py文件中：

#!/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 
                                user='dbadmin', 
                                password='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.按照实际集群信息，修改python_dws.py文件中的集群公网访问地址、集群端口号、数据库名称、数据库用户名、数据库密码。

psycopg2接口不提供重试连接的能力，您需要在业务代码中实现重试处理。

conn = psycopg2.connect(host='10.154.70.231', 
                                port='8000', 
                                database='gaussdb',  # 需要连接的database 
                                user='dbadmin', 
                                password='password')  # 数据库用户密码复制

4.执行以下命令，使用psycopg第三方库连接集群。

python python_dws.py复制

在Windows环境使用psycopg2第三方库连接集群

1.在Windows系统中，单击“开始”按钮 ，在搜索框中，键入 cmd ，然后在结果列表中单击“cmd.exe”打开命令提示符窗口。

2.在命令提示符窗口中，执行以下命令创建python_dws.py文件。

type nul> python_dws.py复制

请复制粘贴以下内容放入python_dws.py文件中：

请复制粘贴以下内容放入python_dws.py文件中：
#!/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='postgresgaussdb',  # 需要连接的database 
                                user='dbadmin', 
                                password='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.按照实际集群信息，修改python_dws.py文件中的集群公网访问地址、集群端口号、数据库名称、数据库用户名、数据库密码。

conn = psycopg2.connect(host='10.154.70.231', 
                                port='8000', 
                                database='gaussdb',  # 需要连接的database 
                                user='dbadmin', 
                                password='password')  # 数据库用户密码复制

4.在命令提示符窗口中，执行以下命令，使用psycopg第三方库连接集群。

python python_dws.py复制

psycopg2连接集群不支持CN Retry特性的问题说明

DWS支持在SQL语句执行出错时的自动重试功能（简称CN Retry）。CN Retry对于客户端和驱动发送的SQL语句在执行失败时可以自动识别错误类型，并进行重试。但使用psycopg2默认连接方式创建的连接在语句执行失败时没有自动重试，会直接报错退出。如常见的主备切换场景下，未自动重试会报如下错误，但在自动重试期间完成主备切换，则会返回正确结果。

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复制

报错原因：

1.psycopg2在发送SQL语句前先发送了BEGIN语句开启事务。

2.CN Retry不支持事务块中的语句是特性约束。

解决方案：

• 在同步方式连接时，可以通过主动结束驱动开启的事务。

cursor = conn.cursor() 
# 增加end语句主动结束驱动开启的事务 
cursor.execute("end; select * from test order by 1;")  
rows = cursor.fetchall()复制

• 使用异步连接方式主动开启事务，异步连接介绍具体请参见pyscopg官网：https://www.psycopg.org/docs/advanced.html?highlight=async。

#!/usr/bin/env python3 
# _*_ encoding=utf-8 _*_ 
  
import psycopg2 
import select 
  
# psycopg2官方提供的异步连接方式时的wait函数 
# 详见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_sync(): 
    # 创建连接 
    conn = psycopg2.connect(host='10.154.70.231', 
                                port='8000', 
                                database='gaussdb',  # 需要连接的database 
                                user='dbadmin', 
                                password='password',  # 数据库用户密码 
                                async=1) # 使用异步方式连接 
    wait(conn) 
  
    # 执行查询 
    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]) 
  
    # 关闭连接 
    conn.close() 
  
if __name__ == '__main__': 
    psycopg2_cnretry_async()