Support explicit SQL function deregistration

docs/python-api.rst

@@ -3347,6 +3347,28 @@ If you want to deliberately replace the registered function with a new implement
     def reverse_string(s):
         return s[::-1]
 
+To deregister a callable registered through this ``Database`` instance, pass it to ``db.register_function()`` with ``deregister=True``:
+
+.. code-block:: python
+
+    db.register_function(reverse_string, deregister=True)
+
+This stops SQL from invoking the original Python callable for that name and number of arguments, and clears the tracked registration from the ``Database`` instance. You can register it again without ``replace=True``:
+
+.. code-block:: python
+
+    db.register_function(reverse_string)
+
+If the function was registered with ``name=``, use that same SQL name when removing it. SQL function names are case-insensitive for ASCII letters. Other names and overloads with a different number of arguments are left in place:
+
+.. code-block:: python
+
+    db.register_function(reverse_string, name="rev", deregister=True)
+
+Deregistering a name and number of arguments that this ``Database`` instance does not track has no effect. Built-in SQL functions and functions registered directly on the connection are left alone. Deregistration does not restore any earlier function that was replaced. The ``deterministic`` and ``replace`` options are ignored when ``deregister=True``.
+
+Deregistration uses the connection's ``create_function(name, arity, None)`` API. Some Python ``sqlite3`` bindings retain a ``None`` callback placeholder instead of deleting the SQL function entry. Invoking that placeholder raises ``OperationalError: user-defined function raised exception`` and can report an unraisable ``TypeError`` if callback tracebacks are enabled. Other bindings may report ``no such function``. The original Python callable is no longer invoked in either case, but the SQL name may still appear in ``pragma_function_list``; this API does not promise to remove that catalog entry.
+
 Exceptions that occur inside a user-defined function default to returning the following error::
 
     Unexpected error: user-defined function raised exception

sqlite_utils/db.py

@@ -796,6 +796,7 @@ def register_function(
         deterministic: bool = False,
         replace: bool = False,
         name: str | None = None,
+        deregister: bool = False,
     ) -> Callable[[Callable], Callable] | None:
         """
         ``fn`` will be made available as a function within SQL, with the same name and number
@@ -817,11 +818,34 @@ def upper(value):
         :param deterministic: set ``True`` for functions that always returns the same output for a given input
         :param replace: set ``True`` to replace an existing function with the same name - otherwise throw an error
         :param name: name of the SQLite function - if not specified, the Python function name will be used
+        :param deregister: set ``True`` to deregister a callable tracked by this instance with this name and number of arguments.
+          ``deterministic`` and ``replace`` are ignored. Python's sqlite3 binding may retain an error-raising SQL placeholder
+          after the original callback is cleared.
         """
 
         def register(fn: Callable) -> Callable:
             fn_name = name or fn.__name__  # type: ignore
             arity = len(inspect.signature(fn).parameters)
+            if deregister:
+                # SQLite function names are case-insensitive for ASCII letters.
+                # The existing registration cache uses exact-case names.
+                fold_ascii = str.maketrans(
+                    "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
+                )
+                registrations = {
+                    (registered_name, registered_arity)
+                    for registered_name, registered_arity in self._registered_functions
+                    if registered_arity == arity
+                    and registered_name.translate(fold_ascii)
+                    == fn_name.translate(fold_ascii)
+                }
+                if not registrations:
+                    # Do not shadow built-ins or functions registered directly
+                    # on the connection with the driver's None placeholder.
+                    return fn
+                self.conn.create_function(fn_name, arity, None)
+                self._registered_functions.difference_update(registrations)
+                return fn
             if not replace and (fn_name, arity) in self._registered_functions:
                 return fn
             kwargs: dict[str, bool] = {}

tests/test_register_function.py

@@ -5,6 +5,22 @@
 from sqlite_utils.utils import sqlite3
 
 
+def assert_deregistered(db, sql, monkeypatch):
+    # Some sqlite3 bindings keep a None callback placeholder. Capture and
+    # validate its expected unraisable error without leaking a pytest warning.
+    errors = []
+    with monkeypatch.context() as context:
+        context.setattr(sys, "unraisablehook", errors.append)
+        with pytest.raises(
+            sqlite3.OperationalError,
+            match="no such function|user-defined function raised exception",
+        ):
+            db.execute(sql).fetchone()
+    for error in errors:
+        assert error.exc_type is TypeError
+        assert str(error.exc_value) == "'NoneType' object is not callable"
+
+
 def test_register_function(fresh_db):
     @fresh_db.register_function
     def reverse_string(s):
@@ -104,3 +120,125 @@ def one():  # pyright: ignore[reportRedeclaration]
         return "two"
 
     assert "two" == fresh_db.execute("select one()").fetchone()[0]
+
+
+@pytest.mark.parametrize("deterministic", (False, True))
+def test_deregister_function(fresh_db, deterministic, monkeypatch):
+    calls = []
+
+    def temporary(value):
+        calls.append(value)
+        return value * 2
+
+    fresh_db.register_function(temporary, deterministic=deterministic)
+    assert fresh_db.execute("select temporary(3)").fetchone()[0] == 6
+    assert fresh_db.register_function(temporary, deregister=True) is None
+    assert_deregistered(fresh_db, "select temporary(4)", monkeypatch)
+    assert calls == [3]
+
+    fresh_db.register_function(temporary)
+    assert fresh_db.execute("select temporary(5)").fetchone()[0] == 10
+    assert calls == [3, 5]
+
+
+@pytest.mark.parametrize("removal_name", ("custom_name", "CUSTOM_NAME"))
+def test_deregister_function_custom_name(fresh_db, removal_name, monkeypatch):
+    def temporary(value):
+        return value * 2
+
+    fresh_db.register_function(temporary)
+    fresh_db.register_function(temporary, name="custom_name")
+    fresh_db.register_function(temporary, name=removal_name, deregister=True)
+    assert_deregistered(fresh_db, "select custom_name(3)", monkeypatch)
+    assert fresh_db.execute("select temporary(3)").fetchone()[0] == 6
+    fresh_db.register_function(temporary, name="custom_name")
+    assert fresh_db.execute("select custom_name(4)").fetchone()[0] == 8
+
+
+def test_deregister_function_preserves_other_arity(fresh_db, monkeypatch):
+    def one_arg(value):
+        return value + 1
+
+    def two_args(left, right):
+        return left + right
+
+    fresh_db.register_function(one_arg, name="temporary")
+    fresh_db.register_function(two_args, name="temporary")
+    fresh_db.register_function(one_arg, name="temporary", deregister=True)
+    assert_deregistered(fresh_db, "select temporary(1)", monkeypatch)
+    assert fresh_db.execute("select temporary(1, 2)").fetchone()[0] == 3
+
+
+def test_deregister_function_preserves_non_ascii_names(fresh_db, monkeypatch):
+    def temporary(value):
+        return value * 2
+
+    fresh_db.register_function(temporary, name="FÜ")
+    fresh_db.register_function(temporary, name="fü")
+    fresh_db.register_function(temporary, name="fÜ", deregister=True)
+    assert_deregistered(fresh_db, 'select "FÜ"(1)', monkeypatch)
+    assert fresh_db.execute('select "fü"(2)').fetchone()[0] == 4
+    # A surviving registration must still suppress replacement by default.
+    fresh_db.register_function(lambda value: 0, name="fü")
+    assert fresh_db.execute('select "fü"(2)').fetchone()[0] == 4
+
+
+def test_deregister_function_idempotent(fresh_db):
+    def temporary():
+        return 1
+
+    # Removing an unregistered function or removing it twice is harmless.
+    fresh_db.register_function(temporary, deregister=True)
+    fresh_db.register_function(temporary)
+    fresh_db.register_function(temporary, deregister=True)
+    fresh_db.register_function(temporary, deregister=True)
+    fresh_db.register_function(temporary)
+    assert fresh_db.execute("select temporary()").fetchone()[0] == 1
+
+
+@pytest.mark.parametrize("name", ("upper", "UPPER"))
+def test_deregister_unregistered_preserves_builtin(fresh_db, name):
+    def upper(value):
+        return value
+
+    assert fresh_db.execute("select upper('hello')").fetchone()[0] == "HELLO"
+    fresh_db.register_function(upper, name=name, deregister=True)
+    assert fresh_db.execute("select upper('hello')").fetchone()[0] == "HELLO"
+
+
+def test_deregister_untracked_preserves_connection_function(fresh_db):
+    def temporary(value):
+        return value * 2
+
+    fresh_db.conn.create_function("temporary", 1, temporary)
+    fresh_db.register_function(temporary, deregister=True)
+    assert fresh_db.execute("select temporary(3)").fetchone()[0] == 6
+
+
+def test_deregister_function_failure_preserves_cache(fresh_db):
+    def temporary(value):
+        return value * 2
+
+    fresh_db.register_function(temporary)
+    original_conn = fresh_db.conn
+    fresh_db.conn = MagicMock()
+    fresh_db.conn.create_function.side_effect = sqlite3.OperationalError("busy")
+    try:
+        with pytest.raises(sqlite3.OperationalError, match="busy"):
+            fresh_db.register_function(temporary, deregister=True)
+        fresh_db.conn.create_function.assert_called_once_with("temporary", 1, None)
+    finally:
+        fresh_db.conn = original_conn
+    fresh_db.register_function(lambda value: 0, name="temporary")
+    assert fresh_db.execute("select temporary(3)").fetchone()[0] == 6
+
+
+def test_deregister_function_decorator(fresh_db, monkeypatch):
+    def temporary():
+        return 1
+
+    fresh_db.register_function(temporary)
+    assert fresh_db.register_function(deregister=True)(temporary) is temporary
+    assert_deregistered(fresh_db, "select temporary()", monkeypatch)
+    fresh_db.register_function()(temporary)
+    assert fresh_db.execute("select temporary()").fetchone()[0] == 1