python3操作pg库

安装库

pip install psycopg2

如果报错可以尝试升级pip

python -m pip install --upgrade setupTools
python -m pip install --upgrade pip

DEMO

import psycopg2
from constant import const

class pgutil():
    # env :  dev 开发  prod 生产  默认开发
    def __init__(self, env):
        # 从常量配置类中读取
        pgcfg = const.DB_DEV_PG
        self.env = 'dev'
        if env =='prod':
            pgcfg = const.DB_PROD_PG
            self.env = env
        self.dbname = pgcfg.get('dbname')
        self.ip = pgcfg.get('host')
        self.user = pgcfg.get('user')
        self.passwd = pgcfg.get('passwd')
        self.dbport = pgcfg.get('port')

    def connectDb(self):
        self.conn = psycopg2.connect(database=self.dbname, user=self.user, password=self.passwd, host=self.ip,
                                     port=self.dbport)
        print('成功连接数据库：' + self.dbname )
        if self.env == 'prod':
            print('当前数据库为生产环境数据库,请谨慎处理')
        self.cur = self.conn.cursor()  # 创建指针对象

    # @exeDec
    def exeQuerySql(self, sql):
        try:
            # 生产环境防止误操作这里仅支持查询操作
            if self.env == 'prod':
                opt_dangerous = ['UPDATE','INSERT','DELETE','TRUNCATE']
                if any(key in str(sql).upper() for key in opt_dangerous):
                    print("当前连接为生产库，仅支持查询操作！！！")
                    return ''
            self.connectDb()
            self.cur.execute(sql)
            self.conn.commit()
            res = self.cur.fetchall()
            self.quitDb()
            return res
        except Exception as e:
            print('数据库脚本执行异常：{}'.format(e))

    def quitDb(self):
        self.cur.close()
        self.conn.close()

    def exeNormalSql(self, sql):
        try:
            # 生产环境防止误操作这里仅支持查询操作
            if self.env == 'prod':
                opt_dangerous = ['UPDATE', 'INSERT', 'DELETE', 'TRUNCATE']
                if any(key in str(sql).upper() for key in opt_dangerous):
                    print("当前连接为生产库，仅支持查询操作！！！")
                    return ''
            # self.connectDb()
            self.cur.execute(sql)
            # self.conn.commit()
            # self.quitDb()
        except Exception as e:
            print('数据库脚本执行异常：{}'.format(e))

    def commit(self):
        try:
            self.conn.commit()
        except Exception as e:
            print('数据库commit：{}'.format(e))


if __name__ == '__main__':
    # conn = psycopg2.connect('dev')
    # # 创建指针对象
    # cur = conn.cursor()
    # sql = """SELECT * FROM setl_d limit 1;"""
    # cur.execute(sql)
    #
    # res = cur.fetchall()
    # print(res)
    pgutil = pgutil('dev')
    sql = """SELECT * FROM setl_d limit 1;"""
    res = pgutil.exeQuerySql(sql)
    print(res)