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

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)

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

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

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

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

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

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

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

import sqlite3

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

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

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

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

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

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

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;

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

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

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

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

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

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

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

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

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

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'