enable-strict-quoting-sqlite3.py

import sqlite3
import ctypes
from ctypes.util import find_library

# https://www.sqlite.org/c3ref/c_dbconfig_defensive.html
SQLITE_DBCONFIG_DQS_DML = 1013
SQLITE_DBCONFIG_DQS_DDL = 1014

class PyObject_HEAD(ctypes.Structure):
    _fields_ = (
        ('ob_refcnt', ctypes.c_ssize_t),
        ('ob_type', ctypes.c_void_p),
    )

class Sqlite3Connection(ctypes.Structure):
    # https://github.com/python/cpython/blob/a87c46eab3c306b1c5b8a072b7b30ac2c50651c0/Modules/_sqlite/connection.h#L48-L107
    _fields_ = (
        ('ob_base', PyObject_HEAD),
        ('db', ctypes.c_void_p),
    )

def enable_strict_quoting(conn):
    lib = find_library('sqlite3')
    so = ctypes.cdll.LoadLibrary(lib)
    # https://www.sqlite.org/c3ref/db_config.html
    sqlite3_db_config = getattr(so, 'sqlite3_db_config')
    sqlite3_db_config.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int]
    
    # Go spelunking in the connection to find it's internal sqlite3 db handle;
    # this is very unsafe, but ought to work consistently on a given Python build.
    unsafe = Sqlite3Connection.from_address(id(conn))

    sqlite3_db_config(unsafe.db, SQLITE_DBCONFIG_DQS_DDL, 0, 0);
    sqlite3_db_config(unsafe.db, SQLITE_DBCONFIG_DQS_DML, 0, 0);

def monkey_patch():
    from datasette import database

    original_connect = database.Database.connect

    def patched_connect(self, write=False):
        conn = original_connect(self, write)
        enable_strict_quoting(conn)
        return conn

    database.Database.connect = patched_connect

def test_enable_strict_quoting():
    conn = sqlite3.connect(':memory:')
    print(conn.execute('select sqlite_version()').fetchall())
    enable_strict_quoting(conn)

    print(conn.execute('SELECT "foo"').fetchall())
    # If all goes well, we'll raise
    # sqlite3.OperationalError: no such column: foo

if __name__ == '__main__':
    test_enable_strict_quoting()
else:
    monkey_patch()