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

Preserve untracked SQL functions and describe deregistration behavior

Browse source
This commit is contained in:
goingforstudying-ctrl 2026-10-14 14:43:08 -04:00
commit d6c49ab0c7
3 changed files with 63 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/python-api.rst
View file

@ -3347,13 +3347,13 @@ If you want to deliberately replace the registered function with a new implement
def reverse_string(s):
return s[::-1]
To remove a registered function, pass the callable to ``db.register_function()`` with ``deregister=True``:
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 removes the SQL function with the callable's name and number of arguments, and clears its registration from the ``Database`` instance. You can register it again without ``replace=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
@ -3365,7 +3365,9 @@ If the function was registered with ``name=``, use that same SQL name when remov
db.register_function(reverse_string, name="rev", deregister=True)
Removing a function that is not registered has no effect. Deregistration does not restore any earlier function that was replaced. SQLite raises an ``OperationalError`` if you try to invoke the removed function; the error message can differ between SQLite implementations.
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::

20
sqlite_utils/db.py
View file

@ -818,27 +818,33 @@ class Database:
: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 remove the function with this name and number of arguments
: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:
self.conn.create_function(fn_name, arity, None)
# SQLite function names are case-insensitive for ASCII letters.
# The existing registration cache uses exact-case names.
fold_ascii = str.maketrans(
"ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
)
self._registered_functions.difference_update(
registrations = {
(registered_name, registered_arity)
for registered_name, registered_arity in tuple(
self._registered_functions
)
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

60
tests/test_register_function.py
View file

@ -5,6 +5,22 @@ from unittest.mock import MagicMock, call
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):
@ -107,7 +123,7 @@ def test_register_function_replace(fresh_db):
@pytest.mark.parametrize("deterministic", (False, True))
def test_deregister_function(fresh_db, deterministic):
def test_deregister_function(fresh_db, deterministic, monkeypatch):
calls = []
def temporary(value):
@ -117,8 +133,7 @@ def test_deregister_function(fresh_db, deterministic):
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
with pytest.raises(sqlite3.OperationalError):
fresh_db.execute("select temporary(4)").fetchone()
assert_deregistered(fresh_db, "select temporary(4)", monkeypatch)
assert calls == [3]
fresh_db.register_function(temporary)
@ -127,21 +142,20 @@ def test_deregister_function(fresh_db, deterministic):
@pytest.mark.parametrize("removal_name", ("custom_name", "CUSTOM_NAME"))
def test_deregister_function_custom_name(fresh_db, removal_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)
with pytest.raises(sqlite3.OperationalError):
fresh_db.execute("select custom_name(3)").fetchone()
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):
def test_deregister_function_preserves_other_arity(fresh_db, monkeypatch):
def one_arg(value):
return value + 1
@ -151,20 +165,18 @@ def test_deregister_function_preserves_other_arity(fresh_db):
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)
with pytest.raises(sqlite3.OperationalError):
fresh_db.execute("select temporary(1)").fetchone()
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):
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)
with pytest.raises(sqlite3.OperationalError):
fresh_db.execute('select "FÜ"(1)').fetchone()
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ü")
@ -184,6 +196,25 @@ def test_deregister_function_idempotent(fresh_db):
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
@ -202,13 +233,12 @@ def test_deregister_function_failure_preserves_cache(fresh_db):
assert fresh_db.execute("select temporary(3)").fetchone()[0] == 6
def test_deregister_function_decorator(fresh_db):
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
with pytest.raises(sqlite3.OperationalError):
fresh_db.execute("select temporary()").fetchone()
assert_deregistered(fresh_db, "select temporary()", monkeypatch)
fresh_db.register_function()(temporary)
assert fresh_db.execute("select temporary()").fetchone()[0] == 1