remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-24 01:44:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

Split out _iter_complete_sql_statements

Browse source 
To make it clearer why we are avoiding executescript()
This commit is contained in:
Simon Willison 2026-06-21 10:18:25 -07:00
commit a65df4cc69
2 changed files with 57 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

26
sqlite_utils/db.py
View file

@ -196,6 +196,19 @@ DEFAULT = Default()
Tracer = Callable[[str, Optional[Union[Sequence[Any], Dict[str, Any]]]], None]
def _iter_complete_sql_statements(sql: str) -> Generator[str, None, None]:
statement = []
for char in sql:
statement.append(char)
statement_sql = "".join(statement).strip()
if statement_sql and sqlite3.complete_statement(statement_sql):
yield statement_sql
statement = []
statement_sql = "".join(statement).strip()
if statement_sql:
yield statement_sql
COLUMN_TYPE_MAPPING: Dict[Any, str] = {
float: "REAL",
int: "INTEGER",
@ -618,16 +631,9 @@ class Database:
def _executescript(self, sql: str) -> sqlite3.Cursor:
if self.conn.in_transaction:
cursor = self.conn.cursor()
statement = []
for char in sql:
statement.append(char)
statement_sql = "".join(statement).strip()
if statement_sql and sqlite3.complete_statement(statement_sql):
cursor.execute(statement_sql)
statement = []
statement_sql = "".join(statement).strip()
if statement_sql:
cursor.execute(statement_sql)
# avoid sqlite3.executescript()'s implicit commit:
for statement in _iter_complete_sql_statements(sql):
cursor.execute(statement)
return cursor
return self.conn.executescript(sql)

43
tests/test_atomic.py
View file

@ -1,8 +1,46 @@
import pytest
from sqlite_utils.db import _iter_complete_sql_statements
from sqlite_utils.utils import sqlite3
@pytest.mark.parametrize(
"sql,expected",
(
(
"CREATE TABLE t(id); INSERT INTO t VALUES (1)",
["CREATE TABLE t(id);", "INSERT INTO t VALUES (1)"],
),
(
"INSERT INTO t VALUES ('a;b');",
["INSERT INTO t VALUES ('a;b');"],
),
(
"-- comment;\nCREATE TABLE t(id);",
["-- comment;\nCREATE TABLE t(id);"],
),
(
"""
CREATE TRIGGER t_ai AFTER INSERT ON t
BEGIN
UPDATE t SET value = 'a;b' WHERE id = new.id;
INSERT INTO log VALUES ('x;y');
END;
""",
[
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;"
],
),
),
)
def test_iter_complete_sql_statements(sql, expected):
assert list(_iter_complete_sql_statements(sql)) == expected
def test_atomic_commits(fresh_db):
with fresh_db.atomic():
fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
@ -54,9 +92,10 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
BEGIN
UPDATE dogs SET name = upper(new.name) WHERE id = new.id;
UPDATE dogs SET name = upper(new.name) || '; updated' WHERE id = new.id;
END;
INSERT INTO dogs VALUES (1, 'Cleo');
-- This comment has a semicolon;
INSERT INTO dogs VALUES (1, 'Cleo; the first');
""")
raise RuntimeError("boom")