Работа с двигателями и соединениями¶

Вложенность блоков транзакций¶

Объект Transaction также обрабатывает «вложенное» поведение, отслеживая крайнюю пару begin/commit. В этом примере две функции обе выдают транзакцию на Connection, но только крайний объект Transaction фактически вступает в силу после фиксации.

# 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)

Выше, сначала вызывается method_a, который вызывает connection.begin(). Затем вызывается method_b. Когда method_b вызывает connection.begin(), он просто увеличивает счетчик, который уменьшается при вызове commit(). Если либо method_a, либо method_b вызывает rollback(), вся транзакция откатывается. Транзакция не фиксируется до тех пор, пока method_a не вызовет метод commit(). Такое «вложенное» поведение позволяет создавать функции, которые «гарантируют» использование транзакции, если таковой еще не было, но автоматически участвуют в охватывающей транзакции, если таковая существует.

# 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))

Понимание уровня изоляции автокоммита на уровне DBAPI¶

В предыдущем разделе мы познакомились с понятием параметра Connection.execution_options.isolation_level и тем, как его можно использовать для установки уровней изоляции базы данных, включая «автокоммит» на уровне DBAPI, который рассматривается SQLAlchemy как еще один уровень изоляции транзакций. В этом разделе мы попытаемся прояснить последствия такого подхода.

Если бы мы хотели проверить объект Connection и использовать его в режиме «autocommit», мы бы поступили следующим образом:

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

Выше показано обычное использование режима «DBAPI autocommit». Нет необходимости использовать такие методы, как Connection.begin() или Connection.commit() (примечание: последнее относится к использованию 2.0 style).

Однако важно отметить, что указанный выше режим автокоммита постоянный для данного конкретного соединения, пока мы не изменим его непосредственно с помощью isolation_level снова. Уровень изоляции также сбрасывается для соединения DBAPI, когда мы release возвращаем соединение обратно в пул соединений. Однако вызов Connection.begin() не изменит уровень изоляции, что означает, что мы остаемся в автокоммите. Пример ниже иллюстрирует это:

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>)

Когда мы запускаем блок, подобный приведенному выше, с включенным протоколированием, протоколирование попытается указать, что, хотя вызывается уровень DBAPI .commit(), он, вероятно, не будет иметь никакого эффекта из-за режима autocommit:

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

Аналогично, при использовании режима 2.0 style create_engine.future, Connection будет использовать поведение autobegin, что означает, что приведенный ниже шаблон вызовет ошибку:

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>)

Это все для того, чтобы продемонстрировать, что установка уровня изоляции autocommit полностью независима от поведения begin/commit объекта SQLAlchemy Connection. Режим «autocommit» никак не взаимодействует с Connection.begin(), а Connection не обращается к этому статусу при выполнении своих собственных изменений состояния транзакции (за исключением сообщения в журнале движка о том, что эти блоки на самом деле не фиксируются). Обоснованием такой конструкции является поддержание полностью последовательной схемы использования Connection, в которой режим DBAPI-автокоммита может быть изменен независимо без указания каких-либо изменений кода в других местах.

Настройки уровней изоляции, включая режим автокоммита, сбрасываются автоматически, когда соединение возвращается обратно в пул соединений. Поэтому желательно не пытаться переключать уровни изоляции на одном объекте Connection, так как это приводит к излишней многословности.

Чтобы проиллюстрировать, как использовать «autocommit» в ad-hoc режиме в рамках одной проверки Connection, параметр Connection.execution_options.isolation_level должен быть повторно применен с предыдущим уровнем изоляции. Мы можем «правильно» записать наш вышеприведенный блок как (отмечая использование стиля 2.0 ниже):

# 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>)

Выше, чтобы вручную изменить уровень изоляции, мы использовали Connection.default_isolation_level для восстановления уровня изоляции по умолчанию (предполагая, что это то, что мы хотим здесь). Однако, вероятно, лучше работать с архитектурой Connection, которая уже обрабатывает сброс уровня изоляции автоматически при регистрации. предпочтительный способ написать вышеописанное - использовать два блока

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>)

Подведем итоги:

1. Уровень изоляции «автокоммит на уровне DBAPI» полностью независим от понятия «начать» и «зафиксировать» объекта Connection.

2. используйте отдельные проверки Connection для каждого уровня изоляции. Избегайте попыток переключения туда и обратно между «autocommit» при проверке одного соединения; позвольте движку выполнить работу по восстановлению уровней изоляции по умолчанию

Потоковая передача с фиксированным буфером через yield_per¶

Поскольку отдельные операции выборки строк с полностью небуферизованными курсорами на стороне сервера обычно обходятся дороже, чем выборка сразу нескольких строк, параметр выполнения Connection.execution_options.yield_per настраивает Connection или оператор на использование доступных курсоров на стороне сервера, в то же время настраивая буфер строк фиксированного размера, который будет получать строки с сервера партиями по мере их потребления. Этот параметр может иметь целочисленное положительное значение при использовании метода Connection.execution_options() на Connection или на операторе при использовании метода Executable.execution_options().

Использование этой опции эквивалентно ручной установке опции Connection.execution_options.stream_results, описанной в следующем разделе, а затем вызову метода Result.yield_per() на объекте Result с заданным целочисленным значением. В обоих случаях эффект, который дает эта комбинация, включает:

• режим курсоров на стороне сервера выбирается для данного бэкенда, если он доступен и не является поведением по умолчанию для данного бэкенда

• по мере получения строк результата, они будут буферизироваться партиями, где размер каждой партии до последней партии будет равен целочисленному аргументу, переданному опции Connection.execution_options.yield_per или методу Result.yield_per(); последняя партия затем буферизируется относительно оставшихся строк, меньших, чем этот размер

• Размер раздела по умолчанию, используемый методом Result.partitions(), если он используется, также будет сделан равным этому целочисленному размеру.

Эти три вида поведения проиллюстрированы в примере ниже:

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}")

Приведенный выше пример иллюстрирует комбинацию yield_per=100 вместе с использованием метода Result.partitions() для запуска обработки строк в партиях, соответствующих размеру, получаемых с сервера. Использование Result.partitions() является необязательным, и если Result итерируется напрямую, новая партия строк будет буферизироваться для каждых 100 полученных строк. Вызов такого метода, как Result.all(), не должен использоваться, так как это приведет к полной выборке всех оставшихся строк сразу и нарушит цель использования yield_per.

Опция Connection.execution_options.yield_per переносится и в ORM, используется Session для получения объектов ORM, где она также ограничивает количество объектов ORM, создаваемых одновременно. Более подробную информацию об использовании Выборка больших наборов результатов с доходностью за с ORM смотрите в разделе Руководство по составлению запросов ORM - в Connection.execution_options.yield_per.

Потоковая передача с динамически растущим буфером с использованием stream_results¶

Чтобы включить курсоры на стороне сервера без определенного размера раздела, можно использовать опцию Connection.execution_options.stream_results, которая, как и Connection.execution_options.yield_per, может быть вызвана на объекте Connection или объекте statement.

Когда объект Result, переданный с помощью опции Connection.execution_options.stream_results, итерируется напрямую, строки извлекаются внутренне, используя схему буферизации по умолчанию, которая буферизирует сначала небольшой набор строк, затем все больший и больший буфер при каждом извлечении до предварительно настроенного предела в 1000 строк. На максимальный размер этого буфера можно повлиять с помощью опции выполнения Connection.execution_options.max_row_buffer:

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}")

Хотя опция Connection.execution_options.stream_results может быть совмещена с использованием метода Result.partitions(), в опцию Result.partitions() следует передать определенный размер раздела, чтобы не получить весь результат. Обычно проще использовать опцию Connection.execution_options.yield_per при настройке на использование метода Result.partitions().

Конфигурация¶

Сам кэш представляет собой объект типа словаря LRUCache, который является внутренним подклассом словаря SQLAlchemy, который отслеживает использование определенных ключей и имеет периодический шаг «обрезки», который удаляет наименее недавно использованные элементы, когда размер кэша достигает определенного порога. Размер этого кэша по умолчанию равен 500 и может быть настроен с помощью параметра create_engine.query_cache_size:

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

Размер кэша может вырасти в 150% от заданного размера, прежде чем он будет сокращен до целевого размера. Таким образом, кэш размером 1200 может вырасти до 1800 элементов, после чего он будет обрезан до 1200.

Размер кэша основан на одной записи для каждого уникального SQL-запроса, созданного на каждом движке. SQL-запросы, генерируемые как ядром, так и ORM, обрабатываются одинаково. Операторы DDL обычно не кэшируются. Для того чтобы определить, что делает кэш, в журнале регистрации движка будут содержаться подробности о поведении кэша, описанные в следующем разделе.

Оценка производительности кэша с помощью протоколирования¶

Указанный выше размер кэша 1200 на самом деле довольно велик. Для небольших приложений, скорее всего, будет достаточно размера 100. Для оценки оптимального размера кэша, при условии наличия достаточного количества памяти на целевом узле, размер кэша должен быть основан на количестве уникальных строк SQL, которые могут быть отображены для используемого целевого механизма. Наиболее целесообразным способом увидеть это является использование эха SQL, которое наиболее непосредственно включается с помощью флага create_engine.echo или с помощью протоколирования Python; см. раздел Настройка ведения журнала о конфигурации протоколирования.

В качестве примера мы рассмотрим журнал, создаваемый следующей программой:

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)

При запуске каждый SQL-оператор, который регистрируется, будет включать значок статистики кэша в скобках слева от переданных параметров. Ниже перечислены четыре типа сообщений, которые мы можем увидеть:

• [raw sql] - драйвер или конечный пользователь выдал необработанный SQL, используя Connection.exec_driver_sql() - кэширование не применяется

• [no key] - объект оператора является оператором DDL, который не кэшируется, или объект оператора содержит некэшируемые элементы, такие как определяемые пользователем конструкции или произвольно большие предложения VALUES.

• [generated in Xs] - утверждение было пропуском кэша и должно было быть скомпилировано, а затем сохранено в кэше. на создание скомпилированной конструкции ушло X секунд. Число X будет в мелких долях секунды.

• [cached since Xs ago] - утверждение было попаданием в кэш и не требовало перекомпиляции. Утверждение было сохранено в кэше с X секунд назад. Число X будет пропорционально длительности работы приложения и длительности кэширования утверждения, например, 86400 для 24-часового периода.

Каждый бейдж более подробно описан ниже.

Первые утверждения, которые мы увидим для вышеприведенной программы, будут диалектом SQLite, проверяющим существование таблиц «a» и «b»:

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] ()

Для двух вышеприведенных утверждений SQLite PRAGMA значок читается как [raw sql], что указывает на то, что драйвер отправляет строку Python непосредственно в базу данных с помощью Connection.exec_driver_sql(). Кэширование не применяется к таким утверждениям, поскольку они уже существуют в строковой форме, и ничего не известно о том, какие типы строк результатов будут возвращены, поскольку SQLAlchemy не разбирает строки SQL заранее.

Следующие утверждения, которые мы видим, это утверждения CREATE TABLE:

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] ()

Для каждого из этих утверждений значок читается как [no key 0.00006s]. Это указывает на то, что в этих двух конкретных утверждениях кэширование не произошло, поскольку DDL-ориентированная конструкция CreateTable не создала ключ кэша. Конструкции DDL обычно не участвуют в кэшировании, потому что они обычно не повторяются во второй раз, а DDL также является этапом конфигурирования базы данных, где производительность не так критична.

Значок [no key] важен еще по одной причине, поскольку он может быть получен для SQL-запросов, которые можно кэшировать, за исключением некоторой определенной подконструкции, которая в настоящее время не является кэшируемой. Примерами этого являются пользовательские элементы SQL, которые не определяют параметры кэширования, а также некоторые конструкции, которые генерируют произвольно длинные и невоспроизводимые строки SQL, основными примерами которых являются конструкция Values, а также при использовании «многозначных вставок» с помощью метода Insert.values().

Пока наш кэш все еще пуст. Однако следующие утверждения будут кэшироваться, сегмент выглядит следующим образом:

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)

Выше мы видим, по сути, две уникальные SQL-строки: "INSERT INTO a (data) VALUES (?)" и "INSERT INTO b (a_id, data) VALUES (?, ?)". Поскольку SQLAlchemy использует связанные параметры для всех буквальных значений, даже если эти утверждения повторяются много раз для разных объектов, поскольку параметры разделены, фактическая строка SQL остается неизменной.

Значок кэширования, который мы видим для первого появления каждого из этих двух утверждений, - [generated in 0.00011s]. Это означает, что утверждение не было в кэше, было скомпилировано в строку за .00011 с и затем было кэшировано. Когда мы видим значок [generated], мы знаем, что это означает, что произошел промах в кэше. Этого следует ожидать при первом появлении определенного оператора. Однако, если наблюдается много новых значков [generated] для долго работающего приложения, которое обычно использует одну и ту же серию SQL-запросов снова и снова, это может быть признаком того, что параметр create_engine.query_cache_size слишком мал. Когда оператор, который был кэширован, затем удаляется из кэша из-за обрезки менее используемых элементов кэша LRU, при следующем использовании он отобразит значок [generated].

Значок кэширования, который мы затем видим для последующих вхождений каждого из этих двух утверждений, выглядит как [cached since 0.0003533s ago]. Это означает, что утверждение было найдено в кэше, и первоначально было помещено в кэш .0003533 секунды назад. Важно отметить, что хотя значки [generated] и [cached since] означают количество секунд, они означают разные вещи; в случае [generated] это число является приблизительным отсчетом времени, которое потребовалось для компиляции утверждения, и будет чрезвычайно малым. В случае [cached since] это общее время, в течение которого утверждение находилось в кэше. Для приложения, работающего в течение шести часов, это число может быть [cached since 21600 seconds ago], и это хорошо. Большое число «кэшировано с тех пор» является признаком того, что эти утверждения не подвергались промахам кэша в течение длительного времени. Утверждения, которые часто имеют низкое число «cached since», даже если приложение работает долгое время, могут указывать на то, что эти утверждения слишком часто подвергаются промахам кэша, и что значение create_engine.query_cache_size может быть увеличено.

Затем наша примерная программа выполняет несколько SELECT’ов, где мы видим ту же схему «сгенерировано», затем «кэшировано», для SELECT’а таблицы «a», а также для последующих ленивых загрузок таблицы «b»:

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

Из нашей вышеприведенной программы, полный запуск показывает, что всего кэшируется четыре различных SQL-строки. Это указывает на то, что размер кэша четыре будет достаточным. Очевидно, что это очень маленький размер, и размер по умолчанию 500 вполне можно оставить по умолчанию.

Сколько памяти использует кэш?¶

В предыдущем разделе были описаны некоторые методы проверки того, нужно ли увеличивать размер create_engine.query_cache_size. Как узнать, не слишком ли велик кэш? Причина, по которой мы можем захотеть установить create_engine.query_cache_size не выше определенного числа, заключается в том, что у нас есть приложение, которое может использовать очень большое количество различных запросов, например, приложение, которое строит запросы на лету из поискового UX, и мы не хотим, чтобы у нашего хоста закончилась память, если, например, за последние 24 часа было выполнено сто тысяч различных запросов, и все они были закешированы.

Очень трудно измерить, сколько памяти занимают структуры данных Python, однако, используя процесс измерения роста памяти через top по мере добавления в кэш последовательной серии из 250 новых утверждений, можно предположить, что умеренное утверждение Core занимает около 12K, а небольшое утверждение ORM занимает около 20K, включая структуры выборки результатов, которые для ORM будут намного больше.

Отключение или использование альтернативного словаря для кэширования некоторых (или всех) утверждений¶

Используемый внутренний кэш известен как LRUCache, но в основном это просто словарь. Любой словарь можно использовать в качестве кэша для любой серии утверждений, используя опцию Connection.execution_options.compiled_cache в качестве опции выполнения. Опции выполнения могут быть установлены на операторе, на Engine или Connection, а также при использовании метода ORM Session.execute() для вызовов в стиле SQLAlchemy-2.0. Например, чтобы запустить серию SQL-запросов и кэшировать их в определенном словаре:

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

SQLAlchemy ORM использует вышеупомянутую технику для хранения кэша каждого маппера в рамках процесса «промывки» единицы работы, который отделен от кэша по умолчанию, настроенного на Engine, а также для некоторых запросов загрузчика отношений.

Кэш также можно отключить с помощью этого аргумента, отправив значение None:

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

Кэширование для диалектов сторонних производителей¶

Функция кэширования требует, чтобы компилятор диалекта создавал SQL-строки, которые безопасно использовать повторно для многих вызовов операторов, учитывая определенный ключ кэша, привязанный к этой SQL-строке. Это означает, что любые буквальные значения в операторе, такие как значения LIMIT/OFFSET для SELECT, не могут быть жестко закодированы в схеме компиляции диалекта, поскольку скомпилированная строка не будет пригодна для повторного использования. SQLAlchemy поддерживает визуализацию связанных параметров с помощью метода BindParameter.render_literal_execute(), который может быть применен к существующим атрибутам Select._limit_clause и Select._offset_clause пользовательским компилятором, что проиллюстрировано далее в этом разделе.

Поскольку существует множество сторонних диалектов, многие из которых могут генерировать литеральные значения из SQL-операторов без использования новой функции «literal execute», SQLAlchemy в версии 1.4.5 добавила атрибут диалектов, известный как Dialect.supports_statement_cache. Этот атрибут проверяется во время выполнения на наличие непосредственно в классе конкретного диалекта, даже если он уже присутствует в суперклассе, так что даже сторонний диалект, который является подклассом существующего диалекта SQLAlchemy с возможностью кэширования, такого как sqlalchemy.dialects.postgresql.PGDialect, все равно должен явно включать этот атрибут, чтобы кэширование было включено. Атрибут должен только быть включен после того, как диалект был изменен по мере необходимости и протестирован на возможность повторного использования скомпилированных SQL-запросов с различными параметрами.

Для всех сторонних диалектов, которые не поддерживают этот атрибут, в журнале регистрации для такого диалекта будет указано dialect does not support caching.

Если диалект был протестирован на кэширование, и, в частности, компилятор SQL был обновлен, чтобы не отображать любой литерал LIMIT / OFFSET в строке SQL напрямую, авторы диалекта могут применять атрибут следующим образом:

from sqlalchemy.engine.default import DefaultDialect


class MyDialect(DefaultDialect):
    supports_statement_cache = True

Флаг должен применяться также ко всем подклассам диалекта:

class MyDBAPIForMyDialect(MyDialect):
    supports_statement_cache = True

Типичный случай для модификации диалекта следующий.

# 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]

Использование ламбдас для значительного увеличения скорости создания операторов¶

Функции Python, обычно выраженные в виде лямбд, могут использоваться для генерации SQL-выражений, которые кэшируются на основе расположения в коде Python самой лямбда-функции, а также закрывающих переменных внутри лямбды. Это делается для того, чтобы кэшировать не только строково-компилированную форму SQL-выражения, как это обычно происходит в SQLAlchemy, когда не используется система лямбд, но и сам состав SQL-выражения на языке Python, который также имеет некоторую степень накладных расходов на Python.

Функция лямбда-выражения SQL доступна в качестве функции повышения производительности, а также опционально используется в опции with_loader_criteria() ORM для предоставления общего фрагмента SQL.

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())]

Вызов строк SQL непосредственно драйвером¶

Для случая, когда необходимо вызвать текстовый SQL, непосредственно переданный базовому драйверу (известный как DBAPI) без вмешательства конструкции text(), можно использовать метод Connection.exec_driver_sql():

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

Работа с курсором DBAPI напрямую¶

В некоторых случаях SQLAlchemy не предоставляет обобщенного способа доступа к некоторым функциям DBAPI, таким как вызов хранимых процедур, а также работа с несколькими наборами результатов. В этих случаях целесообразно работать с необработанным DBAPI-соединением напрямую.

Наиболее распространенный способ доступа к необработанному соединению DBAPI - получить его из уже присутствующего объекта Connection напрямую. Он присутствует с помощью атрибута Connection.connection:

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

DBAPI-соединение здесь фактически является «проксированным» с точки зрения пула исходных соединений, однако это деталь реализации, которую в большинстве случаев можно игнорировать. Поскольку это DBAPI-соединение по-прежнему находится в области видимости объекта-владельца Connection, лучше всего использовать объект Connection для большинства функций, таких как управление транзакциями, а также вызов метода Connection.close(); если эти операции выполняются непосредственно на DBAPI-соединении, объект-владелец Connection не будет знать об этих изменениях в состоянии.

Чтобы преодолеть ограничения, накладываемые соединением DBAPI, которое поддерживается владельцем Connection, соединение DBAPI также доступно без необходимости сначала приобретать Connection, используя метод Engine.raw_connection() из Engine:

dbapi_conn = engine.raw_connection()

Это соединение DBAPI, как и в предыдущем случае, является «проксированной» формой. Цель этого проксирования теперь очевидна, поскольку когда мы вызываем метод .close() этого соединения, соединение DBAPI обычно не закрывается, а вместо этого released возвращается в пул соединений движка:

dbapi_conn.close()

Хотя SQLAlchemy может в будущем добавить встроенные паттерны для большего количества случаев использования DBAPI, отдача от них снижается, так как эти случаи, как правило, нужны редко, и они также сильно зависят от типа используемого DBAPI, поэтому в любом случае паттерн прямого вызова DBAPI всегда доступен для тех случаев, когда он необходим.

Ниже приведены некоторые рецепты использования соединений DBAPI.

Вызов хранимых процедур и функций, определяемых пользователем¶

SQLAlchemy поддерживает вызов хранимых процедур и функций, определенных пользователем, несколькими способами. Обратите внимание, что все DBAPI имеют различные практики, поэтому вы должны обратиться к документации вашей базовой DBAPI для уточнения специфики вашего конкретного использования. Следующие примеры являются гипотетическими и могут не работать с вашим базовым DBAPI.

Для хранимых процедур или функций с особыми синтаксическими особенностями или параметрами потенциально может быть использован уровень DBAPI callproc с вашим DBAPI. Примером такого шаблона является:

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()

Ваш DBAPI может не иметь требования callproc или может требовать, чтобы хранимая процедура или функция, определяемая пользователем, вызывалась с другим шаблоном, например, с обычным использованием соединения SQLAlchemy. Одним из примеров такого использования является, на момент написания этой документации, выполнение хранимой процедуры в базе данных PostgreSQL с помощью psycopg2 DBAPI, которая должна вызываться с обычным использованием соединения:

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

Приведенный выше пример является гипотетическим. Не гарантируется, что базовая база данных поддерживает «CALL» или «SELECT» в этих ситуациях, и ключевое слово может меняться в зависимости от того, является ли функция хранимой процедурой или функцией, определенной пользователем. В таких ситуациях следует обратиться к документации по DBAPI и базе данных, чтобы определить правильный синтаксис и шаблоны для использования.

Множественные наборы результатов¶

Поддержка нескольких наборов результатов доступна из необработанного курсора DBAPI с помощью метода nextset:

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()

Регистрация диалектов в процессе¶

SQLAlchemy также позволяет регистрировать диалект в текущем процессе, минуя необходимость отдельной установки. Используйте функцию register() следующим образом:

from sqlalchemy.dialects import registry

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

Вышеприведенное ответит на create_engine("mysql+foodialect://") и загрузит класс MyMySQLDialect из модуля myapp.dialect.