使用的模块为psycopg2

安装方式:

pip3 install psycopg2-binary

模块测试

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import psycopg2

#   PostgreSQL 数据入库模块 正确执行返回Y 否则返回N  增 / 删 / 改
def runDataInPostgreSQLDB(aHost, aPort, aDB, aUser, aPass, aSQL):
    """ Connect to the PostgreSQL database server. insert/delete/update"""
    conn = None
    try:
        print(' + Database connection START')
        conn = psycopg2.connect(dbname = aDB, user = aUser, password = aPass, host = aHost, port = aPort)		
        cur = conn.cursor()
        cur.execute(aSQL)
        conn.commit()
        cur.close()
        return 'Y'
    except (Exception, psycopg2.DatabaseError) as e:
        print(e)
        return 'N'
    finally:
        if conn is not None:
            conn.close()
            print(' + Database connection CLOSE')

#   PostgreSQL 数据读取模块 正确返回数据列表 错误返回空列表
def runDataOutPostgreSQLDB(aHost, aPort, aDB, aUser, aPass, aSQL):
    """ Connect to the PostgreSQL database server. select"""
    conn = None
    try:
        print(' + Database connection START')
        conn = psycopg2.connect(dbname = aDB, user = aUser, password = aPass, host = aHost, port = aPort)		
        cur = conn.cursor()
        cur.execute(aSQL)       
        data = cur.fetchall()
        cur.close()
        return data
    except (Exception, psycopg2.DatabaseError) as e:
        print(e)
        return []
    finally:
        if conn is not None:
            conn.close()
            print(' + Database connection CLOSE')
            
aSQL = "select vtype,ip,port from mdata limit 3;"            
vData = runDataOutPostgreSQLDB(aHost = '127.0.0.1', aPort = 5432, aDB = "你的数据库实例名称", aUser = "你的数据库用户名", aPass = "你的数据库密码", aSQL = "一个SQL")  
print(vData)    

// 测试结果

root@PostgreSQLDatabase:~/src/invoker# python3 tmp1058.py 
 + Database connection START
 + Database connection CLOSE
[('socks4', '203.207.52.38', 5430), ('socks4', '103.146.170.244', 5678)]
root@PostgreSQLDatabase:~/src/invoker# 

如果用命令行连接数据库
# psql -d 你的数据库实例名称 -h 127.0.0.1 -p 5432 -U 你的数据库用户名 -W