Skip to content

Query Plan SQL construct

Jump to bottom
Michael Bayer edited this page Nov 18, 2024 · 5 revisions

This example illustrates how to implement PostgreSQL's EXPLAIN and EXPLAIN ANALYZE around SQLAlchemy Core statement objects, using a construct that can be executed like any other statement to provide the results of a query plan. Syntaxes for other databases can be added as well.

Examples for 1.4 and 1.3 are shown below. Note the 1.4 interface for creating a custom SQL object might be simplified a bit soon.

"""EXPLAIN any query.

Includes PostgreSQL EXPLAIN syntax.

SQLAlchemy 1.4 / 2.0 version.


"""
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Integer
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import Executable


class explain(Executable, ClauseElement):
    inherit_cache = False

    def __init__(self, stmt, analyze=False):
        self.statement = stmt
        self.analyze = analyze


@compiles(explain, "postgresql")
def pg_explain(element, compiler, **kw):
    text = "EXPLAIN "
    if element.analyze:
        text += "ANALYZE "
    text += compiler.process(element.statement, **kw)

    return text


if __name__ == "__main__":
    from sqlalchemy.orm import aliased
    from sqlalchemy.orm import declarative_base
    from sqlalchemy.orm import Session
    from sqlalchemy import select

    Base = declarative_base()

    class Foo(Base):
        __tablename__ = "foo"
        id = Column(Integer, primary_key=True)

    engine = create_engine(
        "postgresql://scott:tiger@localhost/test", echo=True
    )

    Base.metadata.create_all(engine)

    with Session(engine) as sess:

        falias = aliased(Foo)
        q = select(Foo).where(Foo.id > 5).join(falias, Foo.id == falias.id)

        print(sess.execute(explain(q)).fetchall())
"""EXPLAIN any query.

Includes PostgreSQL EXPLAIN syntax.

SQLAlchemy 1.3 and prior version.


"""

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import Integer
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import _literal_as_text
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import Executable


class explain(Executable, ClauseElement):
    def __init__(self, stmt, analyze=False):
        self.statement = _literal_as_text(stmt)
        self.analyze = analyze


@compiles(explain, "postgresql")
def pg_explain(element, compiler, **kw):
    text = "EXPLAIN "
    if element.analyze:
        text += "ANALYZE "
    text += compiler.process(element.statement, **kw)

    return text


if __name__ == "__main__":
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import Session
    from sqlalchemy.orm import aliased

    Base = declarative_base()

    class Foo(Base):
        __tablename__ = "foo"
        id = Column(Integer, primary_key=True)

    engine = create_engine(
        "postgresql://scott:tiger@localhost/test", echo=True
    )

    Base.metadata.create_all(engine)

    sess = Session(engine)

    falias = aliased(Foo)
    q = sess.query(Foo).filter(Foo.id > 5).join(falias, Foo.id == falias.id)

    print(sess.execute(explain(q)).fetchall())
Clone this wiki locally