SQLAlchemy: как подключить Python к базе данных и вытащить данные в pa

Как аналитику вытащить данные из БД в pandas?

Коротко: создаёшь engine через create_engine() со строкой подключения к своей базе, а дальше отдаёшь запрос в pd.read_sql_query() — и получаешь готовый DataFrame. Три строки, и данные из Postgres уже лежат в памяти Python:

import pandas as pd
from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg2://analyst:pass@10.0.0.5:5432/shop")
df = pd.read_sql_query(text("SELECT * FROM orders WHERE status = 'paid'"), engine)

Всё остальное — это нюансы, которые отличают скрипт «работает на моей машине» от скрипта, который не роняет прод и не течёт памятью. Разберу их по порядку: как устроена строка подключения, почему нельзя клеить запросы f-строками, что делать, когда таблица не влезает в оперативку, и зачем аналитику вообще думать про пул соединений. Я почти всегда работаю через SQLAlchemy Core, а не ORM, и ниже объясню почему.

Что такое engine и как выглядит строка подключения?

Строка подключения (её называют DSN или connection URL) читается как адрес:

диалект+драйвер://пользователь:пароль@хост:порт/имя_базы

Несколько живых примеров:

# PostgreSQL через psycopg2
create_engine("postgresql+psycopg2://analyst:pass@db.internal:5432/shop")

# PostgreSQL через psycopg (версия 3)
create_engine("postgresql+psycopg://analyst:pass@db.internal:5432/shop")

# MySQL
create_engine("mysql+pymysql://analyst:pass@db.internal:3306/shop")

# ClickHouse (нужен пакет clickhouse-sqlalchemy)
create_engine("clickhouse+native://analyst:pass@ch.internal:9000/analytics")

# Локальный SQLite-файл
create_engine("sqlite:///local_cache.db")

Драйвер (psycopg2, pymysql и т.д.) — это отдельный пакет, который надо поставить через pip. Сам SQLAlchemy умеет говорить с базой только через него. Если забыл — получишь ModuleNotFoundError: No module named 'psycopg2' ещё до попытки подключиться.

Пароль в коде я не храню никогда. Достаю из переменных окружения:

import os
from sqlalchemy import create_engine

url = f"postgresql+psycopg2://{os.environ['DB_USER']}:{os.environ['DB_PASS']}@{os.environ['DB_HOST']}:5432/shop"
engine = create_engine(url, pool_pre_ping=True)

Если в пароле есть спецсимволы (@, :, /), их надо экранировать через urllib.parse.quote_plus, иначе URL распарсится криво и ты полчаса будешь думать, почему «неверный хост».

Core или ORM — что нужно аналитику?

У SQLAlchemy два слоя. ORM (Object-Relational Mapping) — это когда строки таблицы превращаются в объекты Python: классы, атрибуты, session.query(User).filter(...). Он хорош для бэкенда, где ты создаёшь и обновляешь сущности. Core — это тонкая обёртка над обычным SQL: ты пишешь SELECT, а библиотека даёт соединение, экранирование параметров и пул.

Аналитику в 99% случаев нужен Core. Мы читаем данные, а не управляем жизненным циклом объектов. Нам не нужны классы-модели, отложенная загрузка связей и прочая машинерия ORM — она только добавит зависимостей и точек отказа. Пара engine + text() + pd.read_sql_query() покрывает почти все задачи выгрузки. Держать в голове ORM стоит, только если ты пишешь ещё и продуктовую часть.

Если хочешь освежить сам SQL перед тем, как оборачивать его в Python, у нас есть интерактивный SQL-тренажёр с настоящим PostgreSQL в браузере и справочник по SQL под рукой.

Как правильно читать данные в pandas?

У pandas есть три родственные функции. Разница между ними небольшая, но знать стоит:

• pd.read_sql_query() — принимает текст SELECT-запроса. Это мой основной инструмент.
• pd.read_sql_table() — читает таблицу целиком по имени (работает только через SQLAlchemy, без сырого SQL).
• pd.read_sql() — универсальная обёртка, которая сама решает, что ей передали: запрос или имя таблицы.

Базовый вызов выглядит так:

query = text("""
    SELECT
        date_trunc('day', created_at) AS d,
        count(*) AS orders,
        sum(amount) AS revenue
    FROM orders
    WHERE status = 'paid'
    GROUP BY 1
    ORDER BY 1
""")
df = pd.read_sql_query(query, engine, parse_dates=["d"])

Пара полезных аргументов, которые я использую постоянно. parse_dates заставляет pandas сразу распарсить колонки в datetime, иначе даты часто приезжают строками. index_col сделает нужную колонку индексом DataFrame. А dtype (в свежих версиях pandas) позволяет задать типы явно, чтобы id не превратился в float из-за одного NULL.

Обрати внимание на text(). В SQLAlchemy 2.0 сырую строку SQL надо оборачивать в text() явно, иначе получишь ошибку вроде Textual SQL expression should be explicitly declared as text(...). Это не каприз — так библиотека отделяет «доверенный шаблон запроса» от «данных снаружи», и это ровно то, что защищает нас в следующем разделе.

Потренироваться писать такие запросы и сразу гонять их через pandas можно в Python-тренажёре, а типовые аналитические расчёты вроде DAU удобно собирать сразу в связке SQL + pandas.

Зачем параметризовать запросы вместо f-строк?

Потому что f-строка в SQL — это дыра размером с грузовик и источник тупых багов на ровном месте. Сравни. Плохо:

user_id = request_param  # пришло снаружи
df = pd.read_sql_query(
    text(f"SELECT * FROM events WHERE user_id = {user_id}"), engine
)

Если в user_id окажется строка 0 OR 1=1, ты выгрузишь всю таблицу. А если туда прилетит '; DROP TABLE events; --, то потеряешь таблицу. Это классическая SQL-инъекция. Даже когда данные «внутренние», f-строки ломаются на кавычках, датах и NULL.

Правильно — передавать значения отдельно, через именованные параметры:

query = text("""
    SELECT user_id, event_type, created_at
    FROM events
    WHERE user_id = :uid
      AND created_at >= :start
      AND event_type = ANY(:types)
""")
df = pd.read_sql_query(
    query,
    engine,
    params={
        "uid": 42,
        "start": "2026-01-01",
        "types": ["login", "purchase"],
    },
)

Двоеточие перед именем (:uid, :start) — это плейсхолдер. SQLAlchemy вместе с драйвером подставит значения безопасно: экранирует, приведёт типы, сам разберётся с датами и списками. Плюс тут не только безопасность — код становится читаемым и переиспользуемым. Один и тот же query гоняешь с разными params в цикле по клиентам или по дням, не собирая строку заново.

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

Что делать с выгрузкой, которая не влезает в память?

Читать её кусками через chunksize. Когда таблица на десятки миллионов строк, наивный pd.read_sql_query() попытается затащить всё в оперативку разом и упадёт с MemoryError (или тихо съест весь swap и повесит машину). Аргумент chunksize превращает функцию в генератор: она отдаёт DataFrame'ы по N строк, а ты обрабатываешь их по одному и держишь в памяти только текущий кусок.

query = text("SELECT user_id, amount FROM payments WHERE created_at >= :start")

total_by_user = pd.Series(dtype="float64")
for chunk in pd.read_sql_query(query, engine, params={"start": "2026-01-01"},
                               chunksize=100_000):
    part = chunk.groupby("user_id")["amount"].sum()
    total_by_user = total_by_user.add(part, fill_value=0)

result = total_by_user.sort_values(ascending=False)

Логика простая: считаем агрегат по каждому куску и складываем промежуточные результаты. По памяти держим сто тысяч строк вместо десятков миллионов.

Но честный совет: если тебе нужен агрегат, а не сырые строки, — считай его на стороне базы. GROUP BY в SQL почти всегда быстрее и дешевле, чем тащить всё в Python и группировать в pandas. chunksize нужен, когда сырые строки действительно надо обработать построчно (сложная логика, обогащение через API, запись в другое хранилище) и они не помещаются в память. Для «посчитать сумму по клиентам» правильнее написать это в SQL — потренироваться в таких агрегатах можно на практических заданиях.

Нужен ли аналитику пул соединений?

Пул у тебя уже есть — engine создаёт его по умолчанию, даже если ты об этом не думал. Вопрос в том, надо ли его настраивать. Для разового скрипта-выгрузки — почти нет. Для долгоживущего процесса (дашборд, планировщик, сервис, который дёргает базу весь день) — да, и вот главные параметры:

engine = create_engine(
    url,
    pool_size=5,        # сколько соединений держим открытыми постоянно
    max_overflow=10,    # сколько ещё можно открыть на пике
    pool_recycle=1800,  # пересоздавать соединение раз в 30 минут
    pool_pre_ping=True, # проверять живость соединения перед выдачей
)

Два параметра спасают чаще всего. pool_pre_ping=True делает лёгкий пинг перед тем, как отдать соединение из пула, — это лечит вечную боль «server closed the connection unexpectedly», когда база или файрвол закрыли простаивающее подключение, а пул об этом не знает. pool_recycle заставляет пересоздавать соединения по таймеру, что нужно для MySQL с его wait_timeout и вообще для любой базы за прокси, который рубит долгие коннекты.

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

with engine.connect() as conn:
    df = pd.read_sql_query(query, conn)
# соединение вернулось в пул автоматически

Какие ошибки чаще всего ломают подключение?

Собрал те, на которые сам натыкался и которые вижу у коллег:

• Забытый драйвер. postgresql://... без +psycopg2 иногда работает, иногда нет — пиши диалект с драйвером явно, чтобы не гадать.
• create_engine() в цикле. Каждый вызов поднимает новый пул. Создавай engine один раз, переиспользуй везде.
• Сырой SQL без text() в SQLAlchemy 2.0. Даёт ошибку про «explicitly declared as text». Оборачивай любой строковый запрос.
• F-строки вместо params. Инъекции, поломка на кавычках и датах, нечитаемый код. Всегда именованные параметры.
• Огромная таблица без chunksize. MemoryError или повисшая машина. Агрегируй в SQL или читай кусками.
• Пароль в коде. Достаём из окружения; спецсимволы — через quote_plus.
• Смешение диалектов. То, что работает в SQLite (strftime, || для конкатенации), в Postgres выглядит иначе (date_trunc, concat). Проверяй запрос против той базы, откуда реально тянешь.
• Отсутствие pool_pre_ping в долгом процессе. Ловишь «connection closed» на ровном месте после простоя.
• Данные приезжают строками. Даты и числа не распарсились — задавай parse_dates и при необходимости dtype.

С чего начать на практике?

Минимальный рабочий цикл аналитика такой: один раз собрал engine из строки подключения, держишь его весь сеанс; запросы пишешь через text() с именованными параметрами; читаешь в pandas через pd.read_sql_query(), а для тяжёлых выгрузок добавляешь chunksize; агрегаты по возможности считаешь в самой базе. Core вместо ORM, пароли из окружения, pool_pre_ping для долгих процессов — этого набора хватает почти на любую задачу выгрузки.

Дальше стоит закрепить руками: пройди базовый курс по SQL, чтобы запросы под pandas были осмысленными, потренируй Python-часть в тренажёре и держи под рукой справочник по Python. А когда захочешь снять все лимиты — открытый доступ ко всем задачам, кейсам и безлимитному AI-собеседованию есть в Pro-подписке. Она окупается на первом же оффере, где тебя попросят «вытащить данные и посчитать».