python的sqlite3连接模块用法.docx

python在2.5版以后，就自带了sqlite3模块，直接import即可。模块可用后，就可以实现连接sqlite数据库的功能。sqlite3数据库与mysql不同，它每一个库叫做一个文件，连接的时候，库名即文件名，库内的表则操作方法与mysql相同，或许支持的sql语句会有一些差异。我根据连接mysql的操作方法的包装，改写成了这个sqlite数据库连接模块。附件有这个源码。Python代码1. #coding=gbk2. importos,sys,string3. importsqlite34. importtraceback5. 6. classSQLObj(object):7. def_init_(self,db=youmoney):8. self.db_name=db9. self.connected=010. self.cur=None#指针对象11. self.conn=None#连接对象12. self._connect()13. 14. def_connect(self):15. try:16. #printconnectto,self.db_name17. self.conn=sqlite3.connect(self.db_name)18. self.cur=self.conn.cursor()19. self.connected=120. except:21. traceback.print_exc()22. self.connected=023. 24. property25. defis_connected(self):26. returnself.connected!=027. 28. def_check_alive(self):29. ifnotself.is_connected:30. self._connect()31. ifnotself.is_connected:32. raiseCantconnecttosqlite333. 34. #直接提取方法35. defquery(self,sql,warning=1):36. self._check_alive()37. try:38. cur=self.conn.cursor()39. cur.execute(sql)40. res=cur.fetchall()41. cur.close()42. except:43. ifwarning:44. traceback.print_exc()45. returnNone46. returnres47. 48. 49. #特殊的提取方法，形成字典返回50. defdquery(self,sql,warning=1):51. self._check_alive()52. try:53. cur=self.conn.cursor()54. cur.execute(sql)55. 56. des=cur.description#获取字段描述57. res=cur.fetchall()#获取字段内容58. ret=59. ifdes:60. names=x0forxindes#从des中提取每个元素（列表）的第一个元素（字符串）61. forlineinres:62. ret.append(dict(zip(names,line)#dict是形成字典的工厂函数，zip用于多个列表循环(name,line是两个列表)63. else:64. ret=res65. cur.close()66. except:67. ifwarning:68. traceback.print_exc()69. returnNone70. returnret71. 72. defexecute(self,sql,warning=1):73. self._check_alive()74. try:75. cur=self.conn.cursor()76. rows=cur.execute(sql)77. mit()78. cur.close()79. returnrows80. except:81. ifwarning:82. traceback.print_exc()83. return-184. 85. defclose(self):86. ifself.connected=0:87. return88. try:89. self.cur.close()90. self.conn.close()91. self.connected=092. except:93. pass94. 95. def_del_(self):96. self.close()97. 98. def_lbsql_test_sqlite3():99. db=SQLObj(db=youmoney.db)100. lists=db.dquery(select*fromcapital)101. forlinlists:102. fork,vinl.items():103. print%s:%s%(k,v),104. break105. break106. 107. if_name_=_main_:108. _lbsql_test_sqlite3() lbsqlite.zip(1 KB) 下载次数: 2本文转载链接：/article/54234.htmPython3实现连接SQLite数据库的方法这篇文章主要介绍了Python3实现连接SQLite数据库的方法,在Python数据库编程中有着广泛的应用,需要的朋友可以参考下本文实例讲述了Python3实现连接SQLite数据库的方法，对于Python的学习有不错的参考借鉴价值。分享给大家供大家参考之用。具体方法如下：实例代码如下：?1234567891011121314151617181920212223242526272829303132333435363738394041import sqlite3db= rD:pyWorktest.db #pyWork目录下test.db数据库文件drp_tb_sql= drop table if exists staffcrt_tb_sql= create table if not exists staff(id integer primary key autoincrement unique not null,name varchar(100),city varchar(100);#连接数据库con= sqlite3.connect(db)cur= con.cursor()#创建表staffcur.execute(drp_tb_sql)cur.execute(crt_tb_sql)#插入记录insert_sql= insert into staff (name,city) values (?,?) #?为占位符cur.execute(insert_sql,(Tom,New York)cur.execute(insert_sql,(Frank,Los Angeles)cur.execute(insert_sql,(Kate,Chicago)cur.execute(insert_sql,(Thomas,Houston)cur.execute(insert_sql,(Sam,Philadelphia)mit()#查询记录select_sql= select * from staffcur.execute(select_sql)#返回一个list，list中的对象类型为tuple（元组）date_set= cur.fetchall()for rowin date_set:print(row)cur.close()con.close()希望本文实例对大家的Python学习有所帮助。本文转载连接：/article/50902.htmPython SQLite3数据库操作类分享这篇文章主要介绍了Python SQLite3数据库操作类分享,需要的朋友可以参考下接触Python时间也不是很长的，最近有个项目需要分析数据，于是选用Python为编程语言，除了语言特性外主要还是看重Python对于SQLite3数据库良好的支持能力了，因为需要灵活处理大量的中间数据。刚开始一些模块我还乐此不疲的写SQL语句，后来渐渐厌倦了，回想到以前捣鼓C#的时候利用反射初步构建了个SQL查询构造器，直到发现linq，于是放弃了这个计划，当然微软后来又推出了Entity Framework，这些都是后话了，而且现在我对微软的东西兴趣不是很大的，好了，扯多了，下面继续正文。对了，再扯一句，优秀的博客程序Drupal也使用了类似的查询构造器进行数据库查询，避免直接写SQL语句，另外这样做的一点点好处就是，可以一定程度的屏蔽平台相关性，对于数据库迁移还是有帮助的。不过我今天介绍的数据库辅助类查询构造器是个很简单的东东，甚至只限于SQLite数据库，如果有童鞋感兴趣可以完善下，我目前只要操作SQLite顺手就可以了，对于比较大的数据库应用就直接上ORM吧。先看代码：复制代码代码如下:import sqlite3# *# *# * Description: Python操作SQLite3数据库辅助类(查询构造器)# * Author: wangye# *# *def _wrap_value(value): return repr(value)def _wrap_values(values): return list(map(_wrap_value, values)def _wrap_fields(fields): for key,value in fields.items(): fieldskey = _wrap_value(value) return fieldsdef _concat_keys(keys): return + ,.join(keys) + def _concat_values(values): return ,.join(values)def _concat_fields(fields, operator = (None, ,): if operator: unit_operator, group_operator = operator # fields = _wrap_fields(fields) compiled = for key,value in fields.items(): compiled.append( + key + ) if unit_operator: compiled.append(unit_operator) compiled.append(value) compiled.append(group_operator) compiled.pop() # pop last group_operator return .join(compiled)class DataCondition(object): 本类用于操作SQL构造器辅助类的条件语句部分 例如: DataCondition(=, AND), id = 26) DataCondition(=, AND), True, id = 26) def _init_(self, operator = (=, AND), ingroup = True, *kwargs): 构造方法 参数: operator 操作符，分为(表达式操作符, 条件运算符) ingroup 是否分组，如果分组，将以括号包含 kwargs 键值元组，包含数据库表的列名以及值 注意这里的等于号不等于实际生成SQL语句符号 实际符号是由operator0控制的 例如: DataCondition(=, AND), id = 26) (id=26) DataCondition(, OR), id = 26, age = 35) (id26 OR age35) DataCondition(LIKE, OR), False, name = John, company = Google) name LIKE John OR company LIKE 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) if self.ingroup: return ( + result + ) return result def _str_(self): return self._unicode_() def toString(self): return self._unicode_()class DataHelper(object): SQLite3 数据查询辅助类 def _init_(self, filename): 构造方法 参数: filename 为SQLite3 数据库文件名 self.file_name = filename def open(self): 打开数据库并设置游标 self.connection = sqlite3.connect(self.file_name) self.cursor = self.connection.cursor() return self def close(self): 关闭数据库，注意若不显式调用此方法， 在类被回收时也会尝试调用 if hasattr(self, connection) and self.connection: self.connection.close() def _del_(self): 析构方法，做一些清理工作 self.close() def commit(self): 提交事务 SELECT语句不需要此操作，默认的execute方法的 commit_at_once设为True会隐式调用此方法， 否则就需要显示调用本方法。 mit() def execute(self, sql = None, commit_at_once = True): 执行SQL语句 参数: sql 要执行的SQL语句，若为None，则调用构造器生成的SQL语句。 commit_at_once 是否立即提交事务，如果不立即提交， 对于非查询操作，则需要调用commit显式提交。 if not sql: sql = self.sql self.cursor.execute(sql) if commit_at_once: mit() def fetchone(self, sql = None): 取一条记录 self.execute(sql, False) return self.cursor.fetchone() def fetchall(self, sql = None): 取所有记录 self.execute(sql, False) return self.cursor.fetchall() def _concat_keys(self, keys): return _concat_keys(keys) def _concat_values(self, values): return _concat_values(values) def table(self, *args): 设置查询的表，多个表名用逗号分隔 self.tables = args self.tables_snippet = self._concat_keys(self.tables) return self 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 if hasattr(self, condition_snippet): self.where_snippet = WHERE + self.condition_snippet def _select(self): template = SELECT %(keys)s FROM %(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 = INSERT INTO %(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)s SET %(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 = DELETE FROM %(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): return self.sql def _str_(self): return self._unicode_() def select(self, *args): self.current_token = SELECT self.body_keys = args self._build() return self def insert(self, *kwargs): self.current_token = INSERT self.body_fields = self._wrap_fields(kwargs) self._build() return self def update(self, *kwargs): self.current_token = UPDATE self.body_fields = self._wrap_fields(kwargs) self._build() return self def delete(self, *conditions): self.current_token = DELETE self._build() #if *conditions: self.where(*conditions) return self def where(self, *conditions): conditions = list(map(str, conditions) self.condition_sn