Эффективное взаимодействие с базами данных в Python: от основ до ORM

import sqlite3

# Создаем подключение к базе данных (или создаем новую, если файла нет)
conn = sqlite3.connect('example.db')

# Создаем курсор для выполнения SQL-запросов
cursor = conn.cursor()

# Выполняем запрос
cursor.execute('SELECT SQLITE_VERSION()')
version = cursor.fetchone()
print(f"SQLite version: {version[0]}")

# Не забываем закрыть подключение
conn.close()

import pymysql

# Установка соединения
conn = pymysql.connect(
host='localhost',
user='username',
password='password',
database='database_name',
charset='utf8mb4'
)

try:
with conn.cursor() as cursor:
# Выполняем запрос
cursor.execute('SELECT VERSION()')
version = cursor.fetchone()
print(f"MySQL version: {version[0]}")

# Фиксируем изменения
conn.commit()
finally:
# Закрываем соединение
conn.close()

import psycopg2

# Установка соединения
conn = psycopg2.connect(
host='localhost',
user='username',
password='password',
dbname='database_name'
)

# Создаем курсор с использованием контекстного менеджера
with conn:
with conn.cursor() as cursor:
# Выполняем запрос
cursor.execute('SELECT version()')
version = cursor.fetchone()
print(f"PostgreSQL version: {version[0]}")

# Соединение закрывается автоматически благодаря контекстному менеджеру

from sqlalchemy import create_engine

# Строки подключения для разных баз данных
sqlite_uri = 'sqlite:///example.db'
mysql_uri = 'mysql+pymysql://username:password@localhost/database_name'
postgres_uri = 'postgresql+psycopg2://username:password@localhost/database_name'

# Выбираем нужную строку подключения
engine = create_engine(sqlite_uri)

# Выполняем запрос
with engine.connect() as connection:
result = connection.execute('SELECT 1')
print(f"Connected successfully: {result.fetchone()[0]}")

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# Создание таблицы
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE NOT NULL,
age INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')

# Добавление индекса
cursor.execute('CREATE INDEX IF NOT EXISTS idx_username ON users(username)')

# Фиксация изменений
conn.commit()
conn.close()

from sqlalchemy import Column, Integer, String, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func

# Создаем базовый класс для моделей
Base = declarative_base()

# Определяем модель (таблицу)
class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(100), unique=True, nullable=False)
age = Column(Integer)
created_at = Column(DateTime, default=func.now())

def __repr__(self):
return f"<User(username='{self.username}', email='{self.email}')>"

# Создаем engine и таблицы
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

# Создаем сессию
Session = sessionmaker(bind=engine)
session = Session()

# Теперь можно использовать сессию для операций с базой данных

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# Добавление новой колонки
try:
cursor.execute('ALTER TABLE users ADD COLUMN last_login TIMESTAMP')
print("Column added successfully")
except sqlite3.OperationalError as e:
# SQLite имеет ограниченную поддержку ALTER TABLE
print(f"Error: {e}")

conn.commit()
conn.close()

# Инициализация Alembic (выполняется один раз)
# $ alembic init migrations

# В файле alembic.ini устанавливаем URL базы данных
# sqlalchemy.url = sqlite:///example.db

# Создание новой миграции
# $ alembic revision --autogenerate -m "Add last_login column"

# Применение миграций
# $ alembic upgrade head

import sqlite3

conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# CREATE – Создание новой записи
def create_user(username, email, age):
cursor.execute(
'INSERT INTO users (username, email, age) VALUES (?, ?, ?)',
(username, email, age)
)
conn.commit()
return cursor.lastrowid

# READ – Чтение данных
def get_user(user_id):
cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
return cursor.fetchone()

def get_all_users():
cursor.execute('SELECT * FROM users')
return cursor.fetchall()

# UPDATE – Обновление данных
def update_user(user_id, username=None, email=None, age=None):
# Формируем части запроса динамически на основе переданных аргументов
updates = []
params = []

if username is not None:
updates.append('username = ?')
params.append(username)

if email is not None:
updates.append('email = ?')
params.append(email)

if age is not None:
updates.append('age = ?')
params.append(age)

if not updates:
return False # Нечего обновлять

# Добавляем id в список параметров
params.append(user_id)

# Формируем и выполняем запрос
query = f"UPDATE users SET {', '.join(updates)} WHERE id = ?"
cursor.execute(query, params)
conn.commit()
return cursor.rowcount > 0

# DELETE – Удаление данных
def delete_user(user_id):
cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
conn.commit()
return cursor.rowcount > 0

# Примеры использования
user_id = create_user('john_doe', 'john@example.com', 30)
print(f"Created user with ID: {user_id}")

user = get_user(user_id)
print(f"User data: {user}")

updated = update_user(user_id, age=31)
print(f"Update successful: {updated}")

deleted = delete_user(user_id)
print(f"Deletion successful: {deleted}")

# Закрываем соединение
conn.close()

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func

Base = declarative_base()

class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(100), unique=True, nullable=False)
age = Column(Integer)
created_at = Column(DateTime, default=func.now())

def __repr__(self):
return f"<User(username='{self.username}', email='{self.email}')>"

# Создаем engine и сессию
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# CREATE – Создание новой записи
def create_user(username, email, age):
user = User(username=username, email=email, age=age)
session.add(user)
session.commit()
return user.id

# READ – Чтение данных
def get_user(user_id):
return session.query(User).filter(User.id == user_id).first()

def get_all_users():
return session.query(User).all()

# UPDATE – Обновление данных
def update_user(user_id, username=None, email=None, age=None):
user = get_user(user_id)
if not user:
return False

if username is not None:
user.username = username

if email is not None:
user.email = email

if age is not None:
user.age = age

session.commit()
return True

# DELETE – Удаление данных
def delete_user(user_id):
user = get_user(user_id)
if not user:
return False

session.delete(user)
session.commit()
return True

# Примеры использования
user_id = create_user('jane_doe', 'jane@example.com', 28)
print(f"Created user with ID: {user_id}")

user = get_user(user_id)
print(f"User data: {user}")

updated = update_user(user_id, age=29)
print(f"Update successful: {updated}")

deleted = delete_user(user_id)
print(f"Deletion successful: {deleted}")

# Создание индекса для часто запрашиваемых полей
cursor.execute('CREATE INDEX idx_username ON users(username)')

# Использование индекса в запросе
cursor.execute('SELECT * FROM users WHERE username = ?', ('john_doe',))

# Вместо этого:
cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))

# Используйте это:
cursor.execute

**Читайте также**
- [Обработка исключений в Python: как защитить код от неожиданных ошибок](/python/obrabotka-isklyuchenij-v-python-osnovy-i-luchshie-praktiki/)
- [Функции в Python: от основ до продвинутых техник программирования](/python/funkcii-v-python-opredelenie-i-ispolzovanie/)
- [Модульное тестирование в Python: защита кода от скрытых ошибок](/python/testirovanie-koda-na-python-modulnoe-testirovanie-i-instrumenty/)
- [Работа с файлами в Python: основы и эффективные практики кодирования](/python/rabota-s-fajlami-v-python-chtenie-i-zapis/)
- [Условные операторы и циклы в Python: основы для новичков](/python/uslovnye-konstrukcii-i-cikly-v-python/)
- [Изучение Python: путь от новичка до профессионала за 5 шагов](/python/kak-stat-horoshim-programmistom-na-python/)
- [Коллекции Python: от списков до словарей, основы и продвинутые техники](/python/rabota-s-kollekciyami-v-python-spiski-kortezhi-slovari-i-mnozhestva/)
- [Бесплатные курсы Python: ТОП-15 ресурсов для будущих разработчиков](/python/luchshie-besplatnye-kursy-dlya-izucheniya-python/)
- [Python для начинающих: от первого кода к практическим навыкам](/python/python-dlya-nachinayushih-pervye-shagi-v-programmirovanii/)
- [Python: история языка от эксперимента до лидера программирования](/python/python-vvedenie-i-istoriya-sozdaniya/)