sqlite3【SQLite データベース】^{3.2～3.113.12}
sqlite3【SQLite モジュール】
Connection【コネクション クラス】
Cursor【カーソル クラス】
Row【行 クラス】
Blob【BLOB クラス】^3.11

sqlite3.complete_statemen(sql)

戻り値チェック結果 (True：有効 / False：無効)
sqlSQL 文

import sqlite3

lst = [
    # True
    "select * from tbl;",
    "select;",
    ";",
    "'string';",

    # False (セミコロンなし)
    "select * from tbl",
    # False (文字列が閉じられていない)
    "'string ;",
]

for sql in lst:
    print(sqlite3.complete_statement(sql), sql)

sqlite3.connect(database, timeout=5.0, detect_types=0, isolation_level='DEFERRED', check_same_thread=True, factory=sqlite3.Connection, cached_statements=100 1283.11, uri3.4 =False, *3.12, autocommit3.12 =sqlite3.LEGACY_TRANSACTION_CONTROL)

Connection【コネクション クラス】オブジェクト
factory指定で継承クラス
databaseデータベース文字列 (パス風オブジェクトも可 3.7 )
ファイルのパス名 (絶対パス または 相対パス)
":memory:"RAM 使用
URI 3.4uri引数指定で URI 解釈 (オプション指定可)
timeoutタイムアウト秒 (デフォルト：5秒)
detect_types型検出
以下の組み合せ (詳細は、register_converter【変換関数登録 (SQLite ⇒ Python)】)
0型検出なし
sqlite3.PARSE_DECLTYPES宣言時の型検出
sqlite3.PARSE_COLNAMES取得時の型検出
isolation_levelトランザクション分離レベル (詳細はSQLite - BEGIN TRANSACTION (英語) 参照)
autocommit (トランザクション制御)が sqlite3.LEGACY_TRANSACTION_CONTROL(旧仕様)で有効 3.12
(Connection.isolation_level【トランザクション分離レベル】属性と同等)
None自動コミット
"DEFERRED" (デフォルト)延期 (自動コミットなし)
"IMMEDIATE"即時
"EXCLUSIVE"排他的
check_same_thread同一スレッドチェック有無
True作成スレッドのみ接続使用可能
False複数スレッドで接続共有可能
factory戻り値のConnection【コネクション クラス】継承クラス
cached_statementsキャッシュSQL文数 (デフォルト：100 1283.11 )
uri3.4database の URI 指定有無 (True：URI 指定)
以下、キーワード引数 3.12
autocommit3.12トランザクション制御
Falseなし
TrueSQLite の自動コミットモード
sqlite3.LEGACY_TRANSACTION_CONTROL旧仕様 (isolation_level (トランザクション分離レベル)を使用)

監査イベントsqlite3.connect(database) 3.8
監査イベントsqlite3.connect/handle(connection_handle) 3.10

import sqlite3

con = sqlite3.connect("./database/test.db")
con.execute("DROP TABLE IF EXISTS tbl")
con.execute("CREATE TABLE tbl(name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
con.commit()

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')

con.close()

sqlite3.enable_callback_tracebacks(flag, /3.10 )

戻り値なし
以下、位置引数 3.10
flagトレースバック出力フラグ (True：有効 / False：無効)

import sqlite3

def user_func():
    # 存在しない表に挿入
    con.execute("INSERT INTO tbl VALUES(1)")
    return None


con = sqlite3.connect(":memory:")
con.create_function("user", 0, user_func)
sqlite3.enable_callback_tracebacks(False)
con.execute("select user()")
# 例外
sqlite3.enable_callback_tracebacks(True)
con.execute("select user()")
# 例外

【出力例】
sqlite3.enable_callback_tracebacks(False) 指定

Traceback (most recent call last):
  File "～.py", line nn, in <module>
    con.execute("select user()")
sqlite3.OperationalError: user-defined function raised exception

【出力例】
sqlite3.enable_callback_tracebacks(True) 指定

Traceback (most recent call last):
  File "～.py", line nn, in user_func
    con.execute("INSERT INTO tbl VALUES(1)")
sqlite3.OperationalError: no such table: tbl
Traceback (most recent call last):
  File "～.py", line nn, in <module>
    con.execute("select user()")
sqlite3.OperationalError: user-defined function raised exception

sqlite3.register_adapter(type, adapter, /3.10 )

戻り値なし
以下、位置引数 3.10
typePython の型
adapter適合関数 (呼び出し可能オブジェクト)

sqlite3.register_converter(typename, converter, /3.10)

戻り値なし
以下、位置引数 3.10
typename型の名前 (SQLに記述・大小文字の区別なし)
converter変換関数 (呼び出し可能オブジェクト)

import sqlite3

def adapter(lst):
    # リスト型を文字列に変換
    return ",".join(lst)


def converter(bstr):
    # バイト文字列をリスト型に変換
    s = bstr.decode()
    lst = s.split(",")
    return lst


sqlite3.register_adapter(list, adapter)
sqlite3.register_converter("LIST", converter)

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES)
con.execute("create table tbl(id INTEGER, userlist LIST)")
lst1 = ['ITEM_11', 'ITEM_12', 'ITEM_13']
lst2 = ['ITEM_21', 'ITEM_22', 'ITEM_23']
con.execute("INSERT INTO tbl VALUES(?, ?)", (1, lst1))
con.execute("INSERT INTO tbl VALUES(?, ?)", (2, lst2))
con.commit()

for row in con.execute("select * from tbl"):
    print(row)
# 出力：(1, ['ITEM_11', 'ITEM_12', 'ITEM_13'])
# 出力：(2, ['ITEM_21', 'ITEM_22', 'ITEM_23'])

con.close()

import sqlite3

def adapter(lst):
    # リスト型を文字列に変換
    return ",".join(lst)


def converter(bstr):
    # バイト文字列をリスト型に変換
    s = bstr.decode()
    lst = s.split(",")
    return lst


sqlite3.register_adapter(list, adapter)
sqlite3.register_converter("LIST", converter)

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_COLNAMES)
con.execute("create table tbl(id, userlist)")
lst1 = ['ITEM_11', 'ITEM_12', 'ITEM_13']
lst2 = ['ITEM_21', 'ITEM_22', 'ITEM_23']
con.execute("INSERT INTO tbl VALUES(?, ?)", (1, lst1))
con.execute("INSERT INTO tbl VALUES(?, ?)", (2, lst2))
con.commit()

for row in con.execute("select id, userlist as 'userlist [LIST]' from tbl"):
    print(row)
# 出力：(1, ['ITEM_11', 'ITEM_12', 'ITEM_13'])
# 出力：(2, ['ITEM_21', 'ITEM_22', 'ITEM_23'])

con.close()

import sqlite3
import datetime

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES)
con.execute("create table tbl(d DATE, dt TIMESTAMP)")
d = datetime.date.fromisoformat("2012-03-04")
dt = datetime.datetime.fromisoformat("2012-03-04T05:06:07")
con.execute("INSERT INTO tbl VALUES(?,?)", (d, dt))
con.commit()

for row in con.execute("select * from tbl"):
    print(row)
# 出力：(datetime.date(2012, 3, 4), datetime.datetime(2012, 3, 4, 5, 6, 7))

con.close()

Connection.row_factory
下記形式の呼び出し可能オブジェクト

import sqlite3

def user_row(cursor, row):
    # list 型に変換
    return [*row]


con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
con.commit()

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')

con.row_factory = user_row
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：['NAME_1', 'MEMO_1']
# 出力：['NAME_2', 'MEMO_2']

con.close()

Connection.text_factory
下記形式の呼び出し可能オブジェクト

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(memo1 TEXT, memo2 TEXT)")
con.execute("INSERT INTO tbl VALUES('a b c', 'A B C')")
con.execute("INSERT INTO tbl VALUES('あ い う', 'ア イ ウ')")
con.commit()

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('a b c', 'A B C')
# 出力：('あ い う', 'ア イ ウ')

con.text_factory = bytes
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：(b'a b c', b'A B C')
# 出力：(b'\xe3\x81\x82 \xe3\x81\x84 \xe3\x81\x86', b'\xe3\x82\xa2 \xe3\x82\xa4 \xe3\x82\xa6')

con.close()

Connection.total_changes 

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
print(con.total_changes)
# 出力：0

con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
con.execute("INSERT INTO tbl VALUES('NAME_3', 'MEMO_3')")
print(con.total_changes)
# 出力：3

con.execute("UPDATE tbl SET memo='MEMO_02' WHERE name='NAME_2'")
con.execute("UPDATE tbl SET memo='MEMO_03' WHERE name='NAME_3'")
print(con.total_changes)
# 出力：5

con.execute("DELETE FROM tbl WHERE name='NAME_3'")
print(con.total_changes)
# 出力：6

con.commit()

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_02')

con.close()

Connection.backup(target, *, pages=0, progress=None, name="main", sleep=0.250) 3.7

戻り値なし
targetコピー先 (別のConnection【コネクション クラス】インスタンス)
pages (キーワード引数)ページ
≦0一括
0＜最大指定ページ毎
progress (キーワード引数)進行状況関数 (呼び出し可能オブジェクト：詳細は下記参照)
name (キーワード引数)コピー元データベース名
"main"メインデータベース
"temp"一時データベース
接続データベース名ATTACH DATABASE ステートメントの AS 指定
sleep (キーワード引数)次ページバックアップまでのスリープ秒数 (整数・浮動小数点数)

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
con.commit()

con2 = sqlite3.connect("./database/backup.db")
con.backup(con2)
con.close()

for row in con2.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')

con2.close()

import sqlite3

def progress(status, remaining, total):
    print(f"progress({status}, {remaining}, {total})")
    print(f"({total - remaining} / {total})")


con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
for i in range(200):
    name = "NAME_" + str(i)
    memo = "MEMO_" + str(i)
    con.execute("INSERT INTO tbl VALUES(?, ?)", (name, memo))

con.commit()

con2 = sqlite3.connect("./database/backup.db")
con.backup(con2, pages=1, progress=progress)
# 出力：progress(0, 3, 4)
# 出力：(1 / 4)
# 出力：progress(0, 2, 4)
# 出力：(2 / 4)
# 出力：progress(0, 1, 4)
# 出力：(3 / 4)
# 出力：progress(101, 0, 4)
# 出力：(4 / 4)

con.close()
con2.close()

Connection.blobopen(table, column, row, /, *, readonly=False, name='main') 3.11

戻り値Blob【BLOB】オブジェクト
table (str)テーブル名
column (str)列名
row (int)行番号

readonly (bool)読み取り専用の指定 (True：読み取り専用 / False：読み書き可能)
name (str)データベース名

sqlite3.OperationalError 例外ROWID なしテーブル (WITHOUT ROWID 指定)

Blob【BLOB クラス】の例 参照

Connection.close()

戻り値なし

import sqlite3

con = sqlite3.connect(":memory:")
# 各種操作
con.close()

Connection.in_transaction  3.2

Connection.isolation_level

Connection.commit()

戻り値なし

Connection.rollback()

戻り値なし

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
print(con.in_transaction)
# 出力：False

con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
print(con.in_transaction)
# 出力：True
con.rollback()
print(con.in_transaction)
# 出力：False

con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
print(con.in_transaction)
# 出力：True
con.commit()
print(con.in_transaction)
# 出力：False

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_2', 'MEMO_2')
print(con.in_transaction)
# 出力：False

con.close()

import sqlite3

con = sqlite3.connect("./database/with.db")
con.execute("DROP TABLE IF EXISTS tbl")
con.execute("CREATE TABLE tbl(name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_11', 'MEMO_11')")
con.execute("INSERT INTO tbl VALUES('NAME_12', 'MEMO_12')")
# 自動コミット なし
con.close()

con = sqlite3.connect("./database/with.db")
con.isolation_level = None
con.execute("INSERT INTO tbl VALUES('NAME_21', 'MEMO_21')")
con.execute("INSERT INTO tbl VALUES('NAME_22', 'MEMO_22')")
# 自動コミット

con.isolation_level = "DEFERRED"
con.execute("INSERT INTO tbl VALUES('NAME_23', 'MEMO_23')")
con.execute("INSERT INTO tbl VALUES('NAME_24', 'MEMO_24')")
# 自動コミット なし
con.close()

con = sqlite3.connect("./database/with.db")
with con:
    con.execute("INSERT INTO tbl VALUES('NAME_31', 'MEMO_31')")
    con.execute("INSERT INTO tbl VALUES('NAME_32', 'MEMO_32')")
    # 自動コミット
con.close()

con = sqlite3.connect("./database/with.db")
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_21', 'MEMO_21')
# 出力：('NAME_22', 'MEMO_22')
# 出力：('NAME_31', 'MEMO_31')
# 出力：('NAME_32', 'MEMO_32')

con.close()

Connection.create_aggregate(name, num_params, aggregate_class)

戻り値なし
nameユーザ集計(集合)関数名
num_paramsユーザ集計(集合)関数の引数の数
aggregate_classstep・finalize メソッドを持つユーザ集計(集合)クラス (詳細は下記参照)

Connection.create_collation(name, callable, /3.10 )

戻り値なし
以下、位置引数 3.11
name照合順序名 (ASCII文字のみ任意のUnicode文字3.11 )
callable呼び出し可能オブジェクト (詳細は下記参照)

Connection.create_function(name, num_params, func, *, deterministic3.8 =False)

戻り値なし
nameユーザー定義関数名
num_paramsユーザー定義関数の引数の数
funcユーザー定義関数 (呼び出し可能オブジェクト)
以下、キーワード引数 3.8
deterministic 3.8決定的関数か否か (SQLite 3.8.3 以上)
True決定的関数 (SQLite 最適化マーク)
False非決定的関数

import sqlite3

class CalcClass:
    def __init__(self):
        self.sum = 0

    def step(self, value1, value2, value3):
        self.sum += value1 + value2 + value3

    def finalize(self):
        return self.sum


con = sqlite3.connect(":memory:")
con.create_aggregate("calc", 3, CalcClass)
cur = con.cursor()
cur.execute("CREATE TABLE tbl(i, j, k)")
cur.execute("INSERT INTO tbl(i, j, k) VALUES (1, 2, 3)")
cur.execute("INSERT INTO tbl(i, j, k) VALUES (2, 3, 4)")
cur.execute("SELECT calc(i, j, k) FROM tbl")
print(cur.fetchone()[0])
# 出力：15
# (1+2+3) + (2+3+4) = 15
con.close()

import sqlite3

# 文字列長で比較
def collate_len(string1, string2):
    len1 = len(string1)
    len2 = len(string2)
    if len1 == len2:
        return 0
    elif len1 > len2:
        return 1
    else:
        return -1


con = sqlite3.connect(":memory:")
con.create_collation("length", collate_len)

con.execute("CREATE TABLE tbl (name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_11')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_222')")
con.execute("INSERT INTO tbl VALUES('NAME_3', 'MEMO_3')")
con.commit()

for row in con.execute("SELECT * FROM tbl ORDER BY memo"):
    print(row)
# 出力：('NAME_1', 'MEMO_11')
# 出力：('NAME_2', 'MEMO_222')
# 出力：('NAME_3', 'MEMO_3')

for row in con.execute("SELECT * FROM tbl ORDER BY memo collate length"):
    print(row)
# 出力：('NAME_3', 'MEMO_3')
# 出力：('NAME_1', 'MEMO_11')
# 出力：('NAME_2', 'MEMO_222')

con.close()

import sqlite3

def rectangle(width, height, depth):
    return width * height * depth


con = sqlite3.connect(":memory:")
con.create_function("rect", 3, rectangle)
cur = con.cursor()
cur.execute("select rect(?, ?, ?)", (2, 3, 4))
print(cur.fetchone()[0])
# 出力：24
con.close()

Connection.create_window_function(name, num_params, aggregate_class, /) 3.11

以下、位置引数
name (str)ウィンドウ関数名
num_params (int)引数の数 (-1：任意)
aggregate_class (class | None)対応クラス
    クラス：必要な実装メソッドは下記参照
    None：ウィンドウ関数削除

NotSupportedError 例外未サポート (SQLite 3.25.0 以降が必要)

aggregate_class (対応クラス) 仕様

import sqlite3

class MyWindow:
    def __init__(self):
        self.sum = 0

    # 現在ウィンドウに行追加
    def step(self, dummy, value):
        print(f'step({dummy}, {value}) ({self.sum=}) + ({value=}) ⇒ {self.sum + value}')
        self.sum += value

    # 現在の結果返却
    def value(self):
        print(f'value({self.sum})\n')
        return self.sum

    # 現在ウィンドウから行削除
    def inverse(self, dummy, value):
        print(f'inverse({dummy}, {value}) ({self.sum=}) - ({value=}) ⇒ {self.sum - value}')
        self.sum -= value

    # 最終結果返却
    def finalize(self):
        print(f'finalize() {self.sum=}')
        return self.sum

con = sqlite3.connect(':memory:')
cur = con.execute('CREATE TABLE tbl(name, point)')
values = [
    ('name_1', 1),
    ('name_2', 2),
    ('name_3', 3),
    ('name_4', 4),
    ('name_5', 5),
    ('name_6', 6),
]
cur.executemany('INSERT INTO tbl VALUES(?, ?)', values)
# 前１行・後２行のポイント加算
con.create_window_function('window_name', 2, MyWindow)
cur.execute('''
    SELECT name, window_name(name, point) OVER (
        ORDER BY name ROWS BETWEEN 1 PRECEDING AND 2 FOLLOWING
    ) AS sum_point
    FROM tbl ORDER BY name
''')
# 出力：
# step(name_1, 1) (self.sum=0) + (value=1) ⇒ 1
# step(name_2, 2) (self.sum=1) + (value=2) ⇒ 3
# step(name_3, 3) (self.sum=3) + (value=3) ⇒ 6
# value(6)
#
print(cur.fetchall())
# 出力：
# step(name_4, 4) (self.sum=6) + (value=4) ⇒ 10
# value(10)
#
# inverse(name_1, 1) (self.sum=10) - (value=1) ⇒ 9
# step(name_5, 5) (self.sum=9) + (value=5) ⇒ 14
# value(14)
#
# inverse(name_2, 2) (self.sum=14) - (value=2) ⇒ 12
# step(name_6, 6) (self.sum=12) + (value=6) ⇒ 18
# value(18)
#
# inverse(name_3, 3) (self.sum=18) - (value=3) ⇒ 15
# value(15)
#
# inverse(name_4, 4) (self.sum=15) - (value=4) ⇒ 11
# value(11)
#
# finalize() self.sum=11
# [('name_1', 6), ('name_2', 10), ('name_3', 14), ('name_4', 18), ('name_5', 15), ('name_6', 11)]

Connection.cursor(factory=Cursor)

戻り値Cursor【カーソル クラス】オブジェクト
factory呼び出し可能オブジェクト (Cursor【カーソル クラス】 または そのサブクラスのインスタンス返却)

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE tbl (name, memo)")
cur.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.commit()

cur.execute("SELECT * FROM tbl")
print(cur.fetchone())
# 出力：('NAME_1', 'MEMO_1')

con.close()

Connection.enable_load_extension(enabled, /3.10 ) 3.2

戻り値なし
以下、位置引数 3.10
enabledSQLite 拡張の読み込み可否 (True：許可 / False：禁止)

監査イベントsqlite3.enable_load_extension(connection, enabled) 3.10

Connection.load_extension(path, /3.12, *3.12, entrypoint3.12 =None) 3.2

戻り値なし
以下、位置引数 3.12
path (str)SQLite 拡張ライブラリのパス
以下、キーワード引数 3.12
entrypoint (str)3.12エントリポイント名 (None：SQLiteが推測)

監査イベントsqlite3.load_extension(connection, path) 3.10

Connection.execute(sql, parameters=(), /3.10 )

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sqlSQL文
parametersSQL文のパラメータ

Connection.executemany(sql, parameters, /3.10 )

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sqlSQL文
parametersSQL文のパラメータ (シーケンス・マッピング)

Connection.executescript(sql_script, /3.10 )

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sql_scriptSQLスクリプト

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(id, memo)")

# execute()
lst = [1, "MEMO_01"]
con.execute("INSERT INTO tbl VALUES(?, ?)", lst)
tpl = (2, "MEMO_02")
con.execute("INSERT INTO tbl VALUES(?, ?)", tpl)
dic = {"id": 3, "memo": "MEMO_03"}
con.execute("INSERT INTO tbl VALUES(:id, :memo)", dic)

# executemany()
lst1 = [
    (11, "MEMO_11"),
    (12, "MEMO_12"),
]
con.executemany("INSERT INTO tbl VALUES(?, ?)", lst1)
tpl1 = (
    [13, "MEMO_13"],
    [14, "MEMO_14"],
)
con.executemany("INSERT INTO tbl VALUES(?, ?)", tpl1)
lst2 = [
    {"id": 15, "memo": "MEMO_15"},
    {"id": 16, "memo": "MEMO_16"},
]
con.executemany("INSERT INTO tbl VALUES(:id, :memo)", lst2)
tpl2 = (
    {"id": 17, "memo": "MEMO_17"},
    {"id": 18, "memo": "MEMO_18"},
)
con.executemany("INSERT INTO tbl VALUES(:id, :memo)", tpl2)

# executescript()
con.executescript("""
    INSERT INTO tbl VALUES(21, 'MEMO_21');
    INSERT INTO tbl VALUES(22, 'MEMO_22');
    """)

con.commit()

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：(1, 'MEMO_01')
# 出力：(2, 'MEMO_02')
# 出力：(3, 'MEMO_03')
# 出力：(11, 'MEMO_11')
# 出力：(12, 'MEMO_12')
# 出力：(13, 'MEMO_13')
# 出力：(14, 'MEMO_14')
# 出力：(15, 'MEMO_15')
# 出力：(16, 'MEMO_16')
# 出力：(17, 'MEMO_17')
# 出力：(18, 'MEMO_18')
# 出力：(21, 'MEMO_21')
# 出力：(22, 'MEMO_22')

con.close()

Connection.interrupt()

戻り値なし

Connection.iterdump()

戻り値SQL testフォーマットでダンプするためのイテレータ

import sqlite3

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE tbl(name, memo)")
con.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.execute("INSERT INTO tbl VALUES('NAME_2', 'MEMO_2')")
with open("./database/dump.sql", "w") as f:
    for dump in con.iterdump():
        print(dump)
        f.write(f"{dump}\n")
# 出力：BEGIN TRANSACTION;
# 出力：CREATE TABLE tbl(name, memo);
# 出力：INSERT INTO "tbl" VALUES('NAME_1','MEMO_1');
# 出力：INSERT INTO "tbl" VALUES('NAME_2','MEMO_2');
# 出力：COMMIT;
con.close()

【dump.sql の出力内容】
BEGIN TRANSACTION;
CREATE TABLE tbl(name, memo);
INSERT INTO "tbl" VALUES('NAME_1','MEMO_1');
INSERT INTO "tbl" VALUES('NAME_2','MEMO_2');
COMMIT;

Connection.set_authorizer(authorizer_callback)

戻り値なし
authorizer_callback下記形式のコールバック (None：コールバック登録解除 3.11)

Connection.set_progress_handler(handler, n)

戻り値なし
handler下記形式のコールバック (None：登録解除)
n呼び出し間隔 (SQLite仮想マシンの命令数)

Connection.set_trace_callback(trace_callback) 3.3

戻り値なし
trace_callback下記形式のコールバック (None：無効)

import sqlite3

def authorizer(p1, p2, p3, p4, p5):
    dic = {
        sqlite3.SQLITE_CREATE_TABLE: "CREATE TABLE",
        sqlite3.SQLITE_INSERT: "INSERT",
        sqlite3.SQLITE_READ: "READ",
        sqlite3.SQLITE_SELECT: "SELECT",
        sqlite3.SQLITE_TRANSACTION: "TRANSACTION",
        sqlite3.SQLITE_UPDATE: "UPDATE",
    }
    action = dic[p1]
    print(f"authorizer({action}, {p2}, {p3}, {p4}, {p5})")
    return sqlite3.SQLITE_OK


def progress():
    print("progress()")
    return 0


def trace(sql):
    print(f"trace('{sql}')")
    return


con = sqlite3.connect(":memory:")
con.set_authorizer(authorizer)
con.set_progress_handler(progress, 1)
con.set_trace_callback(trace)

con.execute("CREATE TABLE tbl(id, name, memo)")
# 出力：authorizer(INSERT, sqlite_master, None, main, None)
# 出力：authorizer(CREATE TABLE, tbl, None, main, None)
# 出力：progress() x 7
# 出力：authorizer(UPDATE, sqlite_master, type, main, None)
# 出力：authorizer(UPDATE, sqlite_master, name, main, None)
# 出力：authorizer(UPDATE, sqlite_master, tbl_name, main, None)
# 出力：authorizer(UPDATE, sqlite_master, rootpage, main, None)
# 出力：authorizer(UPDATE, sqlite_master, sql, main, None)
# 出力：authorizer(READ, sqlite_master, ROWID, main, None)
# 出力：trace('CREATE TABLE tbl(id, name, memo)')
# 出力：progress() x 47

con.execute("INSERT INTO tbl VALUES(1, 'NAME_1', 'MEMO_1')")
# 出力：authorizer(INSERT, tbl, None, main, None)
# 出力：authorizer(TRANSACTION, BEGIN, None, None, None)
# 出力：trace('BEGIN ')
# 出力：progress() x 2
# 出力：trace('INSERT INTO tbl VALUES(1, 'NAME_1', 'MEMO_1')')
# 出力：progress() x11

con.execute("INSERT INTO tbl VALUES(2, 'NAME_2', 'MEMO_2')")
# 出力：authorizer(INSERT, tbl, None, main, None)
# 出力：trace('INSERT INTO tbl VALUES(2, 'NAME_2', 'MEMO_2')')
# 出力：progress() x 11

con.execute("INSERT INTO tbl VALUES(3, 'NAME_3', 'MEMO_3')")
# 出力：authorizer(INSERT, tbl, None, main, None)
# 出力：trace('INSERT INTO tbl VALUES(3, 'NAME_3', 'MEMO_3')')
# 出力：progress() x 11

con.commit()
# 出力：authorizer(TRANSACTION, COMMIT, None, None, None)
# 出力：trace('COMMIT')
# 出力：progress() x 2

for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：authorizer(SELECT, None, None, None, None)
# 出力：authorizer(READ, tbl, id, main, None)
# 出力：authorizer(READ, tbl, name, main, None)
# 出力：authorizer(READ, tbl, memo, main, None)
# 出力：trace('SELECT * FROM tbl')
# 出力：progress() x 14
# 出力：(1, 'NAME_1', 'MEMO_1')
# 出力：progress() x 5
# 出力：(2, 'NAME_2', 'MEMO_2')
# 出力：progress() x 2
# 出力：(3, 'NAME_3', 'MEMO_3')

con.close()

Connection.getlimit(category, /)

戻り値 (int)制限値
以下、位置引数
category (int)制限カテゴリ (下記参照)

ProgrammingError 例外制限カテゴリ不明

Connection.setlimit(category, limit, /)

戻り値 (int)以前の制限値
以下、位置引数
category制限カテゴリ (下記参照)
limit (int)制限値
    ＜0変更なし
    上限値＜上限値設定

ProgrammingError 例外制限カテゴリ不明

import sqlite3

con = sqlite3.connect(":memory:")
print(con.getlimit(sqlite3.SQLITE_LIMIT_LENGTH))
# 出力例：1000000000
print(con.getlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH))
# 出力例：1000000000
print(con.getlimit(sqlite3.SQLITE_LIMIT_COLUMN))
# 出力例：2000
print(con.getlimit(sqlite3.SQLITE_LIMIT_EXPR_DEPTH))
# 出力例：1000
print(con.getlimit(sqlite3.SQLITE_LIMIT_COMPOUND_SELECT))
# 出力例：500
print(con.getlimit(sqlite3.SQLITE_LIMIT_VDBE_OP))
# 出力例：250000000
print(con.getlimit(sqlite3.SQLITE_LIMIT_FUNCTION_ARG))
# 出力例：127
print(con.getlimit(sqlite3.SQLITE_LIMIT_ATTACHED))
# 出力例：10
print(con.getlimit(sqlite3.SQLITE_LIMIT_LIKE_PATTERN_LENGTH))
# 出力例：50000
print(con.getlimit(sqlite3.SQLITE_LIMIT_VARIABLE_NUMBER))
# 出力例：32766
print(con.getlimit(sqlite3.SQLITE_LIMIT_TRIGGER_DEPTH))
# 出力例：1000
print(con.getlimit(sqlite3.SQLITE_LIMIT_WORKER_THREADS))
# 出力例：0

print(con.setlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH, 20))
# 出力例：1000000000
print(con.getlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH))
# 出力例：20
# cur = con.execute("CREATE TABLE tbl(column1, column2, column3)")
# 例外：sqlite3.DataError: query string is too large

print(con.setlimit(sqlite3.SQLITE_LIMIT_ATTACHED, 999))
# 出力例：10
print(con.getlimit(sqlite3.SQLITE_LIMIT_ATTACHED))
# 出力例：10

con.close()

Connection.serialize(*, name='main') 3.11

戻り値 (bytes)
以下、キーワード引数
name (str)データベース名

Connection.deserialize(data, /, *, name='main') 3.11

戻り値
以下、位置引数
data (bytes)データ
以下、キーワード引数
name (str)データベース名

OperationalError 例外読み取りトランザクション・バックアップ操作 関係中
DatabaseError 例外データに有効 SQLite データベースがない
OverflowError 例外データが 263 - 1 より大きい

import sqlite3

con = sqlite3.connect(":memory:")
with con:
    con.execute("CREATE TABLE tbl(col_1, col_2)")
    con.execute("INSERT INTO tbl VALUES(1, 'TEXT_1')")
    con.execute("INSERT INTO tbl VALUES(2, 'TEXT_2')")
    con.execute("INSERT INTO tbl VALUES(3, 'TEXT_3')")
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：
# (1, 'TEXT_1')
# (2, 'TEXT_2')
# (3, 'TEXT_3')

data = con.serialize()
print(data)
# 出力例：
# b'SQLite format 3 (省略)'

with con:
    con.execute("DELETE FROM tbl WHERE col_1 = 2")
    con.execute("INSERT INTO tbl VALUES(99, 'TEXT_99')")
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：
# (1, 'TEXT_1')
# (3, 'TEXT_3')
# (99, 'TEXT_99')

con.deserialize(data)
for row in con.execute("SELECT * FROM tbl"):
    print(row)
# 出力：
# (1, 'TEXT_1')
# (2, 'TEXT_2')
# (3, 'TEXT_3')

con.close()

Connection.setconfig(op, enable=True, /) 3.12

なし
以下、位置引数
op接続構成オプション
enable設定値 (True：有効 / False：無効)

Connection.getconfig(op, /) 3.12

 (bool)設定値
以下、位置引数
op接続構成オプション

Cursor.description

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.execute("CREATE TABLE tbl(column1, column2, column3)")
print(cur.description)
# 出力：None

cur = con.execute("SELECT * FROM tbl")
print(cur.description)
# 出力：(('column1', None, None, None, None, None, None), ('column2', None, None, None, None, None, None), ('column3', None, None, None, None, None, None))

cur = con.execute("SELECT column2 FROM tbl")
print(cur.description)
# 出力：(('column2', None, None, None, None, None, None),)

con.close()

Cursor.lastrowid 

Cursor.rowcount 

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.execute("CREATE TABLE tbl(id PRIMARY KEY, name)")

cur.execute("INSERT INTO tbl VALUES(1, 'NAME_1')")
cur.execute("INSERT INTO tbl VALUES(2, 'NAME_2')")
cur.execute("INSERT INTO tbl VALUES(3, 'NAME_3')")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 3 / cur.rowcount = 1

cur.execute("UPDATE tbl SET name='NAME_02' WHERE name='NAME_2'")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 3 / cur.rowcount = 1

cur.execute("REPLACE INTO tbl VALUES(1, 'NAME_01')")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 4 / cur.rowcount = 1

cur.execute("DELETE FROM tbl WHERE name='NAME_3'")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 4 / cur.rowcount = 1

data = [
    (4, 'NAME_4'),
    (5, 'NAME_5'),
    (6, 'NAME_6'),
]
cur.executemany("INSERT INTO tbl VALUES(?, ?)", data)
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 4 / cur.rowcount = 3

cur.execute("INSERT INTO tbl VALUES(7, 'NAME_7')")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 8 / cur.rowcount = 1

con.commit()

for row in cur.execute("SELECT rowid, * FROM tbl"):
    print(row)
# 出力：(2, 2, 'NAME_02')
# 出力：(4, 1, 'NAME_01')
# 出力：(5, 4, 'NAME_4')
# 出力：(6, 5, 'NAME_5')
# 出力：(7, 6, 'NAME_6')
# 出力：(8, 7, 'NAME_7')
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 8 / cur.rowcount = -1

cur.execute("DELETE FROM  tbl")
print(f"{cur.lastrowid = } / {cur.rowcount = }")
# 出力：cur.lastrowid = 8 / cur.rowcount = 6

con.close()

Cursor.close()

戻り値なし

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE tbl(name, memo)")
cur.execute("INSERT INTO tbl VALUES('NAME_1', 'MEMO_1')")
con.commit()

cur.execute("SELECT * FROM tbl")
row = cur.fetchone()
print(row)
# 出力：('NAME_1', 'MEMO_1')

cur.close()
con.close()

Cursor.execute(sql, parameters=(), /3.10 ) 3.12

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sqlSQL文
parameters (dict | シーケンス)SQL文のパラメータ

Cursor.executemany(sql, parameters, /3.10 ) 3.12

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sqlSQL 文
parametersSQL文のパラメータ (シーケンス・マッピング)

Cursor.executescript(sql_script, /3.10 )

戻り値Cursor【カーソル クラス】オブジェクト
以下、位置引数 3.10
sql_scriptSQLスクリプト

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE tbl(id, memo)")

# execute()
lst = [1, "MEMO_01"]
cur.execute("INSERT INTO tbl VALUES(?, ?)", lst)
tpl = (2, "MEMO_02")
cur.execute("INSERT INTO tbl VALUES(?, ?)", tpl)
dic = {"id":3, "memo": "MEMO_03"}
cur.execute("INSERT INTO tbl VALUES(:id, :memo)", dic)

# executemany()
lst1 = [
    (11, "MEMO_11"),
    (12, "MEMO_12"),
]
cur.executemany("INSERT INTO tbl VALUES(?, ?)", lst1)
tpl1 = (
    [13, "MEMO_13"],
    [14, "MEMO_14"],
)
cur.executemany("INSERT INTO tbl VALUES(?, ?)", tpl1)
lst2 = [
    {"id": 15, "memo": "MEMO_15"},
    {"id": 16, "memo": "MEMO_16"},
]
cur.executemany("INSERT INTO tbl VALUES(:id, :memo)", lst2)
tpl2 = (
    {"id": 17, "memo": "MEMO_17"},
    {"id": 18, "memo": "MEMO_18"},
)
cur.executemany("INSERT INTO tbl VALUES(:id, :memo)", tpl2)

# executescript()
cur.executescript("""
    INSERT INTO tbl VALUES(21, 'MEMO_21');
    INSERT INTO tbl VALUES(22, 'MEMO_22');
    """)

con.commit()

for row in cur.execute("SELECT * FROM tbl"):
    print(row)
# 出力：(1, 'MEMO_01')
# 出力：(2, 'MEMO_02')
# 出力：(3, 'MEMO_03')
# 出力：(11, 'MEMO_11')
# 出力：(12, 'MEMO_12')
# 出力：(13, 'MEMO_13')
# 出力：(14, 'MEMO_14')
# 出力：(15, 'MEMO_15')
# 出力：(16, 'MEMO_16')
# 出力：(17, 'MEMO_17')
# 出力：(18, 'MEMO_18')
# 出力：(21, 'MEMO_21')
# 出力：(22, 'MEMO_22')

con.close()

Cursor.fetchone()

戻り値行情報 (行なし：None)

Cursor.fetchmany(size=cursor.arraysize)

戻り値行情報のリスト (行なし：空リスト)
sizeフェッチ行数

Cursor.fetchall()

戻り値行情報のリスト (行なし：空リスト)

Cursor.arraysize
デフォルト：1

import sqlite3

con = sqlite3.connect(":memory:")
cur = con.cursor()
cur.execute("CREATE TABLE tbl(name, memo)")
for i in range(1, 6):
    name = "NAME_" + str(i)
    memo = "MEMO_" + str(i)
    cur.execute("INSERT INTO tbl VALUES(?,?)", (name, memo))
con.commit()

print("fetchone()")
cur.execute("SELECT * FROM tbl")
row = cur.fetchone()
print(row)
# 出力：('NAME_1', 'MEMO_1')
print(row[0])
# 出力：NAME_1
print(row[1])
# 出力：MEMO_1

print("fetchmany() [1]")
cur.execute("SELECT * FROM tbl")
for row in cur.fetchmany():
    print(row)
# 出力：('NAME_1', 'MEMO_1')

print("fetchmany() [2]")
cur.arraysize = 2
cur.execute("SELECT * FROM tbl")
for row in cur.fetchmany():
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')

print("fetchmany(3)")
cur.execute("SELECT * FROM tbl")
rows = cur.fetchmany(3)
for row in rows:
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')
# 出力：('NAME_3', 'MEMO_3')

print("fetchall()")
cur.execute("SELECT * FROM tbl")
rows = cur.fetchall()
for row in rows:
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')
# 出力：('NAME_3', 'MEMO_3')
# 出力：('NAME_4', 'MEMO_4')
# 出力：('NAME_5', 'MEMO_5')

print("Cursor【カーソル クラス】のイテレーション")
for row in cur.execute("SELECT * FROM tbl"):
    print(row)
# 出力：('NAME_1', 'MEMO_1')
# 出力：('NAME_2', 'MEMO_2')
# 出力：('NAME_3', 'MEMO_3')
# 出力：('NAME_4', 'MEMO_4')
# 出力：('NAME_5', 'MEMO_5')

con.close()

import sqlite3

con = sqlite3.connect(":memory:")
con.row_factory = sqlite3.Row
cur = con.cursor()
cur.execute("CREATE TABLE tbl(name, memo)")
for i in range(1, 6):
    name = "NAME_" + str(i)
    memo = "MEMO_" + str(i)
    cur.execute("INSERT INTO tbl VALUES(?,?)", (name, memo))
con.commit()

print("fetchone()")
cur.execute("SELECT * FROM tbl")
row = cur.fetchone()
print(row[0], row[1])
# 出力：NAME_1 MEMO_1
print(row['name'], row['memo'])
# 出力：NAME_1 MEMO_1

print("fetchmany(3)")
cur.execute("SELECT * FROM tbl")
rows = cur.fetchmany(3)
for row in rows:
    print(row['name'], row['memo'])
# 出力：NAME_1 MEMO_1
# 出力：NAME_2 MEMO_2
# 出力：NAME_3 MEMO_3

print("fetchall()")
cur.execute("SELECT * FROM tbl")
rows = cur.fetchall()
for row in rows:
    print(row['name'], row['memo'])
# 出力：NAME_1 MEMO_1
# 出力：NAME_2 MEMO_2
# 出力：NAME_3 MEMO_3
# 出力：NAME_4 MEMO_4
# 出力：NAME_5 MEMO_5

print("Cursor【カーソル クラス】のイテレーション")
for row in cur.execute("SELECT * FROM tbl"):
    print(row['name'], row['memo'])
# 出力：NAME_1 MEMO_1
# 出力：NAME_2 MEMO_2
# 出力：NAME_3 MEMO_3
# 出力：NAME_4 MEMO_4
# 出力：NAME_5 MEMO_5

con.close()

Row.keys()

戻り値列名のリスト

import sqlite3

con = sqlite3.connect(":memory:")
con.row_factory = sqlite3.Row
cur = con.cursor()
cur.execute("CREATE TABLE tbl(id, name, memo)")
cur.execute("INSERT INTO tbl VALUES(1, 'NAME_1', 'MEMO_1')")
con.commit()

cur.execute("SELECT * FROM tbl")
row = cur.fetchone()
print(row.keys())
# 出力：['id', 'name', 'memo']

# 要素アクセス (インデックス・列名)
print(row['id'], row[1], row['MEMO'])
# 出力：1 NAME_1 MEMO_1

# イテレーション
for item in row:
    print(item)
# 出力：1
# 出力：NAME_1
# 出力：MEMO_1

# 同値テスト
cur.execute("SELECT id, name, memo FROM tbl")
row2 = cur.fetchone()
print(row)
# 出力：<sqlite3.Row object at 0x～>
print(row2)
# 出力：<sqlite3.Row object at 0x～> 上記と別アドレス
print(row == row2)
# 出力：True

# len()
print(len(row))
# 出力：3

# スライス
print(row[0:])
# 出力：(1, 'NAME_1', 'MEMO_1')
print(row[1:3])
# 出力：('NAME_1', 'MEMO_1')
print(row[3:0:-1])
# 出力：('MEMO_1', 'NAME_1')

con.close()

Blob.close()

Blob.read(length=-1, /)

戻り値読み取りデータ (EOFまで)
length読み取りバイト数 (省略：末尾まで)

Blob.write(data, /)

data書き込みデータ
ValueError 例外末尾を超えて書き込み (サイズ変更不可)

Blob.seek(offset, origin=os.SEEK_SET, /)

offsetオフセット
origin基準
    os.SEEK_SET：絶対位置
    os.SEEK_CUR：現在位置を基準
    os.SEEK_END：末尾を基準

Blob.tell()

戻り値現在のアクセス位置

import sqlite3

# データベース作成
con = sqlite3.connect(":memory:")
with con:
    con.execute("CREATE TABLE tbl(name text, blob_data blob)")
    con.execute("INSERT INTO tbl VALUES('NAME_1', zeroblob(10))")
    con.execute("INSERT INTO tbl VALUES('NAME_2', zeroblob(20))")
    con.execute("INSERT INTO tbl VALUES('NAME_3', zeroblob(5))")

# １行目
blob = con.blobopen('tbl', 'blob_data', 1)
blob.write(b'\x11\x12\x13\x14\x15\x16')
print(blob.tell())
# 出力：6
blob.seek(0)
data = blob.read(3)
print(data)
# 出力：b'\x11\x12\x13'
print(blob.tell())
# 出力：3
data = blob.read(3)
print(data)
# 出力：b'\x14\x15\x16'
print(blob.tell())
# 出力：6
data = blob.read(100)
print(data)
# 出力：b'\x00\x00\x00\x00'
print(blob.tell())
# 出力：10
blob.seek(-3, os.SEEK_END)
print(blob.tell())
# 出力：7
blob.close()

# ２行目
with con.blobopen('tbl', 'blob_data', 2) as blob:
    print(blob.tell())
    # 出力：0
    blob.write(b'abcdefgh')
    print(blob.tell())
    # 出力：8
    blob.write(b'12345678')
    print(blob.tell())
    # 出力：16
    print(f'{blob[0]:#x} {chr(blob[0])}')
    # 出力：0x61 a
    print(f'{blob[1]:#x} {chr(blob[1])}')
    # 出力：0x62 b
    print(f'{blob[2]:#x} {chr(blob[2])}')
    # 出力：0x63 c
    blob[0] = ord('A')
    blob[1] = ord('B')
    blob[2] = ord('C')
    blob[-3] = ord('X')
    blob[-2] = ord('Y')
    blob[-1] = ord('Z')

# １～３行目
for row in range(1, 4):
    with con.blobopen('tbl', 'blob_data', row) as blob:
        data = blob.read()
        print(len(blob), data)
# 出力：10 b'\x11\x12\x13\x14\x15\x16\x00\x00\x00\x00'
# 出力：20 b'ABCdefgh12345678\x00XYZ'
# 出力：5 b'\x00\x00\x00\x00\x00'