SQL Expression Language — это способ писать инструкции в приложениях Python вне зависимости от типа используемой базы данных.

Будем использовать таблицу созданную в предыдущей статье.

Вставка (добавление) записей

Есть несколько способов вставить записи в базу данных. Основной — метод insert() экземпляра Table. Его нужно вызвать, после чего использовать метод values() и передать значения для колонок в качестве аргументов-ключевых слов:

ins = customers.insert().values(
    first_name = 'Dmitriy',
    last_name = 'Yatsenko',
    username = 'Moseend',
    email = 'moseend@mail.com',
    address = 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23',
    town = ' Vladivostok'
)

print(ins)

Чтобы увидеть, какой SQL код будет сгенерирован в результате, достаточно вывести ins:

INSERT INTO customers (first_name, last_name, username, email, address, town, created_on, updated_on) VALUES (:first_name, :last_name, :username, :email, :address, :town, :created_on, :updated_on)

Стоит обратить внимание на то, что внутри оператора VALUES находятся связанные параметры (параметры в формате :name), а не сами значения, переданные в метод values().

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

Посмотреть на то, какие значения будут на месте связанных параметров, можно с помощью такой инструкции: ins.compile().params.

Вывод:

{'first_name': 'Dmitriy', 
'last_name': 'Yatsenko', 
'username': 'Moseend', 
'email': 'moseend@mail.com', 
'address': 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', 
'town': ' Vladivostok', 
'created_on': None, 
'updated_on': None}

Инструкция создана, но не отправлена в базу данных. Для ее вызова нужно вызвать метод execute() на объекте Connection:

ins = customers.insert().values(
    first_name = 'Dmitriy',
    last_name = 'Yatsenko',
    username = 'Moseend',
    email = 'moseend@mail.com',
    address = 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23',
    town = ' Vladivostok'
)

conn = engine.connect()
r = conn.execute(ins)

Этот код вставляет следующую запись в таблицу customers:

Метод execute() возвращает объект типа ResultProxy. Последний предоставляет несколько атрибутов, один из которых называется inserted_primary_key. Он возвращает первичный ключ вставленной записи.

Еще один способ создания инструкции для вставки — использование функции insert() из библиотеки sqlalchemy.

from sqlalchemy import insert

ins = insert(customers).values(
    first_name = 'Valeriy',
    last_name = 'Golyshkin',
    username = 'Fortioneaks',
    email = 'fortioneaks@gmail.com',
    address = 'Narovchatova, bld. 8, appt. 37',
    town = 'Magadan'
)
conn = engine.connect()
r = conn.execute(ins)
print(r.inserted_primary_key)

Вывод: (2,).

Вставка (добавление) нескольких записей

Вместо того чтобы передавать значения в метод values() в качестве аргументов-ключевых слов, их можно передать в метод execute().

from sqlalchemy import insert

conn = engine.connect()
ins = insert(customers)

r = conn.execute(ins, 
    first_name = "Vadim", 
    last_name = "Moiseenko", 
    username = "Antence73", 
    email = "antence73@mail.com",
    address = 'Partizanskiy Prospekt, bld. 28/А, appt. 51',
    town = ' Vladivostok'
)

Метод execute() достаточно гибкий, потому что он позволяет вставить несколько записей, передав значения в качестве списка словарей, где каждый — значения для одной строки:

r = conn.execute(ins, [
        {
            "first_name": "Vladimir", 
            "last_name": "Belousov", 
            "username": "Andescols", 
            "email":"andescols@mail.com", 
            "address": "Ul. Usmanova, bld. 70, appt. 223",
            "town": " Naberezhnye Chelny"
        },
        {
            "first_name": "Tatyana", 
            "last_name": "Khakimova", 
            "username": "Caltin1962", 
            "email":"caltin1962@mail.com",
            "address": "Rossiyskaya, bld. 153, appt. 509",
            "town": "Ufa"
        },
        {
            "first_name": "Pavel", 
            "last_name": "Arnautov", 
            "username": "Lablen", 
            "email":"lablen@mail.com",
            "address": "Krasnoyarskaya Ul., bld. 35, appt. 57",
            "town": "Irkutsk"
        },
    ])

print(r.rowcount)

Вывод: 3.

Прежде чем переходить к следующему разделу, добавим также записи в таблицы items, orders и order_lines:

items_list = [
    {
        "name":"Chair",
        "cost_price": 9.21,
        "selling_price": 10.81,
        "quantity": 6
    },
    {
        "name":"Pen",
        "cost_price": 3.45,
        "selling_price": 4.51,
        "quantity": 3
    },
    {
        "name":"Headphone",
        "cost_price": 15.52,
        "selling_price": 16.81,
        "quantity": 50
    },
    {
        "name":"Travel Bag",
        "cost_price": 20.1,
        "selling_price": 24.21,
        "quantity": 50
    },
    {
        "name":"Keyboard",
        "cost_price": 20.12,
        "selling_price": 22.11,
        "quantity": 50
    },
    {
        "name":"Monitor",
        "cost_price": 200.14,
        "selling_price": 212.89,
        "quantity": 50
    },
    {
        "name":"Watch",
        "cost_price": 100.58,
        "selling_price": 104.41,
        "quantity": 50
    },
    {
        "name":"Water Bottle",
        "cost_price": 20.89,
        "selling_price": 25.00,
        "quantity": 50
    },
]

order_list = [
    {
        "customer_id": 1
    },
    {
        "customer_id": 1
    }
]

order_line_list = [
    {
        "order_id": 1,
        "item_id": 1,
        "quantity": 5
    }, 
    {
        "order_id": 1,
        "item_id": 2,
        "quantity": 2
    }, 
    {
        "order_id": 1,
        "item_id": 3,
        "quantity": 1
    },
    {
        "order_id": 2,
        "item_id": 1,
        "quantity": 5
    },
    {
        "order_id": 2,
        "item_id": 2,
        "quantity": 5
    },
]

r = conn.execute(insert(items), items_list)
print(r.rowcount)
r = conn.execute(insert(orders), order_list)
print(r.rowcount)
r = conn.execute(insert(order_lines), order_line_list)
print(r.rowcount)

Вывод:

8
2
5

Получение записей

Для получения записей используется метод select() на экземпляре объекта Table:

s = customers.select()
print(s)

Вывод:

SELECT customers.id, customers.first_name, customers.last_name, customers.username, customers.email, customers.address, customers.town, customers.created_on, customers.updated_on 
FROM customers

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

from sqlalchemy import select
s = select([customers])
print(s)

Вывод буде тот же.

Для отправки запроса нужно выполнить метод execute():

from sqlalchemy import select

conn = engine.connect()

s = select([customers])
r = conn.execute(s)
print(r.fetchall())

Вывод:

[(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583)), (2, 'Valeriy', 'Golyshkin', 'Fortioneaks', 'fortioneaks@gmail.com', 'Narovchatova, bld. 8, appt. 37', 'Magadan', datetime.datetime(2021, 4, 21, 17, 54, 30, 209109), datetime.datetime(2021, 4, 21, 17, 54, 30, 209109)),...)]

Метод fetchall() на объекте ResultProxy возвращает все записи, соответствующие запросу. Как только результаты будут исчерпаны, последующие запросы к fetchall() вернут пустой список.

Метод fetchall() загружает все результаты в память сразу. В случае большого количества данных это не очень эффективно. Как вариант, можно использовать цикл для перебора по результатам:

s = select([customers])
rs = conn.execute(s)
for row in rs:
    print(row)

Вывод:

(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583))
...
(7, 'Pavel', 'Arnautov', 'Lablen', 'lablen@mail.com', 'Krasnoyarskaya Ul., bld. 35, appt. 57', 'Irkutsk', datetime.datetime(2021, 4, 22, 10, 32, 45, 364619), datetime.datetime(2021, 4, 22, 10, 32, 45, 364619))

Дальше список часто используемых методов и атрибутов объекта ResultProxy:

Следующие сессии терминала демонстрируют рассмотренные выше методы и атрибуты в действии, где s = select([customers]).

fetchone()

r = conn.execute(s)
print(r.fetchone())
print(r.fetchone())

(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583))
(2, 'Valeriy', 'Golyshkin', 'Fortioneaks', 'fortioneaks@gmail.com', 'Narovchatova, bld. 8, appt. 37', 'Magadan', datetime.datetime(2021, 4, 21, 17, 54, 30, 209109), datetime.datetime(2021, 4, 21, 17, 54, 30, 209109))

fetchmany()

r = conn.execute(s)
print(r.fetchmany(2))
print(len(r.fetchmany(5)))  # вернется 4, потому что у нас всего 6 записей

[(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583)), (2, 'Valeriy', 'Golyshkin', 'Fortioneaks', 'fortioneaks@gmail.com', 'Narovchatova, bld. 8, appt. 37', 'Magadan', datetime.datetime(2021, 4, 21, 17, 54, 30, 209109), datetime.datetime(2021, 4, 21, 17, 54, 30, 209109))]
4

first()

r = conn.execute(s)
print(r.first())
print(r.first())  # вернется ошибка

(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583))
Traceback (most recent call last):
...
sqlalchemy.exc.ResourceClosedError: This result object is closed.

rowcount

r = conn.execute(s)
print(r.rowcount)
# вернется 6

keys()

r = conn.execute(s)
print(r.keys())

RMKeyView(['id', 'first_name', 'last_name', 'username', 'email', 'address', 'town', 'created_on', 'updated_on'])

scalar()

r = conn.execute(s)
print(r.scalar())
# вернется 1

Важно отметить, что fetchxxx() и first() возвращают не кортежи или словари, а объекты типа LegacyRow, что позволяет получать доступ к данным в записи с помощью названия колонки, индекса или экземпляра Column. Например:

r = conn.execute(s)
row = r.fetchone()
print(row)
print(type(row))
print(row['id'], row['first_name'])  # доступ к данным по названию колонки
print(row[0], row[1])  # доступ к данным по индексу
print(row[customers.c.id], row[customers.c.first_name])  # доступ к данным через объект класса
print(row.id, row.first_name)  # доступ к данным, как к атрибуту

Вывод:

(1, 'Dmitriy', 'Yatsenko', 'Moseend', 'moseend@mail.com', 'Shemilovskiy 2-Y Per., bld. 8/10, appt. 23', ' Vladivostok', datetime.datetime(2021, 4, 21, 17, 33, 35, 172583), datetime.datetime(2021, 4, 21, 17, 33, 35, 172583))
<class 'sqlalchemy.engine.row.LegacyRow'>
1 Dmitriy
1 Dmitriy
1 Dmitriy
1 Dmitriy

Для получения данных из нескольких таблиц нужно передать список экземпляров Table, разделенных запятыми в функцию select():

Этот код вернет Декартово произведение записей из обоих таблиц. О SQL JOIN поговорим позже отдельно.

Фильтр записей

Для фильтрования записей используется метод where(). Он принимает условие и добавляет оператор WHERE к SELECT:

s = select([items]).where(
    items.c.cost_price > 20
)

print(s)
r = conn.execute(s)
print(r.fetchall())

Запрос вернет все элементы, цена которых выше 20.

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items 
WHERE items.cost_price > :cost_price_1

[(4, 'Travel Bag', Decimal('20.10'), Decimal('24.21'), 50), 
 (5, 'Keyboard', Decimal('20.12'), Decimal('22.11'), 50), 
 (6, 'Monitor', Decimal('200.14'), Decimal('212.89'), 50), 
 (7, 'Watch', Decimal('100.58'), Decimal('104.41'), 50), 
 (8, 'Water Bottle', Decimal('20.89'), Decimal('25.00'), 50)]

Дополнительные условия можно задать, просто добавив несколько вызовов метода where().

s = select([items]).\
    where(items.c.cost_price + items.c.selling_price > 50).\
    where(items.c.quantity > 10)

При использовании такого способа операторы просто объединяются при помощи AND. А как использовать OR или NOT?

Для этого есть:

1. Побитовые операторы
2. Союзы

Побитовые операторы

Побитовые операторы &, | и ~ позволяют объединять условия с операторами AND, OR или NOT из SQL.

Предыдущий запрос можно записать вот так с помощью побитовых операторов:

s = select([items]).\
    where(
      (items.c.cost_price + items.c.selling_price > 50) & 
      (items.c.quantity > 10)
    )

Условия заключены в скобки. Это нужно из-за того, что побитовые операторы имеют более высокий приоритет по сравнению с операторами + и >.

s = select([items]).\
    where(
      (items.c.cost_price > 200 ) | 
      (items.c.quantity < 5)
    ) 
print(s)

s = select([items]).\
    where(    
      ~(items.c.quantity == 50)
    ) 
print(s)

s = select([items]).\
    where(
      ~(items.c.quantity == 50) &
      (items.c.cost_price < 20)
    )
print(s)

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items 
WHERE items.cost_price > :cost_price_1 OR items.quantity < :quantity_1

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items 
WHERE items.quantity != :quantity_1

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items 
WHERE items.quantity != :quantity_1 AND items.cost_price < :cost_price_1

Союзы

Условия можно объединять и с помощью функций-союзов and_(), or_() и not_(). Это предпочтительный способ добавления условий в SQLAlchemy.

from sqlalchemy import select, and_, or_, not_

select([items]).\
where(    
    and_(
        items.c.quantity >= 50,
        items.c.cost_price < 100,
    )
)

select([items]).\
where(    
    or_(
        items.c.quantity >= 50,
        items.c.cost_price < 100,
    )
)

select([items]).\
where(    
    and_(
        items.c.quantity >= 50,
        items.c.cost_price < 100,
        not_(
            items.c.name == 'Headphone'            
        ),        
    )
)

Другие распространенные операторы сравнения

Следующий список демонстрирует как использовать остальные операторы сравнения при определении условий в SQLAlchemy

IS NULL

select([orders]).where(
    orders.c.date_shipped == None
)

IS NOT NULL

select([orders]).where(
    orders.c.date_shipped != None
)

IN

select([customers]).where(
    customers.c.first_name.in_(["Valeriy", "Vadim"])
)

NOT IN

select([customers]).where(
    customers.c.first_name.notin_(["Valeriy", "Vadim"])
)

BETWEEN

select([items]).where(
    items.c.cost_price.between(10, 20)
)

NOT BETWEEN

from sqlalchemy import not_

select([items]).where(
    not_(items.c.cost_price.between(10, 20))
)

LIKE

select([items]).where(
    items.c.name.like("Wa%")
)

Метод like() выполняет сравнение с учетом регистра. Для сравнения без учета регистра используйте ilike().

NOT LIKE

select([items]).where(
    not_(items.c.name.like("wa%"))
)

Сортировка результата c order_by

Метод order_by() добавляет оператор ORDER BY к инструкции SELECT. Он принимает одну или несколько колонок для сортировки. Для каждой колонки можно указать, выполнять ли сортировку по возрастанию (asc()) или убыванию (desc()). Если не указать ничего, то сортировка будет выполнена в порядке по возрастанию. Например:

s = select([items]).where(
    items.c.quantity > 10
).order_by(items.c.cost_price)

print(s)
print(conn.execute(s).fetchall())

Вывод:

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items 
WHERE items.quantity > :quantity_1 ORDER BY items.cost_price

[(3, 'Headphone', Decimal('15.52'), Decimal('16.81'), 50), 
 (4, 'Travel Bag', Decimal('20.10'), Decimal('24.21'), 50), 
 (5, 'Keyboard', Decimal('20.12'), Decimal('22.11'), 50), 
 (8, 'Water Bottle', Decimal('20.89'), Decimal('25.00'), 50), 
 (7, 'Watch', Decimal('100.58'), Decimal('104.41'), 50), 
 (6, 'Monitor', Decimal('200.14'), Decimal('212.89'), 50)]

Запрос возвращает записи, отсортированные по cost_price по возрастанию. Это эквивалентно следующему:

from sqlalchemy import asc

s = select([items]).where(
    items.c.quantity > 10
).order_by(asc(items.c.cost_price))

rs = conn.execute(s)
rs.fetchall()

Для сортировки результатов по убыванию используйте функцию desc(). Пример:

from sqlalchemy import desc

s = select([items]).where(
    items.c.quantity > 10
).order_by(desc(items.c.cost_price))

conn.execute(s).fetchall()

Вот еще один пример сортировки по двух колонкам: quantity по возрастанию и cost_price по убыванию.

s = select([items]).order_by(
    items.c.quantity, 
    desc(items.c.cost_price)
)

conn.execute(s).fetchall()

Ограничение результатов c limit

Метод limit() добавляет оператор LIMIT в инструкцию SELECT. Он принимает целое число, определяющее число записей, которые должны вернуться. Например:

s = select([items]).order_by(
    items.c.quantity
).limit(2)

print(s)
print(conn.execute(s).fetchall())

Вывод:

SELECT items.id, items.name, items.cost_price, items.selling_price, items.quantity 
FROM items ORDER BY items.quantity
 LIMIT :param_1

[(2, 'Pen', Decimal('3.45'), Decimal('4.51'), 3), 
 (1, 'Chair', Decimal('9.21'), Decimal('10.81'), 5)]

Чтобы задавать «сдвиг» (начальное положение) в LIMIT, нужно использовать метод offset():

s = select([items]).order_by(
    items.c.quantity
).limit(2).offset(2)

Ограничение колонок

Инструкции SELECT, созданные ранее, возвращают данные из всех колонок. Ограничить количество полей, возвращаемых запросом можно, передав название полей в виде списка в функцию select(). Например:

s = select([items.c.name, items.c.quantity]).where(
    items.c.quantity ==  50
)

print(s)
rs = conn.execute(s)
print(rs.keys())
print(rs.fetchall())

SELECT items.name, items.quantity 
FROM items 
WHERE items.quantity = :quantity_1

RMKeyView(['name', 'quantity'])

[('Headphone', 50), ('Travel Bag', 50), ('Keyboard', 50), ('Monitor', 50), ('Watch', 50), ('Water Bottle', 50)]

Запрос возвращает данные только из колонок name и quantity таблицы items.

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

select([
        items.c.name, 
        items.c.quantity, 
        items.c.selling_price * 5 
    ]).where(
    items.c.quantity ==  50
)

Обратите внимание на то, что items.c.selling_price * 5 — это не реальная колонка, поэтому создается анонимное имя anon_1.

Колонке или выражению можно присвоить метку с помощью метода label(), который работает, добавляя оператор AS к SELECT.

select([
        items.c.name, 
        items.c.quantity, 
        (items.c.selling_price * 5).label('price') 
    ]).where(
    items.c.quantity ==  50
)

Доступ к встроенным функциям

Для доступа к встроенным функциям базы данных используется объект func. Следующий список показывает, как использовать функции для работы с датой/временем, математическими операциями и строками в базе данных PostgreSQL.

from sqlalchemy.sql import func

c = [
    
    ##  функции даты/времени  ##
    
    func.timeofday(),
    func.localtime(),
    func.current_timestamp(),    
    func.date_part("month", func.now()),        
    func.now(),
    
    ##  математические функции  ##
    
    func.pow(4,2),
    func.sqrt(441),
    func.pi(),        
    func.floor(func.pi()),
    func.ceil(func.pi()),
    
    ##  строковые функции  ##
    
    func.lower("ABC"),
    func.upper("abc"),
    func.length("abc"),
    func.trim("  ab c  "),    
    func.chr(65),        
]

s = select(c)
rs = conn.execute(s)
print(rs.keys())
print(rs.fetchall())

RMKeyView(['timeofday_1', 'localtime_1', 'current_timestamp_1', 'date_part_1', 'now_1', 'pow_1', 'sqrt_1', 'pi_1', 'floor_1', 'ceil_1', 'lower_1', 'upper_1', 'length_1', 'trim_1', 'chr_1'])

[('Thu Apr 22 12:33:07.655488 2021 EEST', datetime.time(12, 33, 7, 643174), datetime.datetime(2021, 4, 22, 12, 33, 7, 643174, tzinfo=psycopg2.tz.FixedOffsetTimezone(offset=180, name=None)), 4.0, datetime.datetime(2021, 4, 22, 12, 33, 7, 643174, tzinfo=psycopg2.tz.FixedOffsetTimezone(offset=180, name=None)), 16.0, 21.0, 3.14159265358979, 3.0, 4.0, 'abc', 'ABC', 3, 'ab c', 'A')]

Также можно получить доступ к агрегирующим функциям из объекта object.

c = [ 
    func.sum(items.c.quantity),
    func.avg(items.c.quantity),    
    func.max(items.c.quantity),
    func.min(items.c.quantity),
    func.count(customers.c.id),    
]

s = select(c)
rs = conn.execute(s)
print(rs.fetchall())
# вывод: [(1848, Decimal('38.5000000000000000'), 50, 3, 48)]

Группировка результатов с group_by

Группировка результатов выполняется с помощью оператора GROUP BY. Он часто используется в союзе с агрегирующими функциями. GROUP BY добавляется к SELECT с помощью метода group_by(). Последний принимает одну или несколько колонок и группирует строки по значениям в этих колонках. Например:

c = [ 
    func.count("*").label('count'),         
    customers.c.town      
]

s = select(c).group_by(customers.c.town)

print(conn.execute(s).fetchall())

Вывод:

[(1, 'Ufa'), (1, 'Irkutsk'), (2, ' Vladivostok'), (1, 'Magadan'), (1, ' Naberezhnye Chelny')]

Этот запрос возвращает количество потребителей в каждом городе.

Чтобы отфильтровать результат на основе значений агрегирующих функций, используется метод having(), добавляющий оператор HAVING к SELECT. По аналогии с where() он принимает условие.

c = [
    func.count("*").label('count'),
    customers.c.town      
]

s = select(c).group_by(customers.c.town).having(func.count("*") > 2)

Объединения (joins)

Экземпляр Table предоставляет два метода для создания объединений (joins):

1. join() — создает внутренний join
2. outerjoin() — создает внешний join (LEFT OUTER JOIN, если точнее)

Внутренний join возвращает только те колонки, которые соответствуют условию объединения, а внешний — также некоторые дополнительные.

Оба метода принимают экземпляр Table, определяют условие объединения на основе отношений во внешних ключах и возвращают конструкцию JOIN.

>>> print(customers.join(orders))
customers JOIN orders ON customers.id = orders.customer_id

Если методы не могут определить условия объединения, или нужно предоставить другое условие, то это делается через передачу условия объединения в качестве второго аргумента.

customers.join(items,
               customers.c.address.like(customers.c.first_name + '%')
)

Когда в функции select() указываются таблицы или список колонок, SQLAlchemy автоматически размещает эти таблицы в операторе FROM. Но при использовании объединения таблицы, которые нужны во FROM, точно известны, поэтому используется select_from(). Этот же метод можно применять и для запросов, не использующих объединения. Например:

s = select([        
      customers.c.id,
      customers.c.first_name
]).select_from(customers)

print(s)
rs = conn.execute(s)
print(rs.keys())
print(rs.fetchall())

SELECT customers.id, customers.first_name 
FROM customers

RMKeyView(['id', 'first_name'])

[(1, 'Dmitriy'), (2, 'Valeriy'), (4, 'Vadim'), (5, 'Vladimir'), (6, 'Tatyana'), (7, 'Pavel')]

Используем эти знания, чтобы найти все заказы, размещенные пользователем Dmitriy Yatsenko.

select([
            orders.c.id,
            orders.c.date_placed
]).select_from(
    orders.join(customers)
).where(
    and_(
        customers.c.first_name == "Dmitriy",
        customers.c.last_name == "Yatsenko",
    )
)

Последний запрос возвращает id и date_placed заказа. Было бы неплохо также знать товары и их общее количество.

Для этого нужно сделать 3 объединения вплоть до таблицы items.

s = select([        
    orders.c.id.label('order_id'),
    orders.c.date_placed,
    order_lines.c.quantity,
    items.c.name,
            
]).select_from(
    orders.join(customers).join(order_lines).join(items)
).where(
    and_(
        customers.c.first_name == "Dmitriy",
        customers.c.last_name == "Yatsenko",
    )
)

print(s)
rs = conn.execute(s)
print(rs.keys())
print(rs.fetchall())

SELECT orders.id AS order_id, orders.date_placed, order_lines.quantity, items.name 
FROM orders JOIN customers ON customers.id = orders.customer_id JOIN order_lines ON orders.id = order_lines.order_id JOIN items ON items.id = order_lines.item_id 
WHERE customers.first_name = :first_name_1 AND customers.last_name = :last_name_1

RMKeyView(['order_id', 'date_placed', 'quantity', 'name'])

[(1, datetime.datetime(2021, 4, 22, 10, 34, 39, 548608), 5, 'Chair'), 
 (1, datetime.datetime(2021, 4, 22, 10, 34, 39, 548608), 2, 'Pen'), 
 (1, datetime.datetime(2021, 4, 22, 10, 34, 39, 548608), 1, 'Headphone'), 
 (2, datetime.datetime(2021, 4, 22, 10, 34, 39, 548608), 5, 'Chair'), 
 (2, datetime.datetime(2021, 4, 22, 10, 34, 39, 548608), 5, 'Pen')]

А вот как создавать внешнее объединение.

select([        
    customers.c.first_name,
    orders.c.id,
]).select_from(
    customers.outerjoin(orders)
)

Экземпляр Table, передаваемый в метод outerjoin(), располагается с правой стороны внешнего объединения. В результате последний запрос вернет все записи из таблицы customers (левой таблицы) и только те, которые соответствуют условию объединения из таблицы orders (правой).

Если нужны все записи из таблицы order, но лишь те, которые соответствуют условию, из orders, стоит использовать outerjoin():

select([        
    customers.c.first_name,
    orders.c.id,
]).select_from(
    orders.outerjoin(customers)
)

Также можно создать FULL OUTER JOIN, передав full=True в метод outerjoin(). Например:

select([        
    customers.c.first_name,
    orders.c.id,
]).select_from(
    orders.outerjoin(customers, full=True)
)

Обновление записей

Обновление данных выполняется с помощью функции update(). Например, следующий запрос обновляет selling_price и quantity для Water Bottle и устанавливает значения 30 и 60 соответственно.

from sqlalchemy import update

s = update(items).where(
    items.c.name == 'Water Bottle'
).values(
    selling_price = 30,
    quantity = 60,
)

print(s)
rs = conn.execute(s)

Вывод:

UPDATE items SET selling_price=:selling_price, quantity=:quantity WHERE items.name = :name_1

Удаление записей

Для удаления данных используется функция delete().

from sqlalchemy import delete

s = delete(customers).where(
    customers.c.username.like('Vladim%')
)

print(s)
rs = conn.execute(s)

Вывод:

DELETE FROM customers WHERE customers.username LIKE :username_1

Этот запрос удалит всех покупателей, чье имя пользователя начинается с Vladim.

Работа с дубликатами

Для обработки повторяющихся записей в результатах используется параметр DISTINCT. Его можно добавить в SELECT с помощью метода distinct(). Например:

# без DISTINCT

s = select([customers.c.town]).where(customers.c.id  < 10)
print(s)
rs = conn.execute(s)
print(rs.fetchall())


# с DISTINCT

s = select([customers.c.town]).where(customers.c.id  < 10).distinct()
print(s)
rs = conn.execute(s)
print(rs.fetchall())

Вывод:

SELECT customers.town 
FROM customers 
WHERE customers.id < :id_1

[(' Vladivostok',), ('Magadan',), (' Vladivostok',), (' Naberezhnye Chelny',), ('Ufa',), ('Irkutsk',)]

SELECT DISTINCT customers.town 
FROM customers 
WHERE customers.id < :id_1
[(' Vladivostok',), ('Ufa',), ('Irkutsk',), ('Magadan',), (' Naberezhnye Chelny',)]

Вот еще один пример использования distinct() с агрегирующей функцией count().Здесь считается количество уникальных городов в таблице customers.

select([        
    func.count(distinct(customers.c.town)),
    func.count(customers.c.town)
])

Конвертация данных с cast

Приведение (конвертация) данных из одного типа в другой — это распространенная операция, которая выполняется с помощью функции cast() из библиотеки sqlalchemy.

from sqlalchemy import func, cast, Date

s = select([
    cast(func.pi(), Integer),
    cast(func.pi(), Numeric(10,2)),
    cast("2010-12-01", DateTime),
    cast("2010-12-01", Date),
])

print(s)
rs = conn.execute(s)
print(rs.fetchall())

Вывод:

SELECT CAST(pi() AS INTEGER) AS pi, CAST(pi() AS NUMERIC(10, 2)) AS anon__1, CAST(:param_1 AS DATETIME) AS anon_1, CAST(:param_2 AS DATE) AS anon_2

[(3, Decimal('3.14'), datetime.datetime(2010, 12, 1, 0, 0), datetime.date(2010, 12, 1))]

Union

Оператор UNION позволяет объединять результаты нескольких SELECT. Для добавления его к функции select() используется вызов union().

from sqlalchemy import union, desc

u = union(
    select([items.c.id, items.c.name]).where(items.c.name.like("Wa%")),
    select([items.c.id, items.c.name]).where(items.c.name.like("%e%")),
).order_by(desc("id"))

print(u)
rs = conn.execute(u)
print(rs.fetchall())

Вывод:

SELECT items.id, items.name 
FROM items 
WHERE items.name LIKE :name_1 UNION SELECT items.id, items.name 
FROM items 
WHERE items.name LIKE :name_2 ORDER BY id DESC

[(8, 'Water Bottle'), (7, 'Watch'), (5, 'Keyboard'), (4, 'Travel Bag'), (3, 'Headphone'), (2, 'Pen')]

По умолчанию union() удаляет все повторяющиеся записи из результата. Для их сохранения стоит использовать union_all().

from sqlalchemy import union_all, desc

union_all(
    select([items.c.id, items.c.name]).where(items.c.name.like("Wa%")),
    select([items.c.id, items.c.name]).where(items.c.name.like("%e%")),
).order_by(desc("id"))

Создание подзапросов

Данные можно получать и из нескольких таблиц с помощью подзапросов.

Следующий запрос возвращает идентификатор и название элементов, отсортированных по Dmitriy Yatsenko в его первом заказе:

s = select([items.c.id, items.c.name]).where(
    items.c.id.in_( 
        select([order_lines.c.item_id]).select_from(customers.join(orders).join(order_lines)).where(    
                and_(
                    customers.c.first_name == 'Dmitriy',
                    customers.c.last_name == 'Yatsenko',
                    orders.c.id == 1
                )    
        )
    )
)

print(s)
rs = conn.execute(s)
print(rs.fetchall())

Вывод:

SELECT items.id, items.name 
FROM items 
WHERE items.id IN (SELECT order_lines.item_id 
FROM customers JOIN orders ON customers.id = orders.customer_id JOIN order_lines ON orders.id = order_lines.order_id 
WHERE customers.first_name = :first_name_1 AND customers.last_name = :last_name_1 AND orders.id = :id_1)

[(1, 'Chair'), (2, 'Pen'), (3, 'Headphone')]

Тот же запрос можно написать и с использованием объединений:

select([items.c.id, items.c.name]).select_from(customers.join(orders).join(order_lines).join(items)).where(    
        and_(
            customers.c.first_name == 'Dmitriy',
            customers.c.last_name == 'Yatsenko',
            orders.c.id == 1
        )    
)

«Сырые» запросы

SQLAlchemy предоставляет возможность выполнять сырые SQL-запросы с помощью функции text(). Например, следующая инструкция SELECT возвращает все заказы с товарами для Dmitriy Yatsenko.

from sqlalchemy.sql import text

s = text(
"""
SELECT
    orders.id as "Order ID", items.id, items.name
FROM
    customers
INNER JOIN orders ON customers.id = orders.customer_id
INNER JOIN order_lines ON order_lines.order_id = orders.id
INNER JOIN items ON items.id= order_lines.item_id
where customers.first_name = :first_name and customers.last_name = :last_name
"""
)

print(s)
rs = conn.execute(s, first_name='Dmitriy', last_name='Yatsenko')
print(rs.fetchall())

Вывод:

SELECT
    orders.id as "Order ID", items.id, items.name
FROM
    customers
INNER JOIN orders ON customers.id = orders.customer_id
INNER JOIN order_lines ON order_lines.order_id = orders.id
INNER JOIN items ON items.id= order_lines.item_id
where customers.first_name = :first_name and customers.last_name = :last_name

[(1, 1, 'Chair'), (1, 2, 'Pen'), (1, 3, 'Headphone'), (2, 1, 'Chair'), (2, 2, 'Pen')]

Обратите внимание на то, что инструкция включает пару связанных параметров: first_name и last_name. Сами значения для них передаются уже в метод execute().

Эту же функцию можно встроить в select(). Например:

select([items]).where(
    text("items.name like 'Wa%'")
).order_by(text("items.id desc"))

Выполнить сырой SQL можно и просто передав его прямо в execute(). Например:

rs = conn.execute("select * from  orders;")
rs.fetchall()

Транзакции

Транзакция — это способ выполнять наборы SQL-инструкций так, чтобы выполнились или все, или ни одна из них. Если хотя бы одна из инструкций, участвующих в транзакции, проходит с ошибкой, база данных возвращается к состоянию, которое было до ее начала.

Сейчас в базе данных два заказа. Для совершения заказа нужно выполнить следующие два действия:

1. Удалить заказанные товары из items
2. Обновить колонку date_shipped с датой

Оба действия должны быть выполнены как одно целое, чтобы быть уверенными в том, что данные корректные.

Объект Connection предоставляет метод begin(), который инициирует транзакцию и возвращает соответствующий объект Transaction. Последний в свою очередь предоставляет методы rollback() и commit() для отката до прежнего состояния или сохранения текущего состояния.

В следующем списке метод dispatch_order() принимает order_id в качестве аргумента и выполняет упомянутые выше действия с помощью транзакции.

from sqlalchemy import func, update
from sqlalchemy.exc import IntegrityError


def dispatch_order(order_id):
    #  проверка того, правильно ли указан order_id
    r = conn.execute(select([func.count("*")]).where(orders.c.id == order_id))
    if not r.scalar():
        raise ValueError("Недействительный order_id: {}".format(order_id))

    # брать товары в порядке очереди
    s = select([order_lines.c.item_id, order_lines.c.quantity]).where(
        order_lines.c.order_id == order_id
    )

    rs = conn.execute(s)
    ordered_items_list = rs.fetchall()

    # начало транзакции
    t = conn.begin()

    try:
        for i in ordered_items_list:
            u = update(items).where(
                items.c.id == i.item_id
            ).values(quantity = items.c.quantity - i.quantity)

            rs = conn.execute(u)

        u = update(orders).where(orders.c.id == order_id).values(date_shipped=datetime.now())
        rs = conn.execute(u)
        t.commit()
        print("Транзакция завершена.")

    except IntegrityError as e:
        print(e)
        t.rollback()
        print("Транзакция не удалась.")

dispatch_order(1)

Первый заказ включает 5 стульев и 2 ручки. Вызов функции dispatch_order() с идентификатором заказа 1 вернет такой результат.

Транзакция завершена.

Теперь items и order_lines должны выглядеть следующим образом:

В следующем заказе 5 стульев и 4 ручки, но в запасе остались лишь 1 стул и 1 ручка.

Запустим dispatch_order(2) для второго заказа.

(psycopg2.errors.CheckViolation) ОШИБКА:  новая строка в отношении "items" нарушает ограничение-проверку "quantity_check"
DETAIL:  Ошибочная строка содержит (1, Chair, 9.21, 10.81, -4).

[SQL: UPDATE items SET quantity=(items.quantity - %(quantity_1)s) WHERE items.id = %(id_1)s]
[parameters: {'quantity_1': 5, 'id_1': 1}]
(Background on this error at: http://sqlalche.me/e/14/gkpj)
Транзакция не удалась.

Выполнение закончилось с ошибкой, потому что в запасе недостаточно ручек. В итоге база данных вернулась к состоянию до начала транзакции.