Основы работы с SQL в языке Python: от SQLite до PostgreSQL

Раздел: Работа с данными -> SQLite и PostgreSQL

Основы работы с SQL в Python: SQLite и PostgreSQL

При разработке приложений часто требуется взаимодействие с реляционными базами данных. Два популярных варианта - лёгкая встраиваемая SQLite и полнофункциональная серверная PostgreSQL. В Python существуют стандартные и сторонние библиотеки для подключения к ним. Главные цели - безопасность (защита от SQL-инъекций), надёжность (управление соединениями и транзакциями) и производительность.

Наиболее эффективное решение: контекстные менеджеры и параметризованные запросы

Рекомендуемый подход - использование контекстного менеджера для автоматического закрытия соединения и курсора, а также передача параметров в виде кортежа или словаря. Это единообразно работает для SQLite (модуль sqlite3) и для PostgreSQL (модуль psycopg2).

Пример для SQLite

import sqlite3

try:
    with sqlite3.connect('example.db') as conn:
        cursor = conn.cursor()
        cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)')
        cursor.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30))
        conn.commit()
        cursor.execute('SELECT * FROM users')
        print(cursor.fetchall())
except sqlite3.Error as e:
    print(f'Ошибка: {e}')

Sql server python (sql server и python)

[(1, 'Alice', 30)]

Sql через python (работа с sql через python)

Пример для PostgreSQL

import psycopg2

try:
    with psycopg2.connect(
        dbname='testdb', user='user', password='pass', host='localhost'
    ) as conn:
        with conn.cursor() as cur:
            cur.execute('CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name VARCHAR(100), age INTEGER)')
            cur.execute('INSERT INTO users (name, age) VALUES (%s, %s)', ('Bob', 25))
            conn.commit()
            cur.execute('SELECT * FROM users')
            print(cur.fetchall())
except psycopg2.Error as e:
    print(f'Ошибка: {e}')

Python postgresql (подключение к postgresql из python)

[(1, 'Bob', 25)]

Python база данных таблицу (создание таблицы в базе данных python)

Пояснения шагов:

  • Соединение создаётся через connect() с указанием параметров подключения.
  • Контекстный менеджер with гарантирует вызов close() при выходе из блока, даже при исключении.
  • Курсор создаётся аналогично - с помощью with conn.cursor() для автоматического освобождения ресурсов.
  • Запросы с параметрами используют плейсхолдеры: ? для sqlite3, %s для psycopg2. Значения передаются вторым аргументом execute().
  • После операций изменения данных обязателен вызов conn.commit().

Типичные ошибки и их решения:

  • Забытый commit - изменения не сохраняются. Решение: всегда вызывать conn.commit() после INSERT/UPDATE/DELETE или включить автокоммит.
  • Неправильные типы параметров - например, лишние кавычки. Параметризация решает проблему.
  • Утечка соединений - при отсутствии контекстного менеджера требуется явный вызов conn.close(). Контекстный менеджер исключает это.
  • Блокировка базы данных (SQLite) - при параллельных записях. Можно использовать WAL-режим: conn.execute('PRAGMA journal_mode=WAL').

Различные варианты реализации

Как быстро вставить значение в запрос, не заботясь о безопасности?

Иногда разработчики используют форматирование строк для подстановки значений. Это упрощает запись, но создаёт уязвимость к SQL-инъекциям.

name = "Alice"
age = 30
cursor.execute(f"INSERT INTO users (name, age) VALUES ('{name}', {age})")

библиотека sql python (библиотека для работы с sql в python)

Цель:

показать, как не надо делать.

Проблема: если name содержит "'); DROP TABLE users; --", то выполнится зловредный код. Решение - всегда использовать параметризацию.

Как абстрагироваться от конкретной СУБД и упростить работу с моделями?

ORM SQLAlchemy предоставляет объектно-реляционное отображение, позволяя работать с таблицами как с классами. Код становится универсальным.

from sqlalchemy import create_engine, Column, Integer, String, select
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

with Session(engine) as session:
    new_user = User(name='Charlie', age=35)
    session.add(new_user)
    session.commit()
    stmt = select(User).where(User.name == 'Charlie')
    for user in session.execute(stmt).scalars():
        print(user.id, user.name, user.age)

создание sql python (работа с sql в python)

1 Charlie 35

Цель - снизить объём ручного SQL-кода и обеспечить переносимость.

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

Как выполнять SQL запросы асинхронно для неблокирующего ввода-вывода?

Библиотеки aiosqlite и asyncpg позволяют работать с базами данных в асинхронном коде (asyncio). Это полезно в веб-приложениях.

import asyncio
import aiosqlite

async def main():
    async with aiosqlite.connect('example.db') as db:
        cursor = await db.execute('SELECT * FROM users')
        rows = await cursor.fetchall()
        print(rows)

asyncio.run(main())

Для PostgreSQL аналогично с asyncpg.

Ошибки: неправильное использование await, смешивание синхронного и асинхронного кода. Следует придерживаться одного стиля.

Как повторно использовать соединения для повышения производительности?

Пулы соединений (например, psycopg2.pool.ThreadedConnectionPool) позволяют не создавать новое подключение для каждого запроса.

from psycopg2 import pool

connection_pool = pool.SimpleConnectionPool(1, 10, dbname='testdb', user='user', password='pass', host='localhost')
conn = connection_pool.getconn()
cursor = conn.cursor()
cursor.execute('SELECT 1')
print(cursor.fetchone())
connection_pool.putconn(conn)

Цель: уменьшить накладные расходы на установку соединения.

Проблемы: необходимо корректно возвращать соединение в пул; при разрыве соединения требуется проверка conn.closed.

Как выполнять SQL операции без написания запросов?

Библиотека dataset (надстройка над SQLAlchemy) предоставляет простой интерфейс для работы с таблицами как со словарями.

import dataset

db = dataset.connect('sqlite:///example.db')
table = db['users']
table.insert(dict(name='Diana', age=28))
print(list(table.find(name='Diana')))

Цель: быстрый прототип, простые скрипты.

Ограничения: меньше контроля над запросами, сложнее оптимизировать производительность.

Расширенные примеры

Работа с транзакциями и точками сохранения (savepoints)

Пример 
import sqlite3

conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE test (id INTEGER PRIMARY KEY, val TEXT)')
try:
    conn.execute('BEGIN')
    conn.execute('INSERT INTO test VALUES (1, "first")')
    conn.execute('SAVEPOINT sp1')
    conn.execute('INSERT INTO test VALUES (2, "second")')
    conn.execute('ROLLBACK TO sp1')
    conn.execute('INSERT INTO test VALUES (3, "third")')
    conn.execute('COMMIT')
except Exception as e:
    conn.execute('ROLLBACK')
    print(f'Ошибка: {e}')
finally:
    cur = conn.execute('SELECT * FROM test')
    print(cur.fetchall())
    conn.close()
[(1, 'first'), (3, 'third')]

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

Массовая вставка с executemany

Пример 
import sqlite3

data = [('user1', 20), ('user2', 30), ('user3', 40)]
conn = sqlite3.connect(':memory:')
conn.execute('CREATE TABLE users (name TEXT, age INTEGER)')
conn.executemany('INSERT INTO users VALUES (?, ?)', data)
conn.commit()
cur = conn.execute('SELECT * FROM users')
print(cur.fetchall())
[('user1', 20), ('user2', 30), ('user3', 40)]

executemany эффективен для вставки больших объёмов данных за один запрос.

Работа с JSON в PostgreSQL (адаптер Json)

Пример 
import psycopg2
from psycopg2.extras import Json

conn = psycopg2.connect(...)
cur = conn.cursor()
cur.execute('CREATE TABLE IF NOT EXISTS events (id SERIAL, data JSONB)')
cur.execute('INSERT INTO events (data) VALUES (%s)', (Json({'key': 'value'}),))
conn.commit()
cur.execute('SELECT data FROM events')
print(cur.fetchone()[0])
{'key': 'value'}

Адаптер Json автоматически преобразует словарь Python в JSON для БД.

Подготовленные операторы в psycopg2

Пример 
from psycopg2 import sql

cur = conn.cursor()
query = sql.SQL("INSERT INTO users (name, age) VALUES ({name}, {age})").format(
    name=sql.Identifier('test_name'),
    age=sql.Literal(99)
)
cur.execute(query)
conn.commit()

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

Оптимизация SQLite через PRAGMA

Пример 
conn = sqlite3.connect('example.db')
conn.execute('PRAGMA journal_mode=WAL')  # улучшает многопоточность
conn.execute('PRAGMA synchronous=OFF')   # увеличивает скорость записи
conn.execute('PRAGMA cache_size=-2000')  # 2МБ кеш

Эти настройки полезны при высокой нагрузке на запись.

Работа с SQL в Python - comments

En
создание sql python (python)