python 自带sqlite3数据库访问模块. sqlite3 以下写一个数据库访问类
'''
2016年2月5日
描述: 操作sqlite数据库的封装
主要功能: 将sqlite数据库数据转为python对象集合
错误日志: 如果在操作过程中发生了错误,将会在当前目录下产生dblog目录,并记载. 步骤: 执行一个完整的数据操作需要:建立连接->得到游标->执行sql->分析结果
第二个步骤和ado.net不一样,游标类的作用有点类似于SqlDataReader,有方法可从中读取结果集
结果: 执行SELECT会得到一个含有Model对象的列表,对象的属性名是SQL语句查询的字段名,值是字段值
Model是一个空对象,它的属性都是动态加上去的,在得到结果集之后.
''' import os
import sqlite3
from Loggers.MyLog import MyLog
from DBA.Model import Model class SQLiteDBA:
'''操作sqlite的类'''
#
def __init__(self,connstr=None):
'''// 初始化 主要是传入连接字符串
// 1. connstr:数据库连接串.就是sqlite文件所在的路径
'''
# 连接字符串
self.__ConnStr=connstr
# 当前连接
self.__Conn=None
# 游标
self.__Cur=None #
def ExecuteQuery(self,sql,*paras,EntityClass=None,paraClass=None):
'''// 执行一个查询,返回结果集.如果没有结果或者发生了异常返回None
// 1. sql:sql语句
// 2. *paras:参数元组,列表
// 3. EntityClass:传入实体类,直接写上类名就行.需要写成命名方式EntityClass=XXX
// 4. paraClass:实体类参数 需要写成命名方式paraClass=XXX
'''
self.__OpenDB()
try:
# 执行语句
if paraClass is not None:
self.__Cur.execute(sql,paraClass.__dict__)
else:
self.__Cur.execute(sql,paras)
# 包含查出的所有记录数
data=self.__Cur.fetchall()
# 如果结果集为空
if(len(data)==0):return None ## 将结果集和列名绑定到model上. # 包含col列信息,这返回的是二维元组,每个元组的元素都是一个有七个值的元组.其中第1个值是列名.顺序和SQL语句查询列名顺序一致
colinfo=self.__Cur.description
# 建立空列表放实体对象,datalist=[]也可以
datalist=list()
if(EntityClass is None):EntityClass=Model
# 循环赋值
for row in data:
m=EntityClass()
for colindex in range(len(row)):
m.__dict__[colinfo[colindex][0]]=row[colindex]
# add m
datalist.append(m)
# 返回结果对象列表
return datalist
except sqlite3.Error as e:
MyLog.AddLog('在[ {0} ]数据库执行查询时发生异常.[{1}]'.format(self.__ConnStr,e.args[0]))
return None
finally:
self.__CloseDB() #
def ExecuteScalar(self,sql,*paras,paraClass=None):
'''
// 执行一个查询返回单个值.DBNULL或者异常都返回None
// 1. sql:sql语句
// 2. *paras:参数元组,列表
// 3. paraClass:实体类参数 需要写成命名方式paraClass=XXX
'''
self.__OpenDB()
try:
# 执行语句
if paraClass is not None:
self.__Cur.execute(sql,paraClass.__dict__)
else:
self.__Cur.execute(sql,paras)
# 包含查出的所有记录数
data=self.__Cur.fetchall()
# 如果结果集为空
if(len(data)==0):return None
# 返回单个值 数据库字符值为NULL时,或者值为空字符(要转成字符串后再判断为空)时的情况
if(data[0][0] is None):return None
if(len(str(data[0][0]))==0):return None
return data[0][0]
except sqlite3.Error as e:
MyLog.AddLog('在[ {0} ]数据库执行单值查询时发生异常.[{1}]'.format(self.__ConnStr,e.args[0]))
return None
finally:
self.__CloseDB()
#
def ExecuteNoQuery(self,sql,*paras,paraClass=None):
'''// 执行一个非查询,返回受影响的行数.如果发生异常,返回None
// 1.sql:sql语句
// 2. *paras:参数元组,列表
// 3. paraClass:实体类参数 需要写成命名方式paraClass=XXX
'''
self.__OpenDB()
try:
# 执行语句
if paraClass is not None:
self.__Cur.execute(sql,paraClass.__dict__)
else:
self.__Cur.execute(sql,paras)
# 非查询需要提交事务
cou=self.__Cur.rowcount
self.__Conn.commit()
#return self.__Cur.rowcount
return cou
except sqlite3.Error as e:
MyLog.AddLog('在[ {0} ]数据库执行非查询时发生异常.[{1}]'.format(self.__ConnStr,e.args[0]))
return None
finally:
self.__CloseDB()
#
def __OpenDB(self):
'''打开数据库连接'''
if not os.path.exists(self.__ConnStr):
MyLog.AddLog('在指定的路径[ {0} ]上没有找到数据库文件.'.format(self.__ConnStr))
try:
self.__Conn = sqlite3.connect(self.__ConnStr)
except sqlite3.Error as e:
MyLog.AddLog('打开[ {0} ]数据库失败.[{1}]'.format(self.__ConnStr,e.args[0]))
# 设置游标对象
self.__Cur=self.__Conn.cursor() #
def __CloseDB(self):
'''关闭数据库连接和游标'''
self.__Cur.close()
self.__Conn.close()
model实体对象
class Model():
"""
2016年2月5日
实体对象.该对象没有属性,在查询之后会增加属性和值
属性名就是sql查出的字段名.值为对应的值.
对于各种数据库,这个实体类是通用的.
"""
使用示例:
from Loggers.MyLog import MyLog
from DBA.SQLite import SQLiteDBA
# 数据库对象
db=SQLiteDBA('./database/sanguo.db')
# 实体类
class Wujiang:
def __init__(self):
self.Id=None
self.Name=None
self.WuLi=None
self.ZhiLi=None
self.Sudu=None
self.WuPin=None
self.ZhiWu=None
self.RowNumber=None # 增
caoc=Wujiang()
caoc.Name='曹操'
caoc.Sudu=165
caoc.WuLi=210
caoc.ZhiLi=235
caoc.ZhiWu='丞相'
caoc.WuPin='倚天剑'
isexist="select Id from WuJiang where Name=@name";# 查询是否已有
data=db.ExecuteScalar(isexist,caoc.Name)
if data is not None:
print("已经有了[ ",caoc.Name,' ]此次未做任何增加或修改')
else:
print('还没有添加[ ',caoc.Name,' ],已经执行增加')
sql="insert into WuJiang(Name,WuLi,ZhiLi,Sudu,WuPin,ZhiWu) values(@Name,@WuLi,@ZhiLi,@Sudu,@WuPin,@ZhiWu)"
re=db.ExecuteNoQuery(sql,paraClass=caoc)
print(re==1) # 查
print('已经执行查询',caoc.Name)
query=Wujiang()
query.Name='曹操'
datalist= db.ExecuteQuery("select * from wujiang where Name=@Name",EntityClass=Wujiang,paraClass=query)
if datalist is None:
print("没有任何数据查出来")
else:
for obj in datalist:
print(obj.__dict__)
# 改
caoc.ZhiWu='大将军'
print('已经执行修改',caoc.ZhiWu,caoc.Name)
re=db.ExecuteNoQuery("update wujiang set ZhiWu=@ZhiWu where Name=@Name",paraClass=caoc)
print(re==1)