Skip to content

Instantly share code, notes, and snippets.

@ywzhang909
Created August 26, 2021 05:51
Show Gist options
  • Save ywzhang909/e97bbcd903a12f947c8b28f208ca0f7d to your computer and use it in GitHub Desktop.
Save ywzhang909/e97bbcd903a12f947c8b28f208ca0f7d to your computer and use it in GitHub Desktop.
postgre sql helper #database #postgresql
import pg
class Greenplummanager():
def __init__(self,host,port,dbname,user,pwd):
self._host = host
self._port = port
self._dbname = dbname
self._user = user
self._pwd = pwd
def connectGreenPlum(self):
isconnect = False
try:
# pg.connect()
self.pgdb_conn = pg.connect(dbname = self._dbname, host = self._host, port = self._port, user = self._user, passwd = self._pwd)
# pgdb_conn = pg.connect("host=192.168.103.31 port=5432 dbname=tpc user=gpadmin")
isconnect = True
except Exception as e:
print(e.args[0])
return isconnect
def closeConnection(self):
self.pgdb_conn.close()
def queryDataBySQL(self,querysql):
result = None
try:
result = self.pgdb_conn.query(querysql)
except Exception as e:
print('query record failed')
print('query record failed, %s, %s' %(e.args[0],e.message))
result = None
return result
#insert data INSERT INTO tbl_product3(sv_productname) values('apple')
def insertDataBySQL(self,insertsql):
print(insertsql)
isresult = True
try:
self.pgdb_conn.query(insertsql)
except Exception as e:
print('insert record into table failed')
print('insert record failed, %s, %s' %(e.args[0],e.message))
isresult = False
return isresult
def deleteDataBySQL(self,deletesql):
isresult = True
try:
self.pgdb_conn.query(deletesql)
except Exception as e:
print('delete record failed')
print('delete record failed, %s, %s' %(e.args[0],e.message))
isresult = False
return isresult
def insertBlobData(self,insertsql,imgdata):
isresult = True
try:
print(type(imgdata))
self.pgdb_conn.query(insertsql % pg.escape_bytea(imgdata))
except Exception as e:
print('insert blob failed')
print('insert blob failed, %s, %s' %(e.args[0],e.message))
isresult = False
return isresult
def insertImg(self, path):
img_file = open(path)
print(type(img_file.read()))
self.pgdb_conn.query("insert into gtab_img values('1','%s')" % pg.escape_bytea(img_file.read()))
img_file.close()
def readImg(self, id):
result = self.pgdb_conn.query("select bindata from gtab_img where id = '%s'" % id)
img = result.dictresult()[0]['bindata']
file = open('text.jpg','wb')
file.write(img)
file.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment