博客

import psycopg2


class PostgresDB:
    def __init__(self, dbname, user, password, host, port):
        self.dbname = dbname
        self.user = user
        self.password = password
        self.host = host
        self.port = port
        self.conn = psycopg2.connect(
            dbname=self.dbname,
            user=self.user,
            password=self.password,
            host=self.host,
            port=self.port
        )
        self.cur = self.conn.cursor()

    def execute_tb(self, sql):
        self.cur.execute(sql)
        self.conn.commit()

    def execute(self, sql):
        self.cur.execute(sql)
        self.conn.commit()
        return self.cur.fetchall()

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

    def insert(self, table, fields, values):
        sql = f"INSERT INTO {table} ({','.join(fields)}) VALUES ({','.join(['%s']*len(values))})"
        self.cur.execute(sql, values)
        self.conn.commit()

    def update(self, table, fields, values, condition):
        sql = f"UPDATE {table} SET {','.join([f'{field}=%s' for field in fields])} WHERE {condition}"
        self.cur.execute(sql, values)
        self.conn.commit()

    def delete(self, table, condition):
        sql = f"DELETE FROM {table} WHERE {condition}"
        self.cur.execute(sql)
        self.conn.commit()

    def select(self, table, fields, condition=None):
        if condition is None:
            sql = f"SELECT {','.join(fields)} FROM {table}"
            self.cur.execute(sql)
        else:
            sql = f"SELECT {','.join(fields)} FROM {table} WHERE {condition}"
            self.cur.execute(sql)
        return self.cur.fetchall()


# 使用示例：

# 连接数据库
db = PostgresDB(dbname='slife_db',user='postgres',password='Ab123456',host='localhost',port='5432')

# 插入数据、
table_name = "user1"
# 新建一个方法用于插入表格
db.execute_tb('''CREATE TABLE IF NOT EXISTS %s (
  id SERIAL PRIMARY KEY,
  uuid VARCHAR(255) NOT NULL,
  url VARCHAR(255) NOT NULL,
  device VARCHAR(255) NOT NULL,
  imgtime TIMESTAMP(255) NOT NULL,
  platenumber VARCHAR(255),
  result1 TEXT,
  pro FLOAT NOT NULL,
  img VARCHAR(255) NOT NULL,
  flag VARCHAR(2) NOT NULL,
  tag VARCHAR(2) NOT NULL) ''' % table_name)

# 插入数据
# db.insert(table='user1', fields=['uuid', 'url', 'device', 'imgtime', 'platenumber', 'result1', 'pro', 'img', 'flag', 'tag'],
#           values=('25a5ed28-0e53-11ee-9499-ddcdac16a28e', '/workspace/data/images/2023-06-19/0007/2023-06-19-11-17-45.jpeg',
#                   '0005', '2023-06-19 11:17:45', '沪DP9002', '''{'cls': ['2'], 'pro': [0.534], 'x': [1273], 'y': [597], 'w': [140], 'h': [116]}''',
#                   0.534, './data/images/2023-06-19/box_imgs/0007/2023-06-19-11-17-45.jpeg', '0', '0'))
#
# # 更新数据
db.update(table=table_name, fields=['tag'], values=(1,), condition="uuid='25a5ed28-0e53-11ee-9499-ddcdac16a28e'")

# 删除数据
# db.delete(table='users', condition="name='Alice'")

# 查询数据
result = db.select(table=table_name, fields=['pro', 'img'])
print(result)

# 关闭数据库连接
db.close()