Python sqllit实例
1、操作数据库# -*- coding: UTF-8-*-
'''
Created on 2010-6-7
@author: qxf
'''
import sys
import os
from sqlite3 import connect, Connection
import UpdateLog
class VisitSqlite(object):
def __init__(self, dbname, tablename):
self.dbname = dbname
self.tablename = tablename
self.__myInit()
#表格初始化函数，若不存在，则新建表
def __myInit(self):
"""
如果数据库不存在，则创建数据库并新建数据库所需要的表
"""
if not os.path.exists(self.dbname):
connect(self.dbname)
UpdateLog.log("DateBase has been created!")
#下载任务表TaskList
# strSql='create table TaskList
# (gameID TEXT(64) primary key not null,\项目ID
# TaskState TEXT(64),\任务状态
# InstallMode TEXT(64),\安装方式
# updateQz integer,\是否强制升级
# DownFileNumber integer,\文件数量
# MD5 TEXT(255),\MD5 值
# InstallPacketPath TEXT(255),\安装原路径
# InstallPacketTempPath TEXT(255),\安装临时路径
# DownOppPath TEXT(255),\下载的相对路径
# SetupFile TEXT(255),\安装包可执行文件路径
# szReleaseTimeTEXT(32),\发布时间
# iErroTimes integer,\安装失败次数
# szVestion TEXT(32),\版本信息
# rateOfLoad integer,\下载进度
# rateOfSetup integer,\安装进度
# IsReboot integer,\重启标志
# zipSize integer,\文件大小
# FileSize integer,\解压后大小
# dimRedu integer,\冗余量
# Flag1 TEXT(32),\预留字段1
# Flag2 TEXT(32),\预留字段2
# Flag3 TEXT(32) )预留字段3
# '''
strSql='''create table TaskList
(id integer primary key autoincrement,
gameID TEXT(64),\
TaskState TEXT(64),\
InstallMode TEXT(64),\
updateQz integer,\
DownFileNumber integer,\
MD5 TEXT(255),\
InstallPacketPath TEXT(255),\
InstallPacketTempPath TEXT(255),\
DownOppPath TEXT(255),\
SetupFile TEXT(255),\
szReleaseTimeTEXT(32),\
iErroTimes integer,\
szVestion TEXT(32),\
rateOfLoad integer,\
rateOfSetup integer,\
IsReboot integer,\
zipSize integer,\
FileSize integer,\
dimRedu integer,\
Flag1 TEXT(32),\
Flag2 TEXT(32),\
Flag3 TEXT(32) )'''
self.execute(strSql)
#已安装项目信息表InstallGameInfo
# '''
# (gameID TEXT(64) primary key not null,\项目ID
# gameName TEXT(64),\项目名称
# dispName TEXT(64),\显示名称
# lastVersion TEXT(64),\最新版本
# serVersion TEXT(64))"已安装版本
# '''
strSql="create table InstallGameInfo \
(gameID TEXT(64) primary key not null,\
gameName TEXT(64),\
dispName TEXT(64),\
lastVersion TEXT(64),\
serVersion TEXT(64))"
self.execute(strSql)
#任务执行情况表WorkingInfo
strSql="create table WorkingInfo \
(ID integer primary key,\
NumLoading integer,\
NumSeting integer )"
self.execute(strSql)
#主版本号表
strSql="create table IdongInfo (MainVersion TEXT(32) primary key, path TEXT(256))"
self.execute(strSql)
#写入初始主版本号
strSql = "insert into IdongInfo(MainVersion, path) values('1.6.0', '')"
self.execute(strSql)
def execute(self, strSql):
"""
此函数用户执行不带返回值的SQL语句
strSql:执行的sql语句
"""
try:
dbc = Connection(self.dbname)
dbc.execute(strSql)
dbc.commit()
dbc.close()
return True
except Exception, e:
str = "execute '%s' exception with message '%s'" % (strSql, e)
print str
UpdateLog.log(str)
return False
def executeUpdate(self, strSql):
"""
查询 flag==0 返回记录数；1返回查询结果
fields表示要查询的字段集合 ,不为空则返回字段查询结果
wsm 为条件查询语句 ，不为空，则返回条件查询结果（结果为字典列表）
"""
try:
dbc = Connection(self.dbname)
cur = dbc.cursor()
cur.execute(strSql)
rowall = cur.fetchall()
dbc.commit()
dbc.close()
#将结果转化成utf-8编码
res = []
for i in range(len(rowall)):
tmp = []
for j in range(len(rowall)):
if type(rowall) == type(u"123"):
tmp.append(rowall.encode("utf-8"))
else:
tmp.append(rowall)
res.append(tmp)
return res
# if flag==0:
# return len(rowall)
# else:
# #根据字段列表和返回值生成 返回字典列表
# if strfield=='':
# fields = self.getTableinfo(self.tablename)
# else:
# fields = strfield.rsplit(',')
#
## print 'The Fields of ',tablename,'is:',fields
# rowlist=[]
# for row in rowall:
# dict = {}
## print len(row),len(fields)
# for i in range(0,len(row)):
## print fields,row
# dict] = row
# rowlist.append(dict)
#
## print rowlist
# return rowlist
#
except:
str = "executeUpdate '%s' exception with message '%s'" % (strSql, sys.exc_info())
print str
UpdateLog.log(str)
return None
def getTableinfo(self):
"""
获取表的字段信息
"""
try:
dbc = Connection(self.dbname)
cur = dbc.cursor()
strsql = "select * from sqlite_master where type='table' and name='" + self.tablename+"'"
# print strsql
cur.execute(strsql)
dbinfo = cur.fetchall()
#print dbinfo
dbc.commit()
dbc.close()
except Exception, e:
str = "getTableinfo err with message %s" % e
print str
UpdateLog.log(str)
return None
res = self.__createsqlToFieldlist(dbinfo)
#转化成utf-8
for i in range(len(res)):
res = res.encode("utf-8")
return res
def __createsqlToFieldlist(self,createsql):
"""
根据create sql 语句返回字段列表
"""
strtemp = createsql
list =[]
ifind=strtemp.find('(')
while ifind!=-1:
strtemp = strtemp.lstrip()
list.append(strtemp[:strtemp.find(' ')])
ifind = strtemp.find(',')
return list
# #删除某条记录
# def deletedb(self,tablename,wsm=''):
# log = open('log.txt','w')
# try:
# dbc = Connection(self.dbname)
# if wsm=='':
# dbc.execute("delete from "+tablename)
# else:
# dbc.execute("delete from "+tablename+' where '+wsm)
# dbc.commit()
# dbc.close()
# except:
# print sys.exc_info()
# log.write(sys.exc_info())
# log.close()
# #查询数据库表结构信息
# #输入表名
# #返回表字段列表
#
# def dropTable(self,tablename):
# dbc = Connection(self.dbname)
# dbc.execute('drop table '+tablename)
# dbc.commit()
# dbc.close()
# #根据字段类型和值返回字符串
# def fromtypetovalue(self,tablename,field,value):
# try:
# dbc = Connection(self.dbname)
# dbc.execute(strSql)
# dbc.commit()
# dbc.close()
# except:
# print sys.exc_info()
if __name__ == '__main__':
UpdateLog.init_log("UpdateLog.log")
vs = VisitSqlite('LanucherInfo.db', "WorkingInfo")
print vs.getTableinfo()
2、简易sql控制台
# -*- coding: UTF-8-*-
import VisitSqlite
##getattr(TaskListDBFace)
class Console:
@staticmethod
def start():
while True:
tip1 = "Please enter your sql to perform...\n"
tip2 = "ERROR:your enter sql can not empry...\n"
tip3 = "ERROR:your operate table is not exist...\n"
executeErrorTip = 'Execute sql error,please check your excute sql...'
executeSuccTip = 'Excute sql success...'
sql = raw_input(tip1)
if sql == '':
print tip2
else:
if 'TaskList' in sql:
dbc = VisitSqlite.VisitSqlite('LanucherInfo.db', 'TaskList')
elif 'InstallGameInfo' in sql:
dbc = VisitSqlite.VisitSqlite('LanucherInfo.db', 'InstallGameInfo')
elif 'WorkingInfo' in sql:
dbc = VisitSqlite.VisitSqlite('LanucherInfo.db', 'WorkingInfo')
else:
print tip3
continue
tableInfo = dbc.getTableinfo()
if ('select' in sql) or ('SELECT' in sql):
res = dbc.executeUpdate(sql)
if res is None:
print executeErrorTip
resDict = []
for row in res:
dict = {}
for i in range(0,len(row)):
dict] = row
resDict.append(dict)
if len(resDict) != 0:
print 'The query result is:\n'
print FormatUtil.format(resDict)
else:
print 'The query table is empty!'
else:
ret = dbc.execute(sql)
if ret:
print executeSuccTip
else:
print executeErrorTip
class FormatUtil():
'''
format the select result
'''
@staticmethod
def format(res):
'''
the res is a list that contains dict
'''
resStr = ''
topLine = ''
nameStr = ''
valuesStr = ''
for record in res:
keys = record.keys()
values = record.values()
strLine = ''
valueStr = ''
for key in keys:
strLine += FormatUtil.printLine(20)
nameStr = FormatUtil.printValue(keys)
valueStr = FormatUtil.printValue(values)+'\n'+strLine
valuesStr += valueStr+'\n'
resStr = strLine+'\n'+nameStr+'\n'+strLine+'\n'+valuesStr
return resStr
@staticmethod
def printValue(valueList):
'''
'''
resStr = ''
spac = ''
for i in range(0,len(valueList)):
intSpac = 0
if valueList is not None:
intSpac = len(str(valueList))
else:
valueList = ''
for j in range(0,20-intSpac):
spac += ' '
resStr += str(valueList)+spac+'|'
spac = ''
return resStr
@staticmethod
def printLine(length):
resStr = ''
tems = ''
for i in range(0,length):
tems += '-'
resStr += tems+'|'
return resStr
if __name__ == '__main__':
# print FormatUtil.printCell('dfdf sdfs d')
# print FormatUtil.format([{'name':'Dream','age':'12','job':'coder','school':'xinhua university'},{'name':'Dream2','age':'23','job':'coder2','school':'xinhua university2'}])
# print FormatUtil.printLine('name')
# print FormatUtil.printValue(['name','age','job'])
Console.start()
页:
[1]