基于Python3的接口自动化总结（六）——PostgreSQL数据库

通过python连接PostgreSQL数据库

import psycopg2
import psycopg2.extras


class Database:
    def __init__(self, database, user, password, host, port):
        self.database = database
        self.user = user
        self.password = password
        self.host = host
        self.port = port
        self.connection = None

    def _connect(self):
        self.connection = psycopg2.connect(database=self.database, user=self.user, password=self.password,
                                           host=self.host, port=self.port)
        print(f"连接数据库 postgres://{self.host}:{self.port}/{self.database}成功,用户名:{self.user}")

    def connect(self):
        try:
            if self.connection:
                conn_info = self.connection.options
                if conn_info['database'] == self.database and conn_info['user'] == self.user and conn_info[
                    'password'] == self.password and conn_info['host'] == self.host and conn_info['port'] == self.port:
                    print(f"返回已存在的数据库连接 postgres://{self.host}:{self.port}/{self.database}成功,用户名:{self.user}")
                else:
                    self._connect()
            else:
                self._connect()
        except Exception as e:
            print(f"连接数据库 postgres://{self.host}:{self.port}/{self.database}失败,用户名:{self.user},异常信息:{e.args[0]}")

    def close(self):
        if self.connection:
            self.connection.close()
            self.connection = None
            print(f"关闭据库成功")

    def execute_sql(self, sql):  # 数据库操作，无返回值
        """
        数据库操作，无返回值
            :param sql: 数据库插入语句
            :return:
        """
        self.connect()
        try:
            cur = self.connection.cursor()
            cur.execute(sql)
            self.connection.commit()
            print(f"执行sql语句成功:{sql}")
            cur.close()
        except Exception as e:
            print(f"执行sql语句失败:{sql},异常信息:{e.args[0]}")
        finally:
            self.close()

    def execute_sql_list(self, sql):
        """
        数据库查询，返回为列表
            :param sql: 数据库语句
            :return:
        """
        self.connect()
        row = None
        try:
            cur = self.connection.cursor()
            cur.execute(sql)
            row = cur.fetchall()  # all rows in table
            self.connection.commit()
            print(f"执行sql语句成功:{sql}")
            cur.close()
        except Exception as e:
            print(f"执行sql语句失败:{sql},异常信息:{e.args[0]}")
        finally:
            print(f"sql语句执行结果：{str(row)}")
            self.close()
            return row

    def execute_sql_dict(self, sql):
        """
        数据库查询，返回为字典
            :param sql: 数据库语句
            :return:
        """
        self.connect()
        row = None
        try:
            cur = self.connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
            cur.execute(sql)
            row = cur.fetchall()  # all rows in table
            self.connection.commit()
            print(f"执行sql语句成功:{sql}")
            cur.close()
        except Exception as e:
            print(f"执行sql语句失败:{sql},异常信息:{e.args[0]}")
        finally:
            print(f"sql语句执行结果：{str(row)}")
            self.close()
            return row

本文地址：https://blog.csdn.net/LittleGlassHeart/article/details/107359581