Python SQLite3数据库操作类分享
接触Python时间也不是很长的,最近有个项目需要分析数据,于是选用Python为编程语言,除了语言特性外主要还是看重Python对于SQLite3数据库良好的支持能力了,因为需要灵活处理大量的中间数据。
刚开始一些模块我还乐此不疲的写SQL语句,后来渐渐厌倦了,回想到以前捣鼓C#的时候利用反射初步构建了个SQL查询构造器,直到发现linq,于是放弃了这个计划,当然微软后来又推出了EntityFramework,这些都是后话了,而且现在我对微软的东西兴趣不是很大的,好了,扯多了,下面继续正文。
对了,再扯一句,优秀的博客程序Drupal也使用了类似的查询构造器进行数据库查询,避免直接写SQL语句,另外这样做的一点点好处就是,可以一定程度的屏蔽平台相关性,对于数据库迁移还是有帮助的。
不过我今天介绍的数据库辅助类查询构造器是个很简单的东东,甚至只限于SQLite数据库,如果有童鞋感兴趣可以完善下,我目前只要操作SQLite顺手就可以了,对于比较大的数据库应用就直接上ORM吧。
先看代码:
importsqlite3 #*************************************************** #* #*Description:Python操作SQLite3数据库辅助类(查询构造器) #*Author:wangye #* #*************************************************** def_wrap_value(value): returnrepr(value) def_wrap_values(values): returnlist(map(_wrap_value,values)) def_wrap_fields(fields): forkey,valueinfields.items(): fields[key]=_wrap_value(value) returnfields def_concat_keys(keys): return"["+"],[".join(keys)+"]" def_concat_values(values): return",".join(values) def_concat_fields(fields,operator=(None,",")): ifoperator: unit_operator,group_operator=operator #fields=_wrap_fields(fields) compiled=[] forkey,valueinfields.items(): compiled.append("["+key+"]") ifunit_operator: compiled.append(unit_operator) compiled.append(value) compiled.append(group_operator) compiled.pop()#poplastgroup_operator return"".join(compiled) classDataCondition(object): """ 本类用于操作SQL构造器辅助类的条件语句部分 例如: DataCondition(("=","AND"),id=26) DataCondition(("=","AND"),True,id=26) """ def__init__(self,operator=("=","AND"),ingroup=True,**kwargs): """ 构造方法 参数: operator操作符,分为(表达式操作符,条件运算符) ingroup 是否分组,如果分组,将以括号包含 kwargs 键值元组,包含数据库表的列名以及值 注意这里的等于号不等于实际生成SQL语句符号 实际符号是由operator[0]控制的 例如: DataCondition(("=","AND"),id=26) (id=26) DataCondition((">","OR"),id=26,age=35) (id>26ORage>35) DataCondition(("LIKE","OR"),False,name="John",company="Google") nameLIKE'John'ORcompanyLIKE"Google" """ self.ingroup=ingroup self.fields=kwargs self.operator=operator def__unicode__(self): self.fields=_wrap_fields(self.fields) result=_concat_fields(self.fields,self.operator) ifself.ingroup: return"("+result+")" returnresult def__str__(self): returnself.__unicode__() deftoString(self): returnself.__unicode__() classDataHelper(object): """ SQLite3数据查询辅助类 """ def__init__(self,filename): """ 构造方法 参数:filename为SQLite3数据库文件名 """ self.file_name=filename defopen(self): """ 打开数据库并设置游标 """ self.connection=sqlite3.connect(self.file_name) self.cursor=self.connection.cursor() returnself defclose(self): """ 关闭数据库,注意若不显式调用此方法, 在类被回收时也会尝试调用 """ ifhasattr(self,"connection")andself.connection: self.connection.close() def__del__(self): """ 析构方法,做一些清理工作 """ self.close() defcommit(self): """ 提交事务 SELECT语句不需要此操作,默认的execute方法的 commit_at_once设为True会隐式调用此方法, 否则就需要显示调用本方法。 """ self.connection.commit() defexecute(self,sql=None,commit_at_once=True): """ 执行SQL语句 参数: sql 要执行的SQL语句,若为None,则调用构造器生成的SQL语句。 commit_at_once是否立即提交事务,如果不立即提交, 对于非查询操作,则需要调用commit显式提交。 """ ifnotsql: sql=self.sql self.cursor.execute(sql) ifcommit_at_once: self.commit() deffetchone(self,sql=None): """ 取一条记录 """ self.execute(sql,False) returnself.cursor.fetchone() deffetchall(self,sql=None): """ 取所有记录 """ self.execute(sql,False) returnself.cursor.fetchall() def__concat_keys(self,keys): return_concat_keys(keys) def__concat_values(self,values): return_concat_values(values) deftable(self,*args): """ 设置查询的表,多个表名用逗号分隔 """ self.tables=args self.tables_snippet=self.__concat_keys(self.tables) returnself def__wrap_value(self,value): return_wrap_value(value) def__wrap_values(self,values): return_wrap_values(values) def__wrap_fields(self,fields): return_wrap_fields(fields) def__where(self): #self.condition_snippet ifhasattr(self,"condition_snippet"): self.where_snippet="WHERE"+self.condition_snippet def__select(self): template="SELECT%(keys)sFROM%(tables)s" body_snippet_fields={ "tables":self.tables_snippet, "keys":self.__concat_keys(self.body_keys), } self.sql=template%body_snippet_fields def__insert(self): template="INSERTINTO%(tables)s(%(keys)s)VALUES(%(values)s)" body_snippet_fields={ "tables":self.tables_snippet, "keys":self.__concat_keys(list(self.body_fields.keys())), "values":self.__concat_values(list(self.body_fields.values())) } self.sql=template%body_snippet_fields def__update(self): template="UPDATE%(tables)sSET%(fields)s" body_snippet_fields={ "tables":self.tables_snippet, "fields":_concat_fields(self.body_fields,("=",",")) } self.sql=template%body_snippet_fields def__delete(self): template="DELETEFROM%(tables)s" body_snippet_fields={ "tables":self.tables_snippet } self.sql=template%body_snippet_fields def__build(self): { "SELECT":self.__select, "INSERT":self.__insert, "UPDATE":self.__update, "DELETE":self.__delete }[self.current_token]() def__unicode__(self): returnself.sql def__str__(self): returnself.__unicode__() defselect(self,*args): self.current_token="SELECT" self.body_keys=args self.__build() returnself definsert(self,**kwargs): self.current_token="INSERT" self.body_fields=self.__wrap_fields(kwargs) self.__build() returnself defupdate(self,**kwargs): self.current_token="UPDATE" self.body_fields=self.__wrap_fields(kwargs) self.__build() returnself defdelete(self,*conditions): self.current_token="DELETE" self.__build() #if*conditions: self.where(*conditions) returnself defwhere(self,*conditions): conditions=list(map(str,conditions)) self.condition_snippet="AND".join(conditions) self.__where() ifhasattr(self,"where_snippet"): self.sql+=self.where_snippet returnself
下面举几个例子供大家参考吧:
db=DataHelper("/home/wangye/sample.db3") db.open()#打开数据库 db.execute(""" CREATETABLE[staffs]( [staff_id]INTEGERPRIMARYKEYAUTOINCREMENT, [staff_name]TEXTNOTNULL, [staff_cardnum]TEXTNOTNULL, [staff_reserved]INTEGERNOTNULL ) """)#直接执行SQL语句,注意这里commit_at_once默认为True db.table("staffs").insert(staff_name="John",staff_cardnum="1001",staff_reserved=0) #插入一条记录 rs=db.table("staffs").select("staff_id","staff_name").fetchall() #直接取出所有staff_id和staff_name rs=db.table("staffs").select("staff_name").where(DataCondition(("=","AND"),id=1)).fetchone() #取一条staff_id为1的staff_name rs=db.table("staffs").select("staff_name").where(DataCondition(("<","AND"),id=100),DataCondition(("=","AND"),staff_reserved=1)).fetchone() #取一条id小于100并且staff_reserved为1的staff_name记录 db.close()#关闭数据库
目前还没有让其支持星号(*)操作符,另外在多表同名列操作方面处理得也不是很好,这个只用于日常简单的脚本操作,最好不要用于生产环境,因为可能有未知问题。