Метод exec_driver_sql класса AsyncConnection

Метод exec_driver_sql класса AsyncConnection выполняет SQL-запрос напрямую через драйвер базы данных, минуя слой компиляции выражений SQLAlchemy. Первым параметром метод принимает строку SQL-запроса. Вторым необязательным параметром передаются значения для подстановки в запрос. Такой подход удобен, когда нужно выполнить специфичный для конкретной СУБД синтаксис или уже готовую строку запроса.

Метод является асинхронным, поэтому вызывается через await. Возвращает объект результата, из которого можно получить строки через методы fetchall, fetchone или fetchmany.

Синтаксис

await connection.exec_driver_sql(statement, parameters)

Пример

Давайте создадим асинхронный движок и выполним простой запрос через exec_driver_sql:

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	
	async with engine.connect() as conn:
		res = await conn.exec_driver_sql('SELECT 1')
		print(res.fetchall())
	
	await engine.dispose()

asyncio.run(main())

Результат выполнения кода:

[(1,)]

Пример

Давайте создадим таблицу и вставим данные, используя сырой SQL-запрос с параметрами:

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	
	async with engine.begin() as conn:
		await conn.exec_driver_sql(
			'CREATE TABLE articles ('
			'id INTEGER PRIMARY KEY, '
			'title VARCHAR, '
			'text VARCHAR, '
			'status VARCHAR, '
			'num INTEGER)'
		)
		await conn.exec_driver_sql(
			'INSERT INTO articles (title, text, status, num) '
			'VALUES (?, ?, ?, ?)',
			('article 1', 'text 1', 'published', 10)
		)
		res = await conn.exec_driver_sql('SELECT id, title FROM articles')
		print(res.fetchall())
	
	await engine.dispose()

asyncio.run(main())

Результат выполнения кода:

[(1, 'article 1')]

Пример

Давайте выполним запрос с именованными параметрами через exec_driver_sql:

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	
	async with engine.begin() as conn:
		await conn.exec_driver_sql(
			'CREATE TABLE articles ('
			'id INTEGER PRIMARY KEY, '
			'title VARCHAR, '
			'text VARCHAR, '
			'status VARCHAR, '
			'num INTEGER)'
		)
		await conn.exec_driver_sql(
			'INSERT INTO articles (title, text, status, num) '
			'VALUES (:title, :text, :status, :num)',
			{'title': 'article 2', 'text': 'text 2', 'status': 'draft', 'num': 20}
		)
		res = await conn.exec_driver_sql(
			'SELECT title, num FROM articles WHERE status = :status',
			{'status': 'draft'}
		)
		print(res.fetchall())
	
	await engine.dispose()

asyncio.run(main())

Результат выполнения кода:

[('article 2', 20)]

Смотрите также

класс AsyncConnection,
который представляет асинхронное соединение с базой
метод execute,
который выполняет выражения SQLAlchemy
метод begin,
который открывает транзакцию соединения
метод close,
который закрывает асинхронное соединение