Working with Engines and Connections¶

Nesting of Transaction Blocks¶

The Transaction object also handles “nested” behavior by keeping track of the outermost begin/commit pair. In this example, two functions both issue a transaction on a Connection, but only the outermost Transaction object actually takes effect when it is committed.

# method_a starts a transaction and calls method_b
def method_a(connection):
    with connection.begin():  # open a transaction
        method_b(connection)


# method_b also starts a transaction
def method_b(connection):
    with connection.begin():  # open a transaction - this runs in the
        # context of method_a's transaction
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})


# open a Connection and call method_a
with engine.connect() as conn:
    method_a(conn)

Above, method_a is called first, which calls connection.begin(). Then it calls method_b. When method_b calls connection.begin(), it just increments a counter that is decremented when it calls commit(). If either method_a or method_b calls rollback(), the whole transaction is rolled back. The transaction is not committed until method_a calls the commit() method. This “nesting” behavior allows the creation of functions which “guarantee” that a transaction will be used if one was not already available, but will automatically participate in an enclosing transaction if one exists.

# method_a calls method_b
def method_a(connection):
    method_b(connection)


# method_b uses the connection and assumes the transaction
# is external
def method_b(connection):
    connection.execute(text("insert into mytable values ('bat', 'lala')"))
    connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})


# open a Connection inside of a transaction and call method_a
with engine.begin() as conn:
    method_a(conn)

import contextlib


@contextlib.contextmanager
def transaction(connection):
    if not connection.in_transaction():
        with connection.begin():
            yield connection
    else:
        yield connection

# method_a starts a transaction and calls method_b
def method_a(connection):
    with transaction(connection):  # open a transaction
        method_b(connection)


# method_b either starts a transaction, or uses the one already
# present
def method_b(connection):
    with transaction(connection):  # open a transaction
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})


# open a Connection and call method_a
with engine.connect() as conn:
    method_a(conn)

import contextlib


def connectivity(engine):
    connection = None

    @contextlib.contextmanager
    def connect():
        nonlocal connection

        if connection is None:
            connection = engine.connect()
            with connection:
                with connection.begin():
                    yield connection
        else:
            yield connection

    return connect

# method_a passes along connectivity context, at the same time
# it chooses to establish a connection by calling "with"
def method_a(connectivity):
    with connectivity():
        method_b(connectivity)


# method_b also wants to use a connection from the context, so it
# also calls "with:", but also it actually uses the connection.
def method_b(connectivity):
    with connectivity() as connection:
        connection.execute(text("insert into mytable values ('bat', 'lala')"))
        connection.execute(mytable.insert(), {"col1": "bat", "col2": "lala"})


# create a new connection/transaction context object and call
# method_a
method_a(connectivity(engine))

Understanding the DBAPI-Level Autocommit Isolation Level¶

In the parent section, we introduced the concept of the Connection.execution_options.isolation_level parameter and how it can be used to set database isolation levels, including DBAPI-level “autocommit” which is treated by SQLAlchemy as another transaction isolation level. In this section we will attempt to clarify the implications of this approach.

If we wanted to check out a Connection object and use it “autocommit” mode, we would proceed as follows:

with engine.connect() as connection:
    connection.execution_options(isolation_level="AUTOCOMMIT")
    connection.execute(<statement>)
    connection.execute(<statement>)

Above illustrates normal usage of “DBAPI autocommit” mode. There is no need to make use of methods such as Connection.begin() or Connection.commit() (noting the latter applies to 2.0 style usage).

What’s important to note however is that the above autocommit mode is persistent on that particular Connection until we change it directly using isolation_level again. The isolation level is also reset on the DBAPI connection when we release the connection back to the connection pool. However, calling upon Connection.begin() will not change the isolation level, meaning we stay in autocommit. The example below illustrates this:

with engine.connect() as connection:
    connection = connection.execution_options(isolation_level="AUTOCOMMIT")

    # this begin() does nothing, isolation stays at AUTOCOMMIT
    with connection.begin() as trans:
        connection.execute(<statement>)
        connection.execute(<statement>)

When we run a block like the above with logging turned on, the logging will attempt to indicate that while a DBAPI level .commit() is called, it probably will have no effect due to autocommit mode:

INFO sqlalchemy.engine.Engine BEGIN (implicit)
...
INFO sqlalchemy.engine.Engine COMMIT using DBAPI connection.commit(), DBAPI should ignore due to autocommit mode

Similarly, when using 2.0 style create_engine.future mode, the Connection will use autobegin behavior, meaning that the pattern below will raise an error:

engine = create_engine(..., future=True)

with engine.connect() as connection:
    connection = connection.execution_options(isolation_level="AUTOCOMMIT")

    # "transaction" is autobegin (but has no effect due to autocommit)
    connection.execute(<statement>)

    # this will raise; "transaction" is already begun
    with connection.begin() as trans:
        connection.execute(<statement>)

This is all to demonstrate that the autocommit isolation level setting is completely independent from the begin/commit behavior of the SQLAlchemy Connection object. The “autocommit” mode will not interact with Connection.begin() in any way and the Connection does not consult this status when performing its own state changes with regards to the transaction (with the exception of suggesting within engine logging that these blocks are not actually committing). The rationale for this design is to maintain a completely consistent usage pattern with the Connection where DBAPI-autocommit mode can be changed independently without indicating any code changes elsewhere.

Isolation level settings, including autocommit mode, are reset automatically when the connection is released back to the connection pool. Therefore it is preferable to avoid trying to switch isolation levels on a single Connection object as this leads to excess verbosity.

To illustrate how to use “autocommit” in an ad-hoc mode within the scope of a single Connection checkout, the Connection.execution_options.isolation_level parameter must be re-applied with the previous isolation level. We can write our above block “correctly” as (noting 2.0 style usage below):

# if we wanted to flip autocommit on and off on a single connection/
# which... we usually don't.

engine = create_engine(..., future=True)

with engine.connect() as connection:

    connection.execution_options(isolation_level="AUTOCOMMIT")

    # run statement(s) in autocommit mode
    connection.execute(<statement>)

    # "commit" the autobegun "transaction" (2.0/future mode only)
    connection.commit()

    # switch to default isolation level
    connection.execution_options(isolation_level=connection.default_isolation_level)

    # use a begin block
    with connection.begin() as trans:
        connection.execute(<statement>)

Above, to manually revert the isolation level we made use of Connection.default_isolation_level to restore the default isolation level (assuming that’s what we want here). However, it’s probably a better idea to work with the architecture of of the Connection which already handles resetting of isolation level automatically upon checkin. The preferred way to write the above is to use two blocks

engine = create_engine(..., future=True)

# use an autocommit block
with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as connection:

    # run statement in autocommit mode
    connection.execute(<statement>)

# use a regular block
with engine.begin() as connection:
    connection.execute(<statement>)

To sum up:

1. “DBAPI level autocommit” isolation level is entirely independent of the Connection object’s notion of “begin” and “commit”

2. use individual Connection checkouts per isolation level. Avoid trying to change back and forth between “autocommit” on a single connection checkout; let the engine do the work of restoring default isolation levels

Streaming with a fixed buffer via yield_per¶

As individual row-fetch operations with fully unbuffered server side cursors are typically more expensive than fetching batches of rows at once, The Connection.execution_options.yield_per execution option configures a Connection or statement to make use of server-side cursors as are available, while at the same time configuring a fixed-size buffer of rows that will retrieve rows from the server in batches as they are consumed. This parameter may be to a positive integer value using the Connection.execution_options() method on Connection or on a statement using the Executable.execution_options() method.

Using this option is equivalent to manually setting the Connection.execution_options.stream_results option, described in the next section, and then invoking the Result.yield_per() method on the Result object with the given integer value. In both cases, the effect this combination has includes:

• server side cursors mode is selected for the given backend, if available and not already the default behavior for that backend

• as result rows are fetched, they will be buffered in batches, where the size of each batch up until the last batch will be equal to the integer argument passed to the Connection.execution_options.yield_per option or the Result.yield_per() method; the last batch is then sized against the remaining rows fewer than this size

• The default partition size used by the Result.partitions() method, if used, will be made equal to this integer size as well.

These three behaviors are illustrated in the example below:

with engine.connect() as conn:
    result = conn.execution_options(yield_per=100).execute(text("select * from table"))

    for partition in result.partitions():
        # partition is an iterable that will be at most 100 items
        for row in partition:
            print(f"{row}")

The above example illustrates the combination of yield_per=100 along with using the Result.partitions() method to run processing on rows in batches that match the size fetched from the server. The use of Result.partitions() is optional, and if the Result is iterated directly, a new batch of rows will be buffered for each 100 rows fetched. Calling a method such as Result.all() should not be used, as this will fully fetch all remaining rows at once and defeat the purpose of using yield_per.

The Connection.execution_options.yield_per option is portable to the ORM as well, used by a Session to fetch ORM objects, where it also limits the amount of ORM objects generated at once. See the section Fetching Large Result Sets with Yield Per - in the ORM Querying Guide for further background on using Connection.execution_options.yield_per with the ORM.

Streaming with a dynamically growing buffer using stream_results¶

To enable server side cursors without a specific partition size, the Connection.execution_options.stream_results option may be used, which like Connection.execution_options.yield_per may be called on the Connection object or the statement object.

When a Result object delivered using the Connection.execution_options.stream_results option is iterated directly, rows are fetched internally using a default buffering scheme that buffers first a small set of rows, then a larger and larger buffer on each fetch up to a pre-configured limit of 1000 rows. The maximum size of this buffer can be affected using the Connection.execution_options.max_row_buffer execution option:

with engine.connect() as conn:
    conn = conn.execution_options(stream_results=True, max_row_buffer=100)
    result = conn.execute(text("select * from table"))

    for row in result:
        print(f"{row}")

While the Connection.execution_options.stream_results option may be combined with use of the Result.partitions() method, a specific partition size should be passed to Result.partitions() so that the entire result is not fetched. It is usually more straightforward to use the Connection.execution_options.yield_per option when setting up to use the Result.partitions() method.

Configuration¶

The cache itself is a dictionary-like object called an LRUCache, which is an internal SQLAlchemy dictionary subclass that tracks the usage of particular keys and features a periodic “pruning” step which removes the least recently used items when the size of the cache reaches a certain threshold. The size of this cache defaults to 500 and may be configured using the create_engine.query_cache_size parameter:

engine = create_engine("postgresql://scott:tiger@localhost/test", query_cache_size=1200)

The size of the cache can grow to be a factor of 150% of the size given, before it’s pruned back down to the target size. A cache of size 1200 above can therefore grow to be 1800 elements in size at which point it will be pruned to 1200.

The sizing of the cache is based on a single entry per unique SQL statement rendered, per engine. SQL statements generated from both the Core and the ORM are treated equally. DDL statements will usually not be cached. In order to determine what the cache is doing, engine logging will include details about the cache’s behavior, described in the next section.

Estimating Cache Performance Using Logging¶

The above cache size of 1200 is actually fairly large. For small applications, a size of 100 is likely sufficient. To estimate the optimal size of the cache, assuming enough memory is present on the target host, the size of the cache should be based on the number of unique SQL strings that may be rendered for the target engine in use. The most expedient way to see this is to use SQL echoing, which is most directly enabled by using the create_engine.echo flag, or by using Python logging; see the section Configuring Logging for background on logging configuration.

As an example, we will examine the logging produced by the following program:

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

Base = declarative_base()


class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    data = Column(String)
    bs = relationship("B")


class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    data = Column(String)


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

s.add_all([A(bs=[B(), B(), B()]), A(bs=[B(), B(), B()]), A(bs=[B(), B(), B()])])
s.commit()

for a_rec in s.query(A):
    print(a_rec.bs)

When run, each SQL statement that’s logged will include a bracketed cache statistics badge to the left of the parameters passed. The four types of message we may see are summarized as follows:

• [raw sql] - the driver or the end-user emitted raw SQL using Connection.exec_driver_sql() - caching does not apply

• [no key] - the statement object is a DDL statement that is not cached, or the statement object contains uncacheable elements such as user-defined constructs or arbitrarily large VALUES clauses.

• [generated in Xs] - the statement was a cache miss and had to be compiled, then stored in the cache. it took X seconds to produce the compiled construct. The number X will be in the small fractional seconds.

• [cached since Xs ago] - the statement was a cache hit and did not have to be recompiled. The statement has been stored in the cache since X seconds ago. The number X will be proportional to how long the application has been running and how long the statement has been cached, so for example would be 86400 for a 24 hour period.

Each badge is described in more detail below.

The first statements we see for the above program will be the SQLite dialect checking for the existence of the “a” and “b” tables:

INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("a")
INFO sqlalchemy.engine.Engine [raw sql] ()
INFO sqlalchemy.engine.Engine PRAGMA main.table_info("b")
INFO sqlalchemy.engine.Engine [raw sql] ()

For the above two SQLite PRAGMA statements, the badge reads [raw sql], which indicates the driver is sending a Python string directly to the database using Connection.exec_driver_sql(). Caching does not apply to such statements because they already exist in string form, and there is nothing known about what kinds of result rows will be returned since SQLAlchemy does not parse SQL strings ahead of time.

The next statements we see are the CREATE TABLE statements:

INFO sqlalchemy.engine.Engine
CREATE TABLE a (
  id INTEGER NOT NULL,
  data VARCHAR,
  PRIMARY KEY (id)
)

INFO sqlalchemy.engine.Engine [no key 0.00007s] ()
INFO sqlalchemy.engine.Engine
CREATE TABLE b (
  id INTEGER NOT NULL,
  a_id INTEGER,
  data VARCHAR,
  PRIMARY KEY (id),
  FOREIGN KEY(a_id) REFERENCES a (id)
)

INFO sqlalchemy.engine.Engine [no key 0.00006s] ()

For each of these statements, the badge reads [no key 0.00006s]. This indicates that these two particular statements, caching did not occur because the DDL-oriented CreateTable construct did not produce a cache key. DDL constructs generally do not participate in caching because they are not typically subject to being repeated a second time and DDL is also a database configurational step where performance is not as critical.

The [no key] badge is important for one other reason, as it can be produced for SQL statements that are cacheable except for some particular sub-construct that is not currently cacheable. Examples of this include custom user-defined SQL elements that don’t define caching parameters, as well as some constructs that generate arbitrarily long and non-reproducible SQL strings, the main examples being the Values construct as well as when using “multivalued inserts” with the Insert.values() method.

So far our cache is still empty. The next statements will be cached however, a segment looks like:

INFO sqlalchemy.engine.Engine INSERT INTO a (data) VALUES (?)
INFO sqlalchemy.engine.Engine [generated in 0.00011s] (None,)
INFO sqlalchemy.engine.Engine INSERT INTO a (data) VALUES (?)
INFO sqlalchemy.engine.Engine [cached since 0.0003533s ago] (None,)
INFO sqlalchemy.engine.Engine INSERT INTO a (data) VALUES (?)
INFO sqlalchemy.engine.Engine [cached since 0.0005326s ago] (None,)
INFO sqlalchemy.engine.Engine INSERT INTO b (a_id, data) VALUES (?, ?)
INFO sqlalchemy.engine.Engine [generated in 0.00010s] (1, None)
INFO sqlalchemy.engine.Engine INSERT INTO b (a_id, data) VALUES (?, ?)
INFO sqlalchemy.engine.Engine [cached since 0.0003232s ago] (1, None)
INFO sqlalchemy.engine.Engine INSERT INTO b (a_id, data) VALUES (?, ?)
INFO sqlalchemy.engine.Engine [cached since 0.0004887s ago] (1, None)

Above, we see essentially two unique SQL strings; "INSERT INTO a (data) VALUES (?)" and "INSERT INTO b (a_id, data) VALUES (?, ?)". Since SQLAlchemy uses bound parameters for all literal values, even though these statements are repeated many times for different objects, because the parameters are separate, the actual SQL string stays the same.

The caching badge we see for the first occurrence of each of these two statements is [generated in 0.00011s]. This indicates that the statement was not in the cache, was compiled into a String in .00011s and was then cached. When we see the [generated] badge, we know that this means there was a cache miss. This is to be expected for the first occurrence of a particular statement. However, if lots of new [generated] badges are observed for a long-running application that is generally using the same series of SQL statements over and over, this may be a sign that the create_engine.query_cache_size parameter is too small. When a statement that was cached is then evicted from the cache due to the LRU cache pruning lesser used items, it will display the [generated] badge when it is next used.

The caching badge that we then see for the subsequent occurrences of each of these two statements looks like [cached since 0.0003533s ago]. This indicates that the statement was found in the cache, and was originally placed into the cache .0003533 seconds ago. It is important to note that while the [generated] and [cached since] badges refer to a number of seconds, they mean different things; in the case of [generated], the number is a rough timing of how long it took to compile the statement, and will be an extremely small amount of time. In the case of [cached since], this is the total time that a statement has been present in the cache. For an application that’s been running for six hours, this number may read [cached since 21600 seconds ago], and that’s a good thing. Seeing high numbers for “cached since” is an indication that these statements have not been subject to cache misses for a long time. Statements that frequently have a low number of “cached since” even if the application has been running a long time may indicate these statements are too frequently subject to cache misses, and that the create_engine.query_cache_size may need to be increased.

Our example program then performs some SELECTs where we can see the same pattern of “generated” then “cached”, for the SELECT of the “a” table as well as for subsequent lazy loads of the “b” table:

INFO sqlalchemy.engine.Engine SELECT a.id AS a_id, a.data AS a_data
FROM a
INFO sqlalchemy.engine.Engine [generated in 0.00009s] ()
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id
INFO sqlalchemy.engine.Engine [generated in 0.00010s] (1,)
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id
INFO sqlalchemy.engine.Engine [cached since 0.0005922s ago] (2,)
INFO sqlalchemy.engine.Engine SELECT b.id AS b_id, b.a_id AS b_a_id, b.data AS b_data
FROM b
WHERE ? = b.a_id

From our above program, a full run shows a total of four distinct SQL strings being cached. Which indicates a cache size of four would be sufficient. This is obviously an extremely small size, and the default size of 500 is fine to be left at its default.

How much memory does the cache use?¶

The previous section detailed some techniques to check if the create_engine.query_cache_size needs to be bigger. How do we know if the cache is not too large? The reason we may want to set create_engine.query_cache_size to not be higher than a certain number would be because we have an application that may make use of a very large number of different statements, such as an application that is building queries on the fly from a search UX, and we don’t want our host to run out of memory if for example, a hundred thousand different queries were run in the past 24 hours and they were all cached.

It is extremely difficult to measure how much memory is occupied by Python data structures, however using a process to measure growth in memory via top as a successive series of 250 new statements are added to the cache suggest a moderate Core statement takes up about 12K while a small ORM statement takes about 20K, including result-fetching structures which for the ORM will be much greater.

Disabling or using an alternate dictionary to cache some (or all) statements¶

The internal cache used is known as LRUCache, but this is mostly just a dictionary. Any dictionary may be used as a cache for any series of statements by using the Connection.execution_options.compiled_cache option as an execution option. Execution options may be set on a statement, on an Engine or Connection, as well as when using the ORM Session.execute() method for SQLAlchemy-2.0 style invocations. For example, to run a series of SQL statements and have them cached in a particular dictionary:

my_cache = {}
with engine.connect().execution_options(compiled_cache=my_cache) as conn:
    conn.execute(table.select())

The SQLAlchemy ORM uses the above technique to hold onto per-mapper caches within the unit of work “flush” process that are separate from the default cache configured on the Engine, as well as for some relationship loader queries.

The cache can also be disabled with this argument by sending a value of None:

# disable caching for this connection
with engine.connect().execution_options(compiled_cache=None) as conn:
    conn.execute(table.select())

Caching for Third Party Dialects¶

The caching feature requires that the dialect’s compiler produces SQL strings that are safe to reuse for many statement invocations, given a particular cache key that is keyed to that SQL string. This means that any literal values in a statement, such as the LIMIT/OFFSET values for a SELECT, can not be hardcoded in the dialect’s compilation scheme, as the compiled string will not be re-usable. SQLAlchemy supports rendered bound parameters using the BindParameter.render_literal_execute() method which can be applied to the existing Select._limit_clause and Select._offset_clause attributes by a custom compiler, which are illustrated later in this section.

As there are many third party dialects, many of which may be generating literal values from SQL statements without the benefit of the newer “literal execute” feature, SQLAlchemy as of version 1.4.5 has added an attribute to dialects known as Dialect.supports_statement_cache. This attribute is checked at runtime for its presence directly on a particular dialect’s class, even if it’s already present on a superclass, so that even a third party dialect that subclasses an existing cacheable SQLAlchemy dialect such as sqlalchemy.dialects.postgresql.PGDialect must still explicitly include this attribute for caching to be enabled. The attribute should only be enabled once the dialect has been altered as needed and tested for reusability of compiled SQL statements with differing parameters.

For all third party dialects that don’t support this attribute, the logging for such a dialect will indicate dialect does not support caching.

When a dialect has been tested against caching, and in particular the SQL compiler has been updated to not render any literal LIMIT / OFFSET within a SQL string directly, dialect authors can apply the attribute as follows:

from sqlalchemy.engine.default import DefaultDialect


class MyDialect(DefaultDialect):
    supports_statement_cache = True

The flag needs to be applied to all subclasses of the dialect as well:

class MyDBAPIForMyDialect(MyDialect):
    supports_statement_cache = True

The typical case for dialect modification follows.

# pre 1.4 style code
def limit_clause(self, select, **kw):
    text = ""
    if select._limit is not None:
        text += " \n LIMIT %d" % (select._limit,)
    if select._offset is not None:
        text += " \n OFFSET %d" % (select._offset,)
    return text

# 1.4 cache-compatible code
def limit_clause(self, select, **kw):
    text = ""

    limit_clause = select._limit_clause
    offset_clause = select._offset_clause

    if select._simple_int_clause(limit_clause):
        text += " \n LIMIT %s" % (
            self.process(limit_clause.render_literal_execute(), **kw)
        )
    elif limit_clause is not None:
        # assuming the DB doesn't support SQL expressions for LIMIT.
        # Otherwise render here normally
        raise exc.CompileError(
            "dialect 'mydialect' can only render simple integers for LIMIT"
        )
    if select._simple_int_clause(offset_clause):
        text += " \n OFFSET %s" % (
            self.process(offset_clause.render_literal_execute(), **kw)
        )
    elif offset_clause is not None:
        # assuming the DB doesn't support SQL expressions for OFFSET.
        # Otherwise render here normally
        raise exc.CompileError(
            "dialect 'mydialect' can only render simple integers for OFFSET"
        )

    return text

SELECT x FROM y
LIMIT __[POSTCOMPILE_param_1]
OFFSET __[POSTCOMPILE_param_2]

Using Lambdas to add significant speed gains to statement production¶

Python functions, typically expressed as lambdas, may be used to generate SQL expressions which are cacheable based on the Python code location of the lambda function itself as well as the closure variables within the lambda. The rationale is to allow caching of not only the SQL string-compiled form of a SQL expression construct as is SQLAlchemy’s normal behavior when the lambda system isn’t used, but also the in-Python composition of the SQL expression construct itself, which also has some degree of Python overhead.

The lambda SQL expression feature is available as a performance enhancing feature, and is also optionally used in the with_loader_criteria() ORM option in order to provide a generic SQL fragment.

from sqlalchemy import lambda_stmt


def run_my_statement(connection, parameter):
    stmt = lambda_stmt(lambda: select(table))
    stmt += lambda s: s.where(table.c.col == parameter)
    stmt += lambda s: s.order_by(table.c.id)

    return connection.execute(stmt)


with engine.connect() as conn:
    result = run_my_statement(some_connection, "some parameter")

>>> from sqlalchemy import select, column
>>> stmt = select(column("q"))
>>> cache_key = stmt._generate_cache_key()
>>> print(cache_key)  # somewhat paraphrased
CacheKey(key=(
  '0',
  <class 'sqlalchemy.sql.selectable.Select'>,
  '_raw_columns',
  (
    (
      '1',
      <class 'sqlalchemy.sql.elements.ColumnClause'>,
      'name',
      'q',
      'type',
      (
        <class 'sqlalchemy.sql.sqltypes.NullType'>,
      ),
    ),
  ),
  # a few more elements are here, and many more for a more
  # complicated SELECT statement
),)

>>> from sqlalchemy import lambda_stmt
>>> stmt = lambda_stmt(lambda: select(column("q")))
>>> cache_key = stmt._generate_cache_key()
>>> print(cache_key)
CacheKey(key=(
  <code object <lambda> at 0x7fed1617c710, file "<stdin>", line 1>,
  <class 'sqlalchemy.sql.lambdas.StatementLambdaElement'>,
),)

>>> def my_stmt(parameter):
...     col = column("q")
...     stmt = lambda_stmt(lambda: select(col))
...     stmt += lambda s: s.where(col == parameter)
...     return stmt

>>> stmt = my_stmt(5)
>>> key = stmt._generate_cache_key()
>>> print(key)
CacheKey(key=(
  <code object <lambda> at 0x7f07323c50e0, file "<stdin>", line 3>,
  (
    '0',
    <class 'sqlalchemy.sql.elements.ColumnClause'>,
    'name',
    'q',
    'type',
    (
      <class 'sqlalchemy.sql.sqltypes.NullType'>,
    ),
  ),
  <code object <lambda> at 0x7f07323c5190, file "<stdin>", line 4>,
  <class 'sqlalchemy.sql.lambdas.LinkedLambdaElement'>,
  (
    '0',
    <class 'sqlalchemy.sql.elements.ColumnClause'>,
    'name',
    'q',
    'type',
    (
      <class 'sqlalchemy.sql.sqltypes.NullType'>,
    ),
  ),
  (
    '0',
    <class 'sqlalchemy.sql.elements.ColumnClause'>,
    'name',
    'q',
    'type',
    (
      <class 'sqlalchemy.sql.sqltypes.NullType'>,
    ),
  ),
),)

>>> key.bindparams
[BindParameter('%(139668884281280 parameter)s', 5, type_=Integer())]

Invoking SQL strings directly to the driver¶

For the use case where one wants to invoke textual SQL directly passed to the underlying driver (known as the DBAPI) without any intervention from the text() construct, the Connection.exec_driver_sql() method may be used:

with engine.connect() as conn:
    conn.exec_driver_sql("SET param='bar'")

Working with the DBAPI cursor directly¶

There are some cases where SQLAlchemy does not provide a genericized way at accessing some DBAPI functions, such as calling stored procedures as well as dealing with multiple result sets. In these cases, it’s just as expedient to deal with the raw DBAPI connection directly.

The most common way to access the raw DBAPI connection is to get it from an already present Connection object directly. It is present using the Connection.connection attribute:

connection = engine.connect()
dbapi_conn = connection.connection

The DBAPI connection here is actually a “proxied” in terms of the originating connection pool, however this is an implementation detail that in most cases can be ignored. As this DBAPI connection is still contained within the scope of an owning Connection object, it is best to make use of the Connection object for most features such as transaction control as well as calling the Connection.close() method; if these operations are performed on the DBAPI connection directly, the owning Connection will not be aware of these changes in state.

To overcome the limitations imposed by the DBAPI connection that is maintained by an owning Connection, a DBAPI connection is also available without the need to procure a Connection first, using the Engine.raw_connection() method of Engine:

dbapi_conn = engine.raw_connection()

This DBAPI connection is again a “proxied” form as was the case before. The purpose of this proxying is now apparent, as when we call the .close() method of this connection, the DBAPI connection is typically not actually closed, but instead released back to the engine’s connection pool:

dbapi_conn.close()

While SQLAlchemy may in the future add built-in patterns for more DBAPI use cases, there are diminishing returns as these cases tend to be rarely needed and they also vary highly dependent on the type of DBAPI in use, so in any case the direct DBAPI calling pattern is always there for those cases where it is needed.

Some recipes for DBAPI connection use follow.

Calling Stored Procedures and User Defined Functions¶

SQLAlchemy supports calling stored procedures and user defined functions several ways. Please note that all DBAPIs have different practices, so you must consult your underlying DBAPI’s documentation for specifics in relation to your particular usage. The following examples are hypothetical and may not work with your underlying DBAPI.

For stored procedures or functions with special syntactical or parameter concerns, DBAPI-level callproc may potentially be used with your DBAPI. An example of this pattern is:

connection = engine.raw_connection()
try:
    cursor_obj = connection.cursor()
    cursor_obj.callproc("my_procedure", ["x", "y", "z"])
    results = list(cursor_obj.fetchall())
    cursor_obj.close()
    connection.commit()
finally:
    connection.close()

Your DBAPI may not have a callproc requirement or may require a stored procedure or user defined function to be invoked with another pattern, such as normal SQLAlchemy connection usage. One example of this usage pattern is, at the time of this documentation’s writing, executing a stored procedure in the PostgreSQL database with the psycopg2 DBAPI, which should be invoked with normal connection usage:

connection.execute("CALL my_procedure();")

This above example is hypothetical. The underlying database is not guaranteed to support “CALL” or “SELECT” in these situations, and the keyword may vary dependent on the function being a stored procedure or a user defined function. You should consult your underlying DBAPI and database documentation in these situations to determine the correct syntax and patterns to use.

Multiple Result Sets¶

Multiple result set support is available from a raw DBAPI cursor using the nextset method:

connection = engine.raw_connection()
try:
    cursor_obj = connection.cursor()
    cursor_obj.execute("select * from table1; select * from table2")
    results_one = cursor_obj.fetchall()
    cursor_obj.nextset()
    results_two = cursor_obj.fetchall()
    cursor_obj.close()
finally:
    connection.close()

Registering Dialects In-Process¶

SQLAlchemy also allows a dialect to be registered within the current process, bypassing the need for separate installation. Use the register() function as follows:

from sqlalchemy.dialects import registry

registry.register("mysql.foodialect", "myapp.dialect", "MyMySQLDialect")

The above will respond to create_engine("mysql+foodialect://") and load the MyMySQLDialect class from the myapp.dialect module.