15.4.SQLite

import sqlite3

Windows 10
Python 3.8.8 @ MSC v.1928 64 bit (AMD64)
Latest build date 2021.05.13
sqlite3 version:  2.6.0
SQLite version:  3.33.0

使用示例

create a connection

# create database from a local file
# db_url = f"example.db"
# create database from memory
db_url = ":memory:"
conn = sqlite3.connect(db_url)

open a cursor to execute SQL

# Open a cursor to perform database operations
cur = conn.cursor()

# Execute a command
cur.execute("select * from sqlite_master;")
# cur.executemany()

cur.fetchall()
# fetchone
# fetchmany

[]

commit and close connection

# Make the changes to the database persistent
conn.commit()

# Close communication with the database
cur.close()
conn.close()

占位符

一般来说，SQL 操作语句可能会使用 Python 变量的值。实际上，不应该使用 Python 的字符串操作来组合 SQL 语句，因为这样做会使程序容易受到 SQL 注入攻击的影响（可以参考xkcd Webcomic）。

# Never do this -- insecure!
symbol = 'RHAT'
cur.execute("SELECT * FROM stocks WHERE symbol = '%s'" % symbol)

应该使用 DB-API 的参数替换方式，将占位符放在要使用值的任何位置，然后将提供值的元组作为execute()方法的第二个参数。

SQL语句可以使用两种风格的占位符：问号（qmark style）或命名的占位符（named style）。对于qmark style，参数必须为序列；对于named style，它可以是序列或字典。序列的长度必须与占位符的数量匹配，否则将引发ProgrammingError。如果传入字典作为参数，则它必须包含所有命名参数的键。任何多余的项目都将被忽略。这是两种样式的示例：

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("create table lang (lang_name, lang_age)")

# This is the qmark style:
cur.execute("insert into lang values (?, ?)", ("C", 49))

# The qmark style used with executemany():
lang_list = [
    ("Fortran", 64),
    ("Python", 30),
    ("Go", 11),
]
cur.executemany("insert into lang values (?, ?)", lang_list)

# And this is the named style:
cur.execute("select * from lang where lang_name=:name and lang_age=:age",
            {"name": "C", "age": 49})
print(cur.fetchall())

con.close()

[('C', 49)]

顶层函数

sqlite3.connect

sqlite3.connect(database[, timeout, detect_types, isolation_level,
                check_same_thread, factory, cached_statements, uri])

• database：一个path-like对象，不仅仅是字符串（Python3.7新增），给出了要打开的SQLite数据库的路径。也可以传入":memory:"，此时会打开驻留在内存中的数据库。

• timeout：当通过多个连接访问数据库，并且其中一个进程修改了数据库时，SQLite数据库将被锁定，直到提交该事务为止。timeout 参数指定 connection 应该等待多长时间才引发异常。默认值为5.0（五秒）。

• check_same_thread：默认情况下，check_same_thread=True，只有创建 connection 的线程可以使用该 connection。如果设置为False，则返回的 connection 可以在多个线程之间共享。当多个线程使用相同 connection 时，应该对写入操作进行序列化，以避免数据损坏。

• cached_statements：sqlite3 模块在内部使用语句高速缓存来避免SQL解析开销。如果想要显式设置为连接缓存的语句数，则可以设置cached_statements参数。默认是缓存100条语句。

• url：如果uri=true，则将database参数解释为URI，Python3.4添加的参数。这可以用来指定一些选项。例如，要以只读模式打开数据库：

  db = sqlite3.connect('file:path/to/database?mode=ro', uri=True)
  

sqlite3.register_converter

sqlite3.register_converter(typename, callable)

注册一个callable，将数据库中的字节串转换为自定义Python类型。即为typename类型的所有数据库值调用callable。需要为connect()方法传入detect_types参数，开启类型检测功能。注意，typename和查询中类型名称在匹配时不区分大小写。

sqlite3.register_adapter

sqlite3.register_adapter(type, callable)

注册一个 callable 以将自定义 Python 类型转换为 SQLite 支持的类型。callable 接受 Python 值作为单个参数，并且必须返回 int、float、str或bytes类型。

sqlite3.complete_statement

sqlite3.complete_statement(sql)

如果 SQL 字符串包含一个或多个以分号结尾的完整 SQL 语句，则返回True。它不会验证 SQL 语句的语法是否正确。

Connection对象的属性和方法

isolation_level：获取或设置当前的隔离级别。

in_transaction：如果事务处于活动状态（有未提交的更改），则为True；否则为False。

create_function(name, num_params, func, ***, deterministic=False)

创建一个名称为name的自定义函数，之后你可以在SQL语句中使用该函数。num_params是函数接受的参数数量，如果num_params=-1，则该函数可以接受任意数量的参数。func是可调用的SQL函数。如果deterministic=True，则将创建的函数标记为deterministic，这使SQLite可以执行额外的优化。SQLite 3.8.3或更高版本支持此标志，否则，将引发NotSupportedError。该函数应该返回SQLite支持的任何类型：bytes、str、int、float和None。

import sqlite3
import 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])

con.close()

acbd18db4cc2f85cedef654fccc4a4d8

create_aggregate(name, num_params, aggregate_class)

创建一个用户定义的聚合函数。aggregate_class类必须实现一个step方法，该方法可以接受的参数数量为num_params，如果num_params=-1，则可以接受任意数量的参数。还需要实现finalize方法，该方法将返回聚合的最终结果。finalize方法应该返回SQLite支持的类型。

import sqlite3

class MySum:
    def __init__(self):
        self.count = 0

    def step(self, value):
        self.count += value

    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])

con.close()

3

create_collation(name, callable)

创建一个排序规则。callable对象接收两个字符串变量，如果第一个字符串排在第二个后面，返回-1；如果两者排序相同，则返回0；如果第一个字符串排在第二个前面，则返回1。注意，callable对象将以 Python bytestrings 的形式获取其参数，通常以UTF-8进行编码。

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()

('a',)
('b',)

interrupt()

可以从其他线程调用此方法，以中止可能在连接上执行的任何查询。查询将中止，调用该查询的地方将返回异常。

将连接用作上下文管理器

连接对象可用作自动提交或回滚事务的上下文管理器。如果发生异常，则事务将回滚。

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("create table person (id integer primary key, firstname varchar unique)")

# Successful, con.commit() is called automatically afterwards
with con:
    con.execute("insert into person(firstname) values (?)", ("Joe",))

# con.rollback() is called after the with block finishes with an exception, the
# exception is still raised and must be caught
try:
    with con:
        con.execute("insert into person(firstname) values (?)", ("Joe",))
except sqlite3.IntegrityError:
    print("couldn't add Joe twice")

# Connection object used as context manager only commits or rollbacks transactions,
# so the connection object should be closed manually
con.close()

couldn't add Joe twice