sqlite数据库在Python中的使用简介
1. sqlite3介绍
import sqlite3
SQLite数据库是一款非常小巧的嵌入式开源数据库软件，没有独立的维护进程，所有的维护都来自于程序本身。在python中，使用sqlite3创建数据库的连接，当数据库文件不存在时，会新建数据库文件；如果数据库文件已经存在了，则打开数据库文件。
连接对象可以是硬盘上面的数据库文件，也可以是建立在内存中的，在内存中的数据库执行完任何操作后，不需要提交事务的commit。
#使用:memory:标识打开的是内存数据库
con = sqlite3.connect(":memory:")
#在本地当前目录下创建数据库文件\打开数据库文件-test.db
conn = sqlite3.connect('test.db')
打开数据库时返回的对象conn就是一个数据库连接对象，它可以有以下操作：
commit() --事务提交
rollback() --事务回滚
close() --关闭一个数据库链接
cursor() --创建一个游标
创建一个游标对象：cu
cu = conn.cursor()
关闭一个游标对象
cu.close()
在sqlite3中，所有sql语句的执行都要在游标对象的参与下完成
对于游标对象cu，具有以下具体操作：
execute() --执行一条sql语句
executemany() --执行多条sql语句
close() --游标关闭
fetchone() --从结果中取出一条记录
fetchmany() --从结果中取出多条记录
fetchall() --从结果中取出所有记录
scroll() --游标滚动
2. 创建table
创建表（id,name,gender,age,address,phone)，通过cu.execute(create_table_sql)执行sql语句创建表，如果失败会抛出异常。“table student already exists”，代表已经存在student表，不能再次创建。
create_table_sql = '''CREATE TABLE `student` (
`id` int(11) NOT NULL,
`name` varchar(20) NOT NULL,
`gender` varchar(4) DEFAULT NULL,
`age` int(11) DEFAULT NULL,
`address` varchar(200) DEFAULT NULL,
`phone` varchar(20) DEFAULT NULL,
PRIMARY KEY (`id`)
)'''
try:
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(create_table_sql)
print 'table create successful'
except sqlite3.Error, why:
print 'create table failed:' + why.args[0]
2. 删除table
删除表，删除失败会抛出异常。
try:
#如果存在表先删除
drop_table_sql = 'DROP TABLE IF EXISTS student'
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(drop_table_sql)
print 'delete table successful'
except sqlite3.Error, why:
print 'delete table failed:' + why.args[0]
2. 插入数据
try:
save_sql = 'INSERT INTO student values (?, ?, ?, ?, ?, ?)'
data = (1, 'zhang', '男', 20, '广东省广州市', '13423****62')
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(save_sql, data)
print 'save data successful'
except sqlite3.Error, why:
print 'save data failed:' + why.args[0]
cu.execute(save_sql, data)，data参数为元组，如果是其他类型会出现错误提示，上面代码执行完毕之后会提示：
save data failed:You must not use 8-bit bytestrings unless you use a text_factory that can interpret 8-bit bytestrings (like text_factory = str). It is highly recommended that you instead just switch your application to Unicode strings.
是由于中文字符串导致的，请先确定你的IDE或者系统默认编码是utf-8,并且在中文字符串前+u来解决。
data = (1, 'Hongten', u'男', 20, u'广东省广州市', '13423****62')
3. 更新数据
更新数据，把ID等于1的Name更新为James。
update_sql = '''UPDATE student SET name = ? WHERE ID = ? '''
data = ('James', 1)
try:
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(update_sql, data)
conn.commit()
print 'table update successful'
except sqlite3.Error, why:
print 'table update failed:' + why.args[0]
4. 查询数据
#查询全部数据
fetchall_sql = '''SELECT * FROM student'''
try:
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(fetchall_sql)
content = cu.fetchall()
print content
# if len(content) > 0:
# for item in content:
# for element in item:
# print element,
# print ''
# else:
# for element in content:
# print element,
# print ''
except sqlite3.Error as why:
print "fetchall data failed:", why.args[0]
return
输出：
[(1, u'Zhang', u'\u7537', 15, u'\u5e7f\u4e1c\u7701\u5e7f\u5dde\u5e02', u'13*******62'), (2, u'Li', u'\u7537', 22, u'\u7f8e\u56fd\u65e7\u91d1\u5c71', u'15*******63'), (3, u'Zhao', u'\u5973', 18, u'\u5e7f\u4e1c\u7701\u5e7f\u5dde\u5e02', u'18*******87'), (4, u'Xi', u'\u5973', 21, u'\u5e7f\u4e1c\u7701\u5e7f\u5dde\u5e02', u'14323****32')]
中文部分仍然显示为乱码，如果要显示出中文字体，那需要按注释部分，依次打印出每个字符串
#查询一条数据
fetchall_sql = '''SELECT * FROM student WHERE ID = ? '''
data = 1
try:
conn = sqlite3.connect('test.db')
cu = conn.cursor()
d = (data,)
cu.execute(fetchall_sql, d)
content = cu.fetchall()
if len(content) > 0:
for item in content:
for element in item:
print element,
print ''
else:
for element in content:
print element,
print ''
print 'fetch the data successful'
except sqlite3.Error, why:
print 'fetch data failed:' + why.args[0]
5. 删除数据
把WHERE后面的去掉，'DELETE FROM student'，删除全部数据
#删除一条数据
'''DELETE FROM student WHERE ID = ? AND NAME = ?'''
data = (1, 'James')
try:
conn = sqlite3.connect('test.db')
cu = conn.cursor()
cu.execute(update_sql, data)
conn.commit()
print 'delete data successful'
except sqlite3.Error, why:
print 'delete data failed:' + why.args[0]
以下是封装成类后的全部代码，新人代码，如果出现问题请多多交流
# coding:utf-8
"""
@author: smartgang
@contact: zhangxingang92@qq.com
@file: SqliteHelper.py
@time: 2017/12/7 18:19
"""
import sqlite3
'''
SQLite数据库是一款非常小巧的嵌入式开源数据库软件，没有独立的维护进程，所有的维护都来自于程序本身。在python中，使用sqlite3创建数据库的连接，当数据库文件不存在时，会新建数据库文件；如果数据库文件已经存在了，则打开数据库文件。
连接对象可以是硬盘上面的数据库文件，也可以是建立在内存中的，在内存中的数据库执行完任何操作后，不需要提交事务的commit。
使用:memory:标识打开的是内存数据库
con = sqlite3.connect(":memory:")
在本地当前目录下创建数据库文件\打开数据库文件-test.db
conn = sqlite3.connect('test.db')
打开数据库时返回的对象conn就是一个数据库连接对象，它可以有以下操作：
commit() --事务提交
rollback() --事务回滚
close() --关闭一个数据库链接
cursor() --创建一个游标
创建一个游标对象：cu
cu = conn.cursor()
关闭一个游标对象
cu.close()
在sqlite3中，所有sql语句的执行都要在游标对象的参与下完成
对于游标对象cu，具有以下具体操作：
execute() --执行一条sql语句
executemany() --执行多条sql语句
close() --游标关闭
fetchone() --从结果中取出一条记录
fetchmany() --从结果中取出多条记录
fetchall() --从结果中取出所有记录
scroll() --游标滚动
'''
class SqliteHelper:
def __init__(self, dataFile):
try:
self.conn = sqlite3.connect(dataFile)
except sqlite3.Error as e:
print "连接sqlite数据库失败:", e.args[0]
def getcursor(self):
return self.conn.cursor()
def drop(self, table):
'''
if the table exist,please be carefull
'''
if table is not None and table != '':
cu = self.getcursor()
sql = 'DROP TABLE IF EXISTS ' + table
try:
cu.execute(sql)
except sqlite3.Error as why:
print "delete table failed:", why.args[0]
return
self.conn.commit()
print "delete table successful!"
cu.close()
else:
print "table does not exist！"
def create(self, sql):
'''
create database table
:param sql:
:return:
'''
if sql is not None and sql != '':
cu = self.getcursor()
try:
cu.execute(sql)
except sqlite3.Error as why:
print "create table failed:", why.args[0]
return
self.conn.commit()
print "create table successful!"
cu.close()
else:
print "sql is empty or None"
def insert(self, sql, data):
'''
insert data to the table
:param sql:
:param data:
:return:
'''
if sql is not None and sql != '':
if data is not None:
cu = self.getcursor()
try:
for d in data:
cu.execute(sql, d)
self.conn.commit()
except sqlite3.Error as why:
print "insert data failed:", why.args[0]
cu.close()
else:
print "sql is empty or None"
def fetchall(self, sql):
'''
query all data
:param sql:
:return:
'''
if sql is not None and sql != '':
cu = self.getcursor()
try:
cu.execute(sql)
content = cu.fetchall()
if len(content) > 0:
for item in content:
for element in item:
print element,
print ''
else:
for element in content:
print element,
print ''
except sqlite3.Error as why:
print "fetchall data failed:", why.args[0]
cu.close()
else:
print "sql is empty or None"
def fetchone(self, sql, data):
'''
query one data
:param sql:
:param data:
:return:
'''
if sql is not None and sql != '':
if data is not None:
cu = self.getcursor()
try:
d = (data,)
cu.execute(sql, d)
content = cu.fetchall()
if len(content) > 0:
for item in content:
for element in item:
print element,
print ''
else:
for element in content:
print element,
print ''
except sqlite3.Error as why:
print "fetch the data failed:", why.args[0]
return
cu.close()
else:
print "sql is empty or None"
def update(self, sql, data):
'''
update the data
:param sql:
:param data:
:return:
'''
if sql is not None and sql != '':
if data is not None:
cu = self.getcursor()
try:
for d in data:
cu.execute(sql, d)
self.conn.commit()
except sqlite3.Error as why:
print "update data failed:", why.args[0]
cu.close()
else:
print "sql is empty or None"
def delete(self, sql, data=None):
'''
delete the data
:param sql:
:param data:
:return:
'''
if sql is not None and sql != '':
cu = self.getcursor()
if data is not None:
try:
for d in data:
cu.execute(sql, d)
self.conn.commit()
except sqlite3.Error as why:
print "delete data failed:", why.args[0]
else:
try:
cu.execute(sql)
self.conn.commit()
except sqlite3.Error as why:
print "delete data failed:", why.args[0]
cu.close()
else:
print "sql is empty or None"
def __del__(self):
self.conn.close()
# test
TABLE_NAME = 'student'
def drop_table_test():
'''删除数据库表测试'''
print('删除数据库表测试...')
sqlhelper = SqliteHelper('test.db')
sqlhelper.drop(TABLE_NAME)
def create_table_test():
'''创建数据库表测试'''
print('创建数据库表测试...')
# 创建表（id,name,gender,age,address,phone)
create_table_sql = '''CREATE TABLE `student` (
`id` int(11) NOT NULL,
`name` varchar(20) NOT NULL,
`gender` varchar(4) DEFAULT NULL,
`age` int(11) DEFAULT NULL,
`address` varchar(200) DEFAULT NULL,
`phone` varchar(20) DEFAULT NULL,
PRIMARY KEY (`id`)
)'''
sqlhelper = SqliteHelper('test.db')
sqlhelper.create(create_table_sql)
sqlhelper = SqliteHelper('test.db')
def save_test():
'''保存数据测试...'''
print('保存数据测试...')
save_sql = 'INSERT INTO student values (?, ?, ?, ?, ?, ?)'
data = [(1, 'Zhang', u'男', 15, u'北京', '12345678910'),
(2, 'Li', u'男', 50, u'吉林省长春市', '1567891****'),
(3, 'Zhao', u'女', 40, u'黑龙江', '18*********'),
(4, 'Xi', u'女', 30, u'上海', '1**********'),
(5, 'Liao', u'男', 15, u'湖南', '12345678910'),
(6, 'Ling', u'男', 50, u'大理', '1567891****'),
(7, 'JJJ', u'女', 40, u'***', '18*********'),
(8, 'LLL', u'女', 30, u'Hongkong', '1**********')]
sqlhelper.insert(save_sql, data)
def fetchall_test():
'''查询所有数据...'''
print('查询所有数据...')
fetchall_sql = '''SELECT * FROM student'''
sqlhelper.fetchall(fetchall_sql)
def fetchone_test():
'''查询所有数据...'''
print('查询一条数据...')
fetchall_sql = '''SELECT * FROM student WHERE ID = ? '''
data = 1
sqlhelper.fetchone(fetchall_sql, data)
update_sql = '''UPDATE student SET name = ? WHERE ID = ? '''
data = [(1, 'James'),
(2, 'Kobe')]
sqlhelper.update(update_sql, data)
def update_test():
'''更新数据'''
print('更新一条数据...')
update_sql = '''UPDATE student SET name = ? WHERE ID = ? '''
data = [('James', 1),
('Kobe', 2)]
sqlhelper.update(update_sql, data)
def delete_test():
'''删除数据'''
print('删除一条数据...')
update_sql = '''DELETE FROM student WHERE ID = ? AND NAME = ?'''
data = [(1, 'James')]
sqlhelper.delete(update_sql, data)
drop_table_test()
create_table_test()
save_test()
update_test()
delete_test()
参考文章：
网友评论