c_db.py

import psycopg2
from psycopg2 import OperationalError, InterfaceError


class UPostgresDB:
    """
    psycopg2.connect(
        dsn         #指定连接参数。可以使用参数形式或 DSN 形式指定。
        host        #指定连接数据库的主机名。
        dbname      #指定数据库名。
        user        #指定连接数据库使用的用户名。
        password    #指定连接数据库使用的密码。
        port        #指定连接数据库的端口号。
        connection_factory  #指定创建连接对象的工厂类。
        cursor_factory      #指定创建游标对象的工厂类。
        async_      #指定是否异步连接(默认False)。
        sslmode     #指定 SSL 模式。
        sslrootcert #指定证书文件名。
        sslkey      #指定私钥文件名。
        sslcert     #指定公钥文件名。
    )
    """

    def __init__(self, host, database, user, password, port=5432):
        self.host = host
        self.database = database
        self.user = user
        self.password = password
        self.port = port
        self.conn = None
        self.cur = None

    def connect(self):
        try:
            self.conn = psycopg2.connect(
                host=self.host,
                database=self.database,
                user=self.user,
                password=self.password,
                port=self.port
            )
            self.cur = self.conn.cursor()
        except Exception as e:
            print(f"连接数据库出现错误: {e}")

    def execute(self, query):
        try:
            if self.conn is None or self.conn.closed:
                self.connect()
            self.cur.execute(query)
            self.conn.commit()
        except InterfaceError as e:
            print(f"数据库连接已经关闭: {e}")
        except OperationalError as e:
            print(f"数据库连接出现问题: {e}")
            self.connect()
            self.retry_execute(query)
        except Exception as e:
            print(f"执行sql语句出现错误: {e}")
            self.conn.rollback()

    def retry_execute(self, query):
        try:
            self.cur.execute(query)
            self.conn.commit()
        except Exception as e:
            print(f"重新执行sql语句再次出现错误: {type(e).__name__}: {e}")
            self.conn.rollback()

    def execute_args(self, query, args):
        try:
            if self.conn is None or self.conn.closed:
                self.connect()
            self.cur.execute(query, args)
            self.conn.commit()
        except InterfaceError as e:
            print(f"数据库连接已经关闭: {e}")
        except OperationalError as e:
            print(f"数据库操作出现问题: {e}")
            self.connect()
            self.retry_execute_args(query, args)
        except Exception as e:
            print(f"执行sql语句出现错误: {e}")
            self.conn.rollback()

    def retry_execute_args(self, query, args):
        try:
            self.cur.execute(query, args)
            self.conn.commit()
        except Exception as e:
            print(f"重新执行sql语句再次出现错误: {type(e).__name__}: {e}")
            self.conn.rollback()

    def search(self, query, params=None):
        if self.conn is None or self.conn.closed:
            self.connect()
        self.cur.execute(query, params)

    def fetchall(self):
        return self.cur.fetchall()

    def fetchone(self):
        return self.cur.fetchone()

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

    def format(self, query):
        try:
            if self.conn is None or self.conn.closed:
                self.connect()
            self.cur.execute(query)
            self.conn.commit()
        except Exception as e:
            print(f"An error occurred: {e}")
            self.conn.rollback()