SQLAlchemy Основные базовые операции: SELECT-запросы

Привет! В этом посте мы обсудим сложные SELECT-запросы, используя как SQLite, так и SQLAlchemy с Python. Наслаждайтесь!

https://t.me/sqlhub – телеграм канал обучающий работе с базами данных на практике.

Подготовка

Во-первых, давайте реализуем функции print_sqlite_cursor_as_table и print_sqlalchemy_cursor_as_table для вывода курсоров в виде красивых таблиц:

import sqlite3
import sqlalchemy

def print_formatted_table(table: list[list]):
    column_lengths = [
        max(len(table[i][column_num]) for i in range(len(table)))
        for column_num in range(len(table[0]))
    ]
    for row in table:
        print("|", end=" ")
        for item, column_length in zip(row, column_lengths):
            print(item.rjust(column_length), "|", end=" ")
        print()


def print_sqlite_cursor_as_table(title: str, cursor: sqlite3.Cursor):
    headers = [d[0] for d in cursor.description]
    
    table: list[list[str]] = [headers]
    for row in cursor.fetchall():
        table.append(list(map(str, row)))
    
    print(title)
    print_formatted_table(table)
    print()


def print_sqlalchemy_cursor_as_table(
    connection: sqlalchemy.Connection,
    title: str,
    statement: sqlalchemy.Select,
):
    cursor = connection.execute(statement)

    headers = list(cursor.keys())

    table: list[list[str]] = [headers]
    for row in cursor.fetchall():
        table.append(list(map(str, row)))
    
    print(title)
    print_formatted_table(table)
    print()

Получаем все записи из таблицы

В SQLite вы бы просто написали оператор SELECT FROM. Например, в случае с таблицей items это можно записать следующим образом:

print_sqlite_cursor_as_table(
    "All Items:",
    sqlite_cursor.execute("SELECT * FROM items")
)

>>>
All Items:
| item_id | item_name | item_price | 
|       1 |      Cola |        2.5 | 
|       2 |     Chips |        3.0 | 
|       3 |     Water |        1.2 | 
|       4 |     Apple |        0.3 | 
|       5 |   Avocado |       1.75 |

В SQLAlchemy вместо этого следует использовать функцию select и передавать таблицу в качестве параметра:

print_sqlalchemy_cursor_as_table(
    sqlalchemy_connection,
    "All items:",
    sqlalchemy.select(items_table)
)

>>>
All items:
| item_id | item_name | item_price | 
|       1 |      Cola |        2.5 | 
|       2 |     Chips |        3.0 | 
|       3 |     Water |        1.2 | 
|       4 |     Apple |        0.3 | 
|       5 |   Avocado |       1.75 |

Получаем определенные столбцы в базе и сортируем их

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

В SQLite вы можете указать имена столбцов и использовать оператор ORDER BY:

print_sqlite_cursor_as_table(
    "Item names and prices:",
    sqlite_cursor.execute(
        """
        SELECT
            item_name,
            item_price
        FROM items
        ORDER BY item_price DESC
        """
    )
)

>>>
Item names and prices:
| item_name | item_price | 
|     Chips |        3.0 | 
|      Cola |        2.5 | 
|   Avocado |       1.75 | 
|     Water |        1.2 | 
|     Apple |        0.3 |

в SQLAlchemy удобно использовать столбцы таблицы в качестве параметров функции select (вместо всей таблицы), затем вызовем метод order_by для результата и передадим параметр table_column.desc():

print_sqlalchemy_cursor_as_table(
    sqlalchemy_connection,
    "Item names and prices:",
    sqlalchemy.select(
        items_table.c.item_name,
        items_table.c.item_price
    ).order_by(items_table.c.item_price.desc())
)

>>>
Item names and prices:
| item_name | item_price | 
|     Chips |        3.0 | 
|      Cola |        2.5 | 
|   Avocado |       1.75 | 
|     Water |        1.2 | 
|     Apple |        0.3 |

Присоединяйтесь

В SQLite используется синтаксис FROM table_a JOIN table_b ON table_a.column_a = table_b.column_b:

print_sqlite_cursor_as_table(
    "Operations with details:",
    sqlite_cursor.execute(
        """
        SELECT
            o.operation_id,
            o.operation_date,
            i.item_name,
            i.item_price,
            d.amount,
            t.operation_type_name
        FROM operations o LEFT JOIN operation_descriptions d
            ON o.operation_id = d.operation_id
        JOIN items i
            ON d.item_id = i.item_id
        JOIN operation_types t
            ON o.operation_type_id = t.operation_type_id 
        """
    )
)

>>>
Operations with details:
| operation_id | operation_date | item_name | item_price | amount | operation_type_name | 
|            1 |     2022-04-22 |      Cola |        2.5 |      1 |                sell | 
|            1 |     2022-04-22 |     Chips |        3.0 |      2 |                sell | 
|            2 |     2022-04-22 |   Avocado |       1.75 |      1 |                sell | 
|            3 |     2022-04-22 |   Avocado |       1.75 |      1 |              return | 
|            4 |     2022-04-23 |     Water |        1.2 |      1 |                sell | 
|            4 |     2022-04-23 |     Apple |        0.3 |      2 |                sell |

Как повторить ту же конструкцию в SQLAlchemy? Вы можете использовать функцию table_a.join(table_b, condition).

Например, если вы выполняете INNER JOIN, это можно записать так: table_a.join(table_b, table_a.c.id = table_b.c.id).

Если вы используете LEFT JOIN, необходимо передать параметр isouter=True:

operations_details_statement = operations_table.join(
    operation_descriptions_table,
    operations_table.c.operation_id == \
        operation_descriptions_table.c.operation_id,
    isouter=True
).join(
    items_table,
    operation_descriptions_table.c.item_id == \
        items_table.c.item_id
).join(
    operation_types_table,
    operations_table.c.operation_type_id == \
        operation_types_table.c.operation_type_id
)

Здесь мы извлекаем оператор join в переменную. Чтобы использовать его в операторе select с определенными столбцами, мы можем использовать синтаксис select_from:

print_sqlalchemy_cursor_as_table(
    sqlalchemy_connection,
    "Operations with details:",
    sqlalchemy.select(
        operations_table.c.operation_id,
        operations_table.c.operation_date,
        items_table.c.item_name,
        items_table.c.item_price,
        operation_descriptions_table.c.amount,
        operation_types_table.c.operation_type_name
    ).select_from(operations_details_statement)
)

>>>
Operations with details:
| operation_id | operation_date | item_name | item_price | amount | operation_type_name | 
|            1 |     2022-04-22 |      Cola |        2.5 |      1 |                sell | 
|            1 |     2022-04-22 |     Chips |        3.0 |      2 |                sell | 
|            2 |     2022-04-22 |   Avocado |       1.75 |      1 |                sell | 
|            3 |     2022-04-22 |   Avocado |       1.75 |      1 |              return | 
|            4 |     2022-04-23 |     Water |        1.2 |      1 |                sell | 
|            4 |     2022-04-23 |     Apple |        0.3 |      2 |                sell | 

Where, Group By и функции SQL

В SQLite мы можем использовать SQL-функции для столбцов, затем отфильтровать записи с помощью некоторого условия, а затем объединить их с помощью GROUP BY. Например, если мы решили подсчитать общую сумму каждого чека, мы можем использовать такой запрос:

print_sqlite_cursor_as_table(
    "Check sums:",
    sqlite_cursor.execute(
        """
        SELECT
            o.operation_date,
            -- multiply amount of items on their price to get per-item total
            ROUND(SUM(d.amount * i.item_price), 2) AS check_sum
        FROM operations o LEFT JOIN operation_descriptions d
            ON o.operation_id = d.operation_id
        JOIN items i
            ON d.item_id = i.item_id
        JOIN operation_types t
            ON o.operation_type_id = t.operation_type_id
        WHERE t.operation_type_name = 'sell'
        GROUP BY o.operation_id, o.operation_date
        """
    )
)

>>>
Check sums:
| operation_date | check_sum | 
|     2022-04-22 |       8.5 | 
|     2022-04-22 |      1.75 | 
|     2022-04-23 |       1.8 |

В SQLAlchemy для вызова функций SQL мы используем модуль sqlalchemy.sql.func. Для фильтрации и агрегации существуют функции where(condition) и group_by(column1, column2, …), которые применяются к запросу select. Кроме того, мы можем переименовать столбец в операторе select, используя функцию label(new_label) для этого столбца. В этом запросе повторно используем переменную operations_details_statement:

from sqlalchemy.sql import func

print_sqlalchemy_cursor_as_table(
    sqlalchemy_connection,
    "Check sums:",
    sqlalchemy.select(
        operations_table.c.operation_date,
        func.round(
            func.sum(
                operation_descriptions_table.c.amount * \
                    items_table.c.item_price
            ),
            2
        ).label("check_sum")
    ).select_from(
        operations_details_statement
    ).where(
        operation_types_table.c.operation_type_name == "sell"
    ).group_by(
        operations_table.c.operation_id,
        operations_table.c.operation_date
    )
)

>>>
Check sums:
| operation_date | check_sum | 
|     2022-04-22 |       8.5 | 
|     2022-04-22 |      1.75 | 
|     2022-04-23 |       1.8 | 

CTE и подзапросы

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

Например, давайте найдем все чеки, у которых общая сумма больше, чем средняя, используя SQLite:

print_sqlite_cursor_as_table(
    "Check greater than average:",
    sqlite_cursor.execute(
        """
        WITH full_checks AS (
            SELECT
                o.operation_id,
                o.operation_date,
                i.item_name,
                d.amount,
                ROUND(d.amount * i.item_price, 2) AS item_sum
            FROM operations o LEFT JOIN operation_descriptions d
                ON o.operation_id = d.operation_id
            JOIN items i
                ON d.item_id = i.item_id
            JOIN operation_types t
                ON o.operation_type_id = t.operation_type_id
            WHERE t.operation_type_name = 'sell'
        ), sell_checks AS (
            SELECT
                f.operation_id,
                f.operation_date,
                ROUND(SUM(f.item_sum), 2) AS check_sum
            FROM full_checks f
            GROUP BY f.operation_id, f.operation_date
        )
        
        SELECT
            f.operation_id,
            f.operation_date,
            f.item_name,
            f.item_sum
        FROM sell_checks s LEFT JOIN full_checks f
            ON s.operation_id = f.operation_id
        WHERE s.check_sum > (SELECT AVG(check_sum) FROM sell_checks)
        """
    )
)

>>>
Check greater than average:
| operation_id | operation_date | item_name | item_sum | 
|            1 |     2022-04-22 |     Chips |      6.0 | 
|            1 |     2022-04-22 |      Cola |      2.5 | 

Альтернативно, в SQLAlchemy мы можем вызвать метод cte(cte_name) на запросе, чтобы превратить его в CTE:

full_checks_cte = sqlalchemy.select(
    operations_table.c.operation_id,
    operations_table.c.operation_date,
    items_table.c.item_name,
    operation_descriptions_table.c.amount,
    func.round(
        operation_descriptions_table.c.amount * items_table.c.item_price,
        2
    ).label("item_sum")
).select_from(
    operations_details_statement
).where(
    operation_types_table.c.operation_type_name == "sell"
).cte("full_checks")


sell_checks_cte = sqlalchemy.select(
    full_checks_cte.c.operation_id,
    full_checks_cte.c.operation_date,
    func.round(
        func.sum(full_checks_cte.c.item_sum),
        2
    ).label("check_sum")
).select_from(
    full_checks_cte
).group_by(
    full_checks_cte.c.operation_id,
    full_checks_cte.c.operation_date
).cte("sell_checks")

Позже мы можем использовать эти CTE в запросе select. Кроме того, SQLAlchemy предупреждает нас об использовании функции scalar_subquery() в подзапросах для преобразования их в скалярные значения:

print_sqlalchemy_cursor_as_table(
    sqlalchemy_connection,
    "Check greater than average:",
    sqlalchemy.select(
        full_checks_cte.c.operation_id,
        full_checks_cte.c.operation_date,
        full_checks_cte.c.item_name,
        full_checks_cte.c.item_sum
    ).select_from(
        sell_checks_cte.join(
            full_checks_cte,
            sell_checks_cte.c.operation_id == full_checks_cte.c.operation_id,
            isouter=True
        )
    ).where(
        sell_checks_cte.c.check_sum > sqlalchemy.select(
            func.avg(sell_checks_cte.c.check_sum)
        ).scalar_subquery()
    )
)

>>>
Check greater than average:
| operation_id | operation_date | item_name | item_sum | 
|            1 |     2022-04-22 |     Chips |      6.0 | 
|            1 |     2022-04-22 |      Cola |      2.5 |

Заключение

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