一、使用conn.executemany批量执行

executemany的速度是execute的2倍。

import os
import sqlite3
import timeit

db_name = "test.db"
insert_size = 1000000


def clear(conn):
    conn.execute("DROP TABLE IF EXISTS  user")
    conn.execute("CREATE TABLE user (name VARCHAR(10),age INT)")


def execute():
    conn = sqlite3.connect(db_name)
    clear(conn)
    for i in range(insert_size):
        conn.execute("INSERT INTO user (name,age) VALUES(?,?)", ("user%s" % i, i))
    conn.close()


def execute_many():
    def data_iter():
        for i in range(insert_size):
            yield "user%s" % i, i

    conn = sqlite3.connect(db_name)
    clear(conn)
    conn.executemany("INSERT INTO user(name,age)VALUES(?,?)", data_iter())
    conn.close()


print(timeit.timeit(execute, number=1)) #6.94724779695861
print(timeit.timeit(execute_many, number=1)) #3.6068240203748756
os.remove(db_name)

二、直接用conn即可，不需要游标

使用游标大概是JDBC倡导的东西，实际上在很多数据库连接库中直接使用连接即可。
在线示例和文档中通常如下：

connection = sqlite3.connect(':memory:')
cursor = connection.cursor()
# Do something with cursor

但大多数情况下，你根本不需要光标，你可以直接使用连接对象（本文末尾会提到）。像execute和executemany类似的操作可以直接在连接上调用。以下是一个证明此事的示例：

import sqlite3
connection = sqlite3(':memory:')
# Create a table
connection.execute('CREATE TABLE events(ts, msg)')

# Insert values
connection.executemany(
 'INSERT INTO events VALUES (?,?)',
    [
        (1, 'foo'),
        (2, 'bar'),
        (3, 'baz')
    ]
)
# Print inserted rows
for row in connnection.execute('SELECT * FROM events'):
    print(row)

三、光标可用于迭代

你可能经常会看到使用fetchone或fetchall来处理SELECT查询结果的示例。但是我发现处理这些结果的最自然的方式是直接在光标上迭代：

for row in connection.execute('SELECT * FROM events'):
  print(row)

四、使用Pragmas

在你的程序中有几个 pragma 可用于调整 sqlite3 的行为。特别地，其中一个可以改善性能的是synchronous

connection.execute('PRAGMA synchronous = OFF')

此命令让sqlite停止了一些安全性检测，这可能是危险的。如果应用程序在事务中间意外崩溃，数据库可能会处于不一致的状态。所以请小心使用！但是如果要更快地插入很多行并且自己确信不需要sqlite帮忙做一些无用的检测，那么这是一个选择。

五、推迟索引创建

假设你需要在数据库上创建几个索引，而你需要在插入很多行的同时创建索引。把索引的创建推迟到所有行的插入之后可以导致实质性的性能改善。

六、使用占位符不要拼串

使用 Python 字符串操作将值包含到查询中是很方便的。但是这样做非常不安全，而 sqlite3 给你提供了更好的方法来做到这一点：

# Do not do this!
my_timestamp = 1
c.execute("SELECT * FROM events WHERE ts = '%s'" % my_timestamp)

# Do this instead
my_timestamp = (1,)
c.execute('SELECT * FROM events WHERE ts = ?', my_timestamp)

此外，使用Python％s（或格式或格式的字符串常量）的字符串插值对于executemany来说并不是总是可行，这更凸显了占位符比拼串有优势。

七、使用反射

查询执行完毕之后得到的结果是一个元组，元组的每列表示什么含义可以通过cursor.description来获得。cursor.description可以看做一个二维字符串。

def to_json(cursor: sqlite3.Cursor, row):
    # 将一行数据和cursor转化为一个dict
    a = {}
    for col in cursor.description:
        a[col[0]] = row
    return a

八、存储二进制数据

存储二进制数据就是存储bytes类型的对象。

import os
import sqlite3

import numpy as np

db_name = "test.db"
conn = sqlite3.connect(db_name)
conn.execute("DROP TABLE IF EXISTS test")
conn.execute("CREATE TABLE test(data BLOB)")
a = np.random.random((3, 4)).astype(np.float32)
print(a)
row = a[0].tobytes().hex()
conn.executemany("INSERT INTO test(data)VALUES (?)", map(lambda x: (sqlite3.Binary(x.tobytes()),), a))
b = conn.execute("SELECT * FROM test").fetchall()
for i in b:
    print(i, len(i[0]))
    i = np.fromstring(i[0], dtype=np.float32)
    print(i)
print(conn.execute("SELECT length(data) FROM test").fetchall())  # 输出16,16,16
conn.close()
os.remove(db_name)

存储二进制容易犯的一个错误就是把blob字段当成字符串来用，这样会导致一个字节使用两个16进制字符来表示，空间平白无故地多用一倍。

import os
import sqlite3

import numpy as np

db_name = "test.db"
conn = sqlite3.connect(db_name)
conn.execute("DROP TABLE IF EXISTS test")
conn.execute("CREATE TABLE test(data BLOB)")
a = np.random.random((3, 4)).astype(np.float32)
print(a)
row = a[0].tobytes().hex()
conn.executemany("INSERT INTO test(data)VALUES (?)", map(lambda x: (x.tobytes().hex(),), a))
b = conn.execute("SELECT * FROM test").fetchall()
for i in b:
    print(i, len(i[0]))
    i = np.fromstring(bytes.fromhex(i[0]), dtype=np.float32)
    print(i)
print(conn.execute("SELECT length(data) FROM test").fetchall())  # 输出32,32,32
conn.close()
os.remove(db_name)

九、sqlite中的锁

sqlite3的锁及事务类型

sqlite3总共有五种锁，按锁的级别依次是：UNLOCKED /SHARED /RESERVERD /PENDING /EXCLUSIVE。
当执行select即读操作时，需要获取到SHARED锁（共享锁），当执行insert/update/delete操作(即内存写操作时)，需要进一步获取到RESERVERD锁（保留锁），当进行commit操作(即磁盘写操作时)，需要进一步获取到EXCLUSIVE锁（排它锁）。
对于RESERVERD锁，sqlite3保证同一时间只有一个连接可以获取到保留锁，也就是同一时间只有一个连接可以写数据库(内存)，但是其它连接仍然可以获取SHARED锁，也就是其它连接仍然可以进行读操作（这里可以认为写操作只是对磁盘数据的一份内存拷贝进行修改，并不影响读操作）。
对于EXCLUSIVE锁，是比保留锁更为严格的一种锁，在需要把修改写入磁盘即commit时需要在保留锁/未决锁的基础上进一步获取到排他锁，顾名思义，排他锁排斥任何其它类型的锁，即使是SHARED锁也不行，所以，在一个连接进行commit时，其它连接是不能做任何操作的（包括读）。
PENDING锁（即未决锁），则是比较特殊的一种锁，它可以允许已获取到SHARED锁的事务继续进行，但不允许其它连接再获取SHARED锁，当已存在的SHARED锁都被释放后（事务执行完成），持有未决锁的事务就可以获得commit的机会了。sqlite3使用这种锁来防止writer starvation（写饿死）。
死锁的情况

死锁的情况

当两个连接使用begin transaction开始事务时，第一个连接执行了一次select操作（已经获取到SHARED锁），第二个连接执行了一次insert操作（已经获取到了RESERVERD锁），此时第一个连接需要进行一次insert/update/delete（需要获取到RESERVERD锁），第二个连接则希望执行commit（需要获取到EXCLUSIVE锁），由于第二个连接已经获取到了RESERVERD锁，根据RESERVERD锁同一时间只有一个连接可以获取的特性，第一个连接获取RESERVERD锁的操作必定失败，而由于第一个连接已经获取到SHARED锁，第二个连接希望进一步获取到EXCLUSIVE锁的操作也必定失败。就导致了事务死锁。

事务类型的使用原则

在用”begin transaction”显式开启一个事务时，默认的事务类型为DEFERRED，锁的状态为UNLOCKED，即不获取任何锁，如果在使用的数据库没有其它的连接，用begin就可以了。如果有多个连接都需要对数据库进行写操作，那就得使用BEGIN IMMEDIATE/EXCLUSIVE开始事务了。
使用事务的好处是：1.一个事务的所有操作相当于一次原子操作，如果其中某一步失败，可以通过回滚来撤销之前所有的操作，只有当所有操作都成功时，才进行commit，保证了操作的原子特性；2.对于多次的数据库操作，如果我们希望提高数据查询或更新的速度，可以在开始操作前显式开启一个事务，在执行完所有操作后，再通过一次commit来提交所有的修改或结束事务。
对SQLITE_BUSY的处理

当有多个连接同时对数据库进行写操作时，根据事务类型的使用原则，我们在每个连接中用BEGIN IMMEDIATE开始事务，即多个连接都尝试取得保留锁的情况，根据保留锁同一时间只有一个连接可以获取到的特性，其它连接都将获取失败，即事务开始失败，这种情况下，sqlite3将返回一个SQLITE_BUSY的错误，如果我们不希望操作就此失败而返回，就必须处理SQLITE_BUSY的情况，sqlite3提供了sqlite3_busy_handler或sqlite3_busy_timeout来处理SQLITE_BUSY，对于sqlite3_busy_handler，我们可以指定一个busy_handler来处理，并可以指定失败重试的次数。而sqlite3_busy_timeout则是由sqlite3自动进行sleep并重试，当sleep的累积时间超过指定的超时时间时，最终返回SQLITE_BUSY。需要注意的是，这两个函数同时只能使用一个，后面的调用会覆盖掉前次调用。从使用上来说，sqlite3_busy_timeout更易用一些，只需要指定一个总的超时时间，然后sqlite自己会决定多久进行重试以及重试的次数，直到达到总的超时时间最终返回SQLITE_BUSY。并且，这两个函数一经调用，对其后的所有数据库操作都有效，非常方便。

参考资料

https://www.cnblogs.com/nice107/p/8067165.html
https://zhuanlan.zhihu.com/p/26576194