python 数据库blob数据操
- 2020 年 1 月 8 日
- 筆記
#coding=gbk from sqlite3 import dbapi2 # ——————————- # jxDataSet 对象 # 用于保存 打开数据表的数据 # —————————— class jxDataSet: fields = [ ] # 字段名列表 data = [ ] # 数据表的记录 sql = [ ] # sql 语句 sERR = "" # 打开数据表时的错误信息 # 如果为空，表示打开成功 def __init__ (self ): pass def ok(self): return self.sERR==""
# 一个简单的显示打开结果的函数 disp def disp(self): if self.sERR!="": print("ERROR:",sERR); return print( "RecordCount=", len(self.data) ) print( "fields=",self.fields) for x in self.data: print( x )
# 数据库联接对象 # class jxconn: DB_Name = "" def __init__ (self, aDB_name ): self.DB_Name = aDB_name
# 打开 sql 指定的数据表 # 入口：sql # 返回： jxDataSet 对象 def opensql( self,sql ): jxData = jxDataSet() jxData.sql = sql try: conn=dbapi2.connect(self.DB_Name,30) except Exception , E: jxData.sERR = str(E) if jxData.sERR!="": return jxData try: cs = conn.cursor() cs.execute( sql ) jxData.fields = [ x[0] for x in cs.description ] jxData.data = cs.fetchall() cs.close() except Exception , E: jxData.sERR = str(E) conn.close() del( conn ) return jxData # 执行 sql 语句 # 入口： sqls # 一个字符串列表,每个元素是一条 sql 语句 # 或者一条 sql 语句 # 返回: 错误信息 # sqls 中所有的 sql语句中， # 要么同时执行，要么同时不执行， # 已执行部分，如果后面的sql语句出现错误，则一起回滚 # def runsql( self,sqls ): if not (type(sqls) in [ str, list]): return "parameter error" if type(sqls)==str: sqls = [ sqls ] n = len(sqls) if (n<1): return "" sERR = "" try: conn=dbapi2.connect(self.DB_Name, 30) #30是超时限制，单位秒,若不指定，是5秒 except Exception , E: return str(E) for i in range(0,n): try: conn.execute( sqls[i] ) except Exception , E: sERR = str(E) break if sERR=="": conn.commit() else: conn.rollback() conn.close() del(conn) return sERR # 保存二进制数据到 blob 字段 # 入口： sql 语句 # 如: "INSERT INTO mypic ( pic ) values( :0 )" # parameters # 一个数组，其每个元素是一个 bytes 类型的二进制数据 # parameters[0] 对应 sql 中的 :0 # parameters[1] 对应 sql 中的 :1 # 返回: 错误信息 def writeblob( self, sql, parameters ): mp = { } for i in range( 0, len(parameters) ): s = parameters[i] if type(s) != bytes : s = str(s).encode("gbk") try: mp[ str(i) ] = dbapi2.Binary( s ) except Exception , E: return str(E) try: conn=dbapi2.connect(self.DB_Name, 30) #30是超时限制，单位秒,若不指定，是5秒 except Exception , E: return str(E) sERR ="" try: conn.execute( sql, mp ) except Exception , E: sERR = str(E) if sERR=="": conn.commit() else: conn.rollback() conn.close() del(conn) return sERR
——————file2————————————————
#coding=gbk from jxconn import * import sys
# 将二进制数据保存到文件中 # 入口: picFileName, picData def savePic( picFileName, picData ): try: f = open( picFileName,"w") f.write( picData ) sERR = "None" f.close() except Exception , E: sERR = str(E) return sERR conn = jxconn("./si_db_isdb_t") cs = conn.opensql("SELECT * FROM logopng where Id=141 and Type=0") if cs.sERR!="": print( "read image from db failed:",cs.sERR) sys.exit(0) if len(cs.data)<1: print( "no data entry!") sys.exit(0) print("the quried entry number is ",len(cs.data)) for i in range (0,len(cs.data)): sERR = savePic( "./%d.png" % i, cs.data[i][4] ) # cs.data[0][1] 是第0条记录的第一个字段的值 if sERR!="None": print( "save image to file failed!",sERR)
print( "Done!");
－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－
python3.0 SQLite3 数据库读写blob字段 – jxconn(续)
http://hi.baidu.com/jxq61/blog/item/c8644f034804a00a1c95837d.html
字串操作
http://www.pythonclub.org/python-basic/string
字串格式化
