如何通过Python API为SQLite开启EXPLAIN PLAN并支持切换

Problem Description #

I'm using Python (with Peewee) to connect to an SQLite database, and my data access layer mixes Peewee ORM with raw SQL functions. I want to enable EXPLAIN PLAN for all queries when connecting to the database, and be able to toggle this feature via a configuration setting or CLI argument.

Here's my current code snippet:

from playhouse.db_url import connect
self._logger.info("opening db connection to database, creating cursor and initializing orm model ...")
self.__db = connect(url)
# add support for a REGEXP and POW implementation
# TODO: this should be added only for the SQLite case and doesn't apply to other vendors.
self.__db.connection().create_function("REGEXP", 2, regexp)
self.__db.connection().create_function("POW", 2, pow)
self.__cursor = self.__db.cursor()
self.__cursor.arraysize = 100
# what shall I do here to enable EXPLAIN PLANs?

Solution #

Great question! There are a couple of clean, maintainable ways to implement this in Peewee—whether you need to cover ORM-generated queries, raw SQL, or both. Let's break it down:

Method 1: Override Peewee's execute_sql (covers ALL queries) #

Peewee's database classes expose an execute_sql method that runs every query sent to the database. By overriding this method, you can automatically wrap all read queries with EXPLAIN PLAN when your toggle is enabled. This works for both ORM queries (like User.select()) and raw SQL executed via db.execute_sql().

First, create a custom SQLite database class with the toggle logic:

from playhouse.sqlite_ext import SqliteExtDatabase  # Use SqliteDatabase if you don't need extensions

class ExplainedSqliteDatabase(SqliteExtDatabase):
    def __init__(self, *args, enable_explain=False, **kwargs):
        super().__init__(*args, **kwargs)
        self.enable_explain = enable_explain

    def execute_sql(self, sql, params=None, commit=True):
        # Skip explain for write ops, existing EXPLAIN/PRAGMA queries to avoid nesting
        if (self.enable_explain 
            and not sql.strip().upper().startswith(("EXPLAIN", "PRAGMA", "INSERT", "UPDATE", "DELETE"))):
            sql = f"EXPLAIN PLAN {sql}"
        return super().execute_sql(sql, params, commit)

Then adjust your connection setup to use this class (pull the toggle from your config or CLI arguments):

# Example: Fetch toggle from CLI or config (replace with your actual logic)
enable_explain = True  # e.g., argparse.parse_args().explain or config.get("enable_explain")

# Handle SQLite connections with our custom class, fallback for other DBs
from urllib.parse import urlparse
parsed_url = urlparse(url)
if parsed_url.scheme == "sqlite":
    # Extract DB path from the URL (handle in-memory DB too)
    db_path = parsed_url.path.lstrip("/") if parsed_url.path else ":memory:"
    self.__db = ExplainedSqliteDatabase(db_path, enable_explain=enable_explain)
else:
    self.__db = connect(url)

# Add your custom functions as before
self.__db.connection().create_function("REGEXP", 2, regexp)
self.__db.connection().create_function("POW", 2, pow)
self.__cursor = self.__db.cursor()
self.__cursor.arraysize = 100

Bonus: Toggle at runtime

If you need to switch explain mode on/off after initialization, add a setter method to the custom class:

def set_explain_enabled(self, enabled):
    self.enable_explain = enabled

Then call self.__db.set_explain_enabled(True) or False whenever needed.

Method 2: Cursor Wrapper (for raw SQL only) #

If you only need to wrap queries executed directly via your self.__cursor (and not ORM queries), create a lightweight wrapper around the cursor's execute methods:

class ExplainCursor:
    def __init__(self, cursor, enable_explain=False):
        self.cursor = cursor
        self.enable_explain = enable_explain

    # Delegate all unmodified methods/properties to the original cursor
    def __getattr__(self, attr):
        return getattr(self.cursor, attr)

    def execute(self, sql, params=None):
        if (self.enable_explain 
            and not sql.strip().upper().startswith(("EXPLAIN", "PRAGMA", "INSERT", "UPDATE", "DELETE"))):
            sql = f"EXPLAIN PLAN {sql}"
        return self.cursor.execute(sql, params)

    # Handle executemany if you use it
    def executemany(self, sql, seq_of_params):
        if (self.enable_explain 
            and not sql.strip().upper().startswith(("EXPLAIN", "PRAGMA", "INSERT", "UPDATE", "DELETE"))):
            sql = f"EXPLAIN PLAN {sql}"
        return self.cursor.executemany(sql, seq_of_params)

Then replace your cursor initialization:

enable_explain = True  # From config/CLI
self.__cursor = ExplainCursor(self.__db.cursor(), enable_explain=enable_explain)
self.__cursor.arraysize = 100

This only affects queries run via self.__cursor.execute() or executemany(), so it's a good choice if you don't need to cover ORM-generated queries.

内容的提问来源于stack exchange，提问作者SkyWalker