python 数据库blob数据操

发布时间：2019-09-02 07:59:21编辑：auto阅读（2181）

#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

字串格式化

http://www.tsnc.edu.cn/default/tsnc_wgrj/doc/pythonhtml/html/native_data_types/formatting_strings.html

 

 

关键字：