postgres篇---python连接postgres数据库2-Toy模板网

这篇具有很好参考价值的文章主要介绍了postgres篇---python连接postgres数据库2。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

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()