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.update(table=table_name, fields=['tag'], values=(1,), condition="uuid='25a5ed28-0e53-11ee-9499-ddcdac16a28e'")
result = db.select(table=table_name, fields=['pro', 'img'])
print(result)
db.close()