k_db.py

import psycopg2


class PostgresDB:
    """
    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):
        self.conn = psycopg2.connect(
            host=self.host,
            database=self.database,
            user=self.user,
            password=self.password,
            port=self.port
        )
        self.cur = self.conn.cursor()

    def execute(self, query):
        try:
            self.cur.execute(query)
            self.conn.commit()
        except Exception as e:
            print(f"An error occurred: {e}")
            self.conn.rollback()

    def execute_args(self, query, args):
        try:
            self.cur.execute(query, args)
            self.conn.commit()
        except Exception as e:
            print(f"An error occurred: {e}")
            self.conn.rollback()

    def search(self, query, params=None):
        self.cur.execute(query, params)

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

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

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