Python基础 - Sqlite3基本使用

​ SQLite是C库，提供一个基于磁盘的轻量级的数据库，允许使用非标准的SQL查询语言访问数据库，不需要一个单独的服务进程。一些程序可以使用SQLite作为内部数据存储。

1. 使用sqlite3模块，首先创建一个数据库连接对象；

# 创建sqlite目录，指定新的工作目录
import os
os.system('mkdir sqlite')
os.chdir('C:\\PythonDemo\\sqlite')
import sqlite3
# 可以使用:memory:在内存中创建数据库
conn = sqlite3.connect('demo.db')

2. 拥有Connection对象后，可以创建一个Cursor对象然后调用execute方法来执行SQL命令；

# 创建游标对象
c = conn.cursor()
# 创建stocks表
c.execute('''CREATE TABLE stocks(date text, trans text, symbol text, qty real,price real)''')
# 插入一条数据
c.execute("INSERT INTO stocks VALUES ('2022-08-18','BUY','RHAT',110,45.5)")
# 提交修改
conn.commit()
# 关闭数据库连接
conn.close()

3. 在字符串中直接拼接SQL是不安全的，可以使用占位符解决；

# 不安全的做法
symbol = 'RHAT'
c.execute("SELECT * FROM stocks WHERE symbol='%s'" % symbol)
# 正确的做法
t = ('RHAT',)
c.execute('SELECT * FROM stocks WHERE symbol=?', t)
print(c.fetchone())
# 一次添加多条记录
purchases = [('2022-08-28', 'BUY', 'IBM', 1000, 45.00),
             ('2022-08-05', 'BUY', 'MSFT', 1000, 72.00),
             ('2022-08-06', 'SELL', 'IBM', 500, 53.00),
            ]
c.executemany('INSERT INTO stocks VALUES (?,?,?,?,?)', purchases)

4. 执行完select语句后，可以使用cursor作为迭代器提取数据，调用cursor的fetchone方法可以提取出一条匹配行，调用fetchall获取匹配到的列表，再次调用则取不到数据了。

>>> for row in c.execute('SELECT * FROM stocks ORDER BY price'):
	print(row)

('2022-08-28', 'BUY', 'IBM', 1000.0, 45.0)
('2022-08-18', 'BUY', 'RHAT', 110.0, 45.5)
('2022-08-06', 'SELL', 'IBM', 500.0, 53.0)
('2022-08-05', 'BUY', 'MSFT', 1000.0, 72.0)

Connection对象

class sqlite3.Connection

isolation_level隔离级别，取值为DEFERRED、IMMEDIATE、EXCLUSIVE、None，自动提交是None。

in_transaction只读属性，True表示开启了事务。

cursor(factory=Cursor)方法接受一个可选的factory参数。如果提供，它必须是一个可调用对象，返回Cursor或其子类的实例。

commit()方法提交当前的事务。如果没有调用commit()方法，数据库的操作对于其他数据库连接不可见。

rollback()方法撤销从上次提交后的任意数据库变更。

close()方法用于关闭数据库连接，close()方法不会自动调用commit()方法提交。

execute(sql[,parameters])方法是一个非标准的快捷方式，它通过调用cursor()方法来创建一个游标对象，然后使用给定的参数调用游标的execute()方法，然后返回游标对象。

executemany(sql[,parameters])方法是一个非标准的快捷方式，它通过调用cursor()方法来创建一个游标对象，然后用给定的参数调用游标的executemany()方法，然后返回游标对象。

create_function(name, num_params,func)创建一个用户定义的函数，以后可以在函数名的名称下从SQL语句中使用该函数。num_params是函数接受的参数数量(如果num_params为-1，函数可以接受任意数量的参数)，而func是一个Python可调用对象，作为SQL函数调用。该函数可以返回SQLite支持的任何类型:bytes, str, int, float和None。

import sqlite3
improt hashlib

def md5sum(t):
    return hashlib.md5(t).hexdigest()

con = sqlite3.connect(':memory:')
con.create_function("md5", 1, md5sum)
cur = con.cursor()
cur.execute('select md5(?)', (b"foo",))
print(cur.fetchone()[0])

create_aggregate(name, num_params, aggregate_class)创建用户定义的聚合函数。

import sqlite3

class MySum:
    def __init__(self):
        self.count = 0
	# 聚合类必须包含一个step方法，可以接受num_params个参数，-1则可以接受任意数量的参数
    def step(self, value):
        self.count += value
	# finalize方法返回聚合函数最终的结果
    def finalize(self):
        return self.count

con = sqlite3.connect(":memory:")
con.create_aggregate("mysum", 1, MySum)
cur = con.cursor()
cur.execute("create table test(i)")
cur.execute("insert into test(i) values (1)")
cur.execute("insert into test(i) values (2)")
cur.execute("select mysum(i) from test")
print(cur.fetchone()[0])

create_collation(name, callable)使用指定的名称和callable创建排序规则。

import sqlite3

def collate_reverse(string1, string2):
    if string1 == string2:
        return 0
    elif string1 < string2:
        return 1
    else:
        return -1

con = sqlite3.connect(":memory:")
con.create_collation("reverse", collate_reverse)
cur = con.cursor()
cur.execute("create table test(x)")
cur.executemany("insert into test(x) values (?)", [("a",), ("b",)])
cur.execute("select x from test order by x collate reverse")
for row in cur:
    print(row)
con.close()
# 删除排序规则
con.create_collation("reverse", None)

interrupt()方法可以从不同的线程调用此方法，以中止可能在连接上执行的任何查询。然后查询将中止，调用者将得到一个异常。

row_factory可以将此属性更改为一个可调用对象，该可调用对象接受游标和原始行作为元组，并返回实际结果行。通过这种方式，您可以实现更高级的返回结果的方法。

import sqlite3

def dict_factory(cursor, row): 
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d

con = sqlite3.connect(":memory:")
con.row_factory = dict_factory
cur = con.cursor()
cur.execute("select 1 as a")
print(cur.fetchone()["a"])

iterdump()以SQL文本格式转储数据库，返回迭代器，把一个内存数据库中的数据转储时很有用。

import sqlite3
con = sqlite3.connect("demo.db")
with open('dump.sql', 'w') as f:
    for line in con.iterdump():
        f.write("%s\n"%line)

Cursor对象

class sqlite3.Cursor

execute(sql[, parameters]) 执行SQL语句，SQL语句可以是参数化的，支持问号占位符以及命名占位符的方式。

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("create table people (name_last, age)")
who = "Thoms"
age = 24
# 问号占位符方式，execute方法只能执行一条SQL，executescript方法可以执行多条SQL
cur.execute("insert into people values (?, ?)", (who, age))
# 命名占位符方式
cur.execute("select * from people where name_last=:who and age=:age", {"who":who, "age": age})
print(cur.fetchone())

executemany(sql, seq_of_parameters) 对序列seq_of_parameters中的所有参数序列或映射执行SQL命令。

import sqlite3

class IterChars:
    def __init__(self):
        self.count = ord('a')

    def __iter__(self):
        return self

    def __next__(self):
        if self.count > ord('z'):
            raise StopIteration
        self.count += 1
        return (chr(self.count - 1),) # this is a 1-tuple

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("create table characters(c)")
theIter = IterChars()
cur.executemany("insert into characters(c) values (?)", theIter)
cur.execute("select c from characters")
print(cur.fetchall())

生成器的简单例子

import sqlite3
import string

def char_generator():
    for c in string.ascii_lowercase:
        yield (c,)

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("create table characters(c)")
cur.executemany("insert into characters(c) values (?)", char_generator())
cur.execute("select c from characters")
print(cur.fetchall())

executescript(sql_script)这是一次执行多个SQL语句的一种非标准的方便方法。它首先发出一个COMMIT语句，然后执行作为参数获取的SQL脚本。

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.executescript("""
    create table person(
        firstname,
        lastname,
        age
    );

    create table book(
        title,
        author,
        published
    );

    insert into book(title, author, published)
    values (
        'Dirk Gently''s Holistic Detective Agency',
        'Douglas Adams',
        1987
    );
    """)

fetchone()：获取查询结果集的下一行，返回单个序列，或在没有更多数据可用时返回None。

fetchmany(size=cursor.arraysize)获取查询结果的下一组行，返回一个列表。当没有更多的行可用时，返回一个空列表。每次调用获取的行数由size参数指定。如果没有指定，游标的arraysize将决定要获取的行数。该方法应该尝试获取size参数所指示的尽可能多的行。如果由于指定的行数不可用而不可能这样做，则可能返回更少的行。注意size参数涉及到性能方面的考虑。为了获得最佳性能，通常最好使用arraysize属性。

fetchall()获取查询结果的所有行，返回一个列表。注意，游标的arraysize属性会影响此操作的性能。如果没有可用的行，则返回空列表。

close()立即关闭游标，调用后游标将不可用。

Row对象

class sqlite3.Row

Row实例用作Connection对象的高度优化的row_factory。它支持通过列名和索引、迭代、表示、相等测试和len()进行映射访问。如果两个Row对象具有完全相同的列并且它们的成员相等，则它们比较时相等。

conn = sqlite3.connect(":memory:")
c = conn.cursor()
c.execute('''create table stocks
(date text, trans text, symbol text,
 qty real, price real)''')
c.execute("""insert into stocks
          values ('2022-08-18','BUY','RHAT',100,35.24)""")
conn.commit()
c.close()

conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute('select * from stocks')
r = c.fetchone()
type(r)
tuple(r)
len(r)
r[2]
r.keys()  # 返回表的列名列表
r['qty']
for member in r:
    print(member)

SQLite和Python类型

Python类型	SQLite类型
None	NULL
int	INTEGER
float	REAL
str	TEXT
bytes	BLOB