数据库的建立
conn = sqlite3.connect('db名')
这是建立connect连接时用到的,当项目中不存在该数据库时,会自动创建。
数据类型:
NULL: 表示该值为NULL值。
INTEGER: 无符号整型值。
REAL: 浮点值。
TEXT: 文本字符串,存储使用的编码方式为UTF-8、UTF-16BE、UTF-16LE。
BLOB: 存储Blob数据,该类型数据和输入数据完全相同。
建表例子:
CREATE TABLE Custom (
id INTEGER PRIMARY KEY DEFAULT NULL,
custom_name VARCHAR(50) NOT NULL,
remark TEXT DEFAULT NULL DEFAULT '无'
);
CREATE TABLE Provice_info(
id INTEGER PRIMARY KEY NOT NULL,
custom INT NOT NULL,
provice_name VARCHAR(20) NOT NULL,
first_weight_num REAL,
first_weight_price REAL,
next_weight_price REAL,
FOREIGN KEY (custom) REFERENCES Custom(id) on delete cascade
)
常见操作
使用sqlite,首先需要建立connect连接,然后通过cursor(游标)来对数据进行操作(execute),当是查操作时,就不需要connect.commit()
了,可以通过cursor.fetchone() / cursor.fetchall()
来获取查到的数据,前者为一个元组,后者为一个列表,列表中存储着元组,最后记得关闭connect连接就行了。
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute(SQL)
conn.commit()
conn.close()
经验操作
当使用SQLite进行操作时,最好以表为类来面向对象编程。
在add操作时,可以在
connect.commit()
后通过cursor.lastrowid
来获取刚刚创建的数据的id并返回,有时会有意想不到的方便。在delete操作时,当有外键时,想联级删除的话,除了在创建表的时候加上
on delete cascade
还需要通过cursor.execute("PRAGMA foreign_keys=ON")
这条语句才能实现联级删除。在针对查找方面,可以创建多个查找方法,如精确查找,模糊匹配,查询全部,这些有时有意想不到的方便。
针对防止SQL注入,可以通过SQLite特有的语法
?
,结合execute(SQL, (parameter1, parameter2...))
来实现。
例子:
class Custom:
def __init__(self):
self.conn = sqlite3.connect('test.db')
self.cur = self.conn.cursor()
def add_custom(self, custom_name, remark='无'):
"""
添加一条客户数据,加了去重检查,保证客户数据唯一性
:param custom_name:
:param remark:
:return: id
"""
if self.find_custom(custom_name) > 0:
raise Exception('你输入的用户名 {} 已存在!'.format(custom_name))
SQL = """
INSERT INTO Custom (custom_name, remark) VALUES (?, ?);
"""
# 防止SQL注入攻击
self.cur.execute(SQL, (custom_name, remark))
self.conn.commit()
custom_id = self.cur.lastrowid
return custom_id
def delete_custom(self, custom_id):
SQL = """
DELETE FROM Custom WHERE id=?
"""
# SQLite在3.6.19版本中才开始支持外键约束,但是为了兼容以前的程序,
# 默认并没有启用该功能,如果要启用该功能每次都要需要使用如下语句:PRAGMA foreign_keys = ON来打开。
self.cur.execute("PRAGMA foreign_keys=ON")
self.cur.execute(SQL, (custom_id,))
self.conn.commit()
def update_custom(self, custom_id, custom_name, remark='无'):
SQL = """
UPDATE Custom SET custom_name=?, remark=? WHERE id=?
"""
self.cur.execute(SQL, (custom_name, remark, custom_id))
self.conn.commit()
def find_custom(self, custom_name):
"""
通过客户名找到id
:param custom_name:
:return: id(int) / 0(没有该数据时)
"""
SQL = """
SELECT id FROM Custom WHERE custom_name=?
"""
self.cur.execute(SQL, (custom_name,))
custom_id = self.cur.fetchone()
return custom_id[0] if custom_id else 0
def find_custom_like(self, custom_name):
"""
通过客户名找到id
:param custom_name:
:return: id(int) / 0(没有该数据时)
"""
SQL = """
SELECT id FROM Custom WHERE custom_name LIKE '%{}%'
""".format(custom_name)
self.cur.execute(SQL)
custom_id = self.cur.fetchall()
print(custom_id)
if len(custom_id) > 1:
raise Exception('客户名: {} 错误,请传入正确的客户名!'.format(custom_name))
return custom_id[0][0] if custom_id else 0
def fetch_custom(self):
"""
返回所有custom对象
:return: {'客户1': 'remark1', '客户2': 'remark2' ...}
"""
SQL = """
SELECT custom_name, remark FROM Custom
"""
self.cur.execute(SQL)
customs = self.cur.fetchall()
data = {custom_name: remark for custom_name, remark in customs}
return data
def close(self):
self.cur.close()
self.conn.close()