Replace sqlite3 with sqlean

CHANGELOG.md

@@ -2,11 +2,11 @@
 
 ### Features
 
-*
+* Replace sqlite3 library with [sqlean](https://antonz.org/sqlean/). It's a drop-in replacement for sqlite3.
 
 ### Bug Fixes
 
-*
+* Fix missing sqlite extensions using sqlean. Note. support only limited set of extensions. [(#119)](https://github.com/dbcli/litecli/issues/119)
 
 
 ## 1.15.0 - 2025-03-15

litecli/main.py

@@ -11,7 +11,7 @@
 from collections import namedtuple
 from datetime import datetime
 from io import open
-from sqlite3 import OperationalError, sqlite_version
+from sqlean import OperationalError, sqlite_version
 from time import time
 
 import click
@@ -260,7 +260,7 @@ def initialize_logging(self):
             )
             return
 
-        formatter = logging.Formatter("%(asctime)s (%(process)d/%(threadName)s) " "%(name)s %(levelname)s - %(message)s")
+        formatter = logging.Formatter("%(asctime)s (%(process)d/%(threadName)s) %(name)s %(levelname)s - %(message)s")
 
         handler.setFormatter(formatter)
 
@@ -361,7 +361,7 @@ def run_cli(self):
         else:
             history = None
             self.echo(
-                'Error: Unable to open the history file "{}". ' "Your query history will not be saved.".format(history_file),
+                'Error: Unable to open the history file "{}". Your query history will not be saved.'.format(history_file),
                 err=True,
                 fg="red",
             )
@@ -452,7 +452,7 @@ def one_iteration(text=None):
                         if context:
                             click.echo("LLM Reponse:")
                             click.echo(context)
-                            click.echo('---')
+                            click.echo("---")
                         click.echo(f"Time: {duration:.2f} seconds")
                         text = self.prompt_app.prompt(default=sql)
                     except KeyboardInterrupt:
@@ -927,7 +927,7 @@ def cli(
 
     litecli.connect(database)
 
-    litecli.logger.debug("Launch Params: \n" "\tdatabase: %r", database)
+    litecli.logger.debug("Launch Params: \n\tdatabase: %r", database)
 
     #  --execute argument
     if execute:

litecli/packages/prompt_utils.py

@@ -16,7 +16,7 @@ def confirm_destructive_query(queries):
     * False if the query is destructive and the user doesn't want to proceed.
 
     """
-    prompt_text = "You're about to run a destructive command.\n" "Do you want to proceed? (y/n)"
+    prompt_text = "You're about to run a destructive command.\nDo you want to proceed? (y/n)"
     if is_destructive(queries) and sys.stdin.isatty():
         return prompt(prompt_text, type=bool)
 

litecli/packages/special/dbcommands.py

@@ -124,6 +124,7 @@ def show_schema(cur, arg=None, **_):
 
     return [(None, tables, headers, status)]
 
+
 @special_command(
     ".databases",
     ".databases",
@@ -142,6 +143,7 @@ def list_databases(cur, **_):
     else:
         return [(None, None, None, "")]
 
+
 @special_command(
     ".indexes",
     ".indexes [tablename]",

litecli/packages/special/llm.py

@@ -194,6 +194,7 @@ def ensure_litecli_template(replace=False):
     run_external_cmd("llm", PROMPT, "--save", "litecli")
     return
 
+
 @export
 def handle_llm(text, cur) -> Tuple[str, Optional[str], float]:
     """This function handles the special command `\\llm`.

litecli/sqlexecute.py

@@ -1,14 +1,16 @@
 import logging
-import sqlite3
+import sqlean as sqlite3
+
 from contextlib import closing
-from sqlite3 import OperationalError
+from sqlean import OperationalError
 from litecli.packages.special.utils import check_if_sqlitedotcommand
 
 import sqlparse
 import os.path
 
 from .packages import special
 
+sqlite3.extensions.enable_all()
 _logger = logging.getLogger(__name__)
 
 # FIELD_TYPES = decoders.copy()
@@ -58,7 +60,7 @@ def __init__(self, database):
 
     def connect(self, database=None):
         db = database or self.dbname
-        _logger.debug("Connection DB Params: \n" "\tdatabase: %r", db)
+        _logger.debug("Connection DB Params: \n\tdatabase: %r", db)
 
         db_name = os.path.expanduser(db)
         db_dir_name = os.path.dirname(os.path.abspath(db_name))

pyproject.toml

@@ -14,8 +14,9 @@ dependencies = [
     "prompt-toolkit>=3.0.3,<4.0.0",
     "pygments>=1.6",
     "sqlparse>=0.4.4",
-    "setuptools",                   # Required by llm commands to install models
+    "setuptools", # Required by llm commands to install models
     "pip",
+    "sqlean-py>=3.47.0",
 ]
 
 [build-system]

release.py

@@ -100,7 +100,7 @@ def checklist(questions):
         action="store_true",
         dest="confirm_steps",
         default=False,
-        help=("Confirm every step. If the step is not " "confirmed, it will be skipped."),
+        help=("Confirm every step. If the step is not confirmed, it will be skipped."),
     )
     parser.add_option(
         "-d",

tests/test_dbspecial.py

@@ -96,7 +96,8 @@ def test_special_d(executor):
     run(executor, """create table tst_tbl1(a text)""")
     results = run(executor, """\\d""")
 
-    assert_result_equal(results, headers=["name"], rows=[("tst_tbl1",)], status="")
+    # 'sqlean_define' is a table created by sqlean. Details: https://github.com/nalgeon/sqlean/blob/main/docs/define.md?plain=1#L3
+    assert_result_equal(results, headers=["name"], rows=[("sqlean_define",), ("tst_tbl1",)], status="")
 
 
 @dbtest

tests/test_llm_special.py

@@ -54,7 +54,7 @@ def test_llm_command_with_c_flag(mock_run_cmd, mock_llm, executor):
 @patch("litecli.packages.special.llm.run_external_cmd")
 def test_llm_command_with_c_flag_and_fenced_sql(mock_run_cmd, mock_llm, executor):
     # The luscious SQL is inside triple backticks
-    return_text = "Here is your query:\n" "```sql\nSELECT * FROM table;\n```"
+    return_text = "Here is your query:\n```sql\nSELECT * FROM table;\n```"
     mock_run_cmd.return_value = (0, return_text)
 
     test_text = r"\llm -c 'Rewrite the SQL without CTE'"
@@ -88,6 +88,7 @@ def test_llm_command_known_subcommand(mock_run_cmd, mock_llm, executor):
     # And the function should raise FinishIteration(None)
     assert exc_info.value.args[0] is None
 
+
 @patch("litecli.packages.special.llm.llm")
 @patch("litecli.packages.special.llm.run_external_cmd")
 def test_llm_command_with_help_flag(mock_run_cmd, mock_llm, executor):
@@ -106,6 +107,7 @@ def test_llm_command_with_help_flag(mock_run_cmd, mock_llm, executor):
     # And the function should raise FinishIteration(None)
     assert exc_info.value.args[0] is None
 
+
 @patch("litecli.packages.special.llm.llm")
 @patch("litecli.packages.special.llm.run_external_cmd")
 def test_llm_command_with_install_flag(mock_run_cmd, mock_llm, executor):

tests/test_main.py

@@ -74,7 +74,7 @@ def test_batch_mode(executor):
     run(executor, """create table test(a text)""")
     run(executor, """insert into test values('abc'), ('def'), ('ghi')""")
 
-    sql = "select count(*) from test;\n" "select * from test limit 1;"
+    sql = "select count(*) from test;\nselect * from test limit 1;"
 
     runner = CliRunner()
     result = runner.invoke(cli, args=CLI_ARGS, input=sql)
@@ -88,7 +88,7 @@ def test_batch_mode_table(executor):
     run(executor, """create table test(a text)""")
     run(executor, """insert into test values('abc'), ('def'), ('ghi')""")
 
-    sql = "select count(*) from test;\n" "select * from test limit 1;"
+    sql = "select count(*) from test;\nselect * from test limit 1;"
 
     runner = CliRunner()
     result = runner.invoke(cli, args=CLI_ARGS + ["-t"], input=sql)

tests/test_parseutils.py

@@ -120,12 +120,12 @@ def test_query_starts_with_comment():
 
 
 def test_queries_start_with():
-    sql = "# comment\n" "show databases;" "use foo;"
+    sql = "# comment\nshow databases;use foo;"
     assert queries_start_with(sql, ("show", "select")) is True
     assert queries_start_with(sql, ("use", "drop")) is True
     assert queries_start_with(sql, ("delete", "update")) is False
 
 
 def test_is_destructive():
-    sql = "use test;\n" "show databases;\n" "drop database foo;"
+    sql = "use test;\nshow databases;\ndrop database foo;"
     assert is_destructive(sql) is True

tests/test_smart_completion_public_schema_only.py

@@ -41,17 +41,18 @@ def complete_event():
 
 
 def test_escape_name(completer):
-
-    for name, expected_name in [# Upper case name shouldn't be escaped
+    for name, expected_name in [  # Upper case name shouldn't be escaped
         ("BAR", "BAR"),
         # This name is escaped and should start with back tick
         ("2025todos", "`2025todos`"),
         # normal case
         ("people", "people"),
         # table name with _underscore should not be escaped
-        ("django_users", "django_users")]:
+        ("django_users", "django_users"),
+    ]:
         assert completer.escape_name(name) == expected_name
 
+
 def test_empty_string_completion(completer, complete_event):
     text = ""
     position = 0

tests/test_sqlexecute.py

@@ -5,7 +5,7 @@
 import pytest
 
 from utils import run, dbtest, set_expanded_output, is_expanded_output, assert_result_equal
-from sqlite3 import OperationalError, ProgrammingError
+from sqlean import OperationalError, ProgrammingError
 
 
 @dbtest
@@ -44,9 +44,13 @@ def test_table_and_columns_query(executor):
     run(executor, "create table b(z text)")
     run(executor, "create table t(t text)")
 
-    assert set(executor.tables()) == set([("a",), ("b",), ("t",)])
-    assert set(executor.table_columns()) == set([("a", "x"), ("a", "y"), ("b", "z"), ("t", "t")])
-    assert set(executor.table_columns()) == set([("a", "x"), ("a", "y"), ("b", "z"), ("t", "t")])
+    assert set(executor.tables()) == set([("sqlean_define",), ("a",), ("b",), ("t",)])
+    assert set(executor.table_columns()) == set(
+        [("sqlean_define", "type"), ("sqlean_define", "body"), ("sqlean_define", "name"), ("a", "x"), ("a", "y"), ("b", "z"), ("t", "t")]
+    )
+    assert set(executor.table_columns()) == set(
+        [("sqlean_define", "type"), ("sqlean_define", "body"), ("sqlean_define", "name"), ("a", "x"), ("a", "y"), ("b", "z"), ("t", "t")]
+    )
 
 
 @dbtest
@@ -252,7 +256,7 @@ def test_favorite_query_multiple_statement(executor):
 
     results = run(
         executor,
-        "\\fs test-ad select * from test where a like 'a%'; " "select * from test where a like 'd%'",
+        "\\fs test-ad select * from test where a like 'a%'; select * from test where a like 'd%'",
     )
     assert_result_equal(results, status="Saved.")
 

tests/utils.py

@@ -7,7 +7,7 @@
 import multiprocessing
 from contextlib import closing
 
-import sqlite3
+import sqlean as sqlite3
 import pytest
 
 from litecli.main import special