Python и базы данных: практическое руководство для разработчиков

import sqlite3

# Установка соединения
conn = sqlite3.connect('example.db')

# Создание курсора
cursor = conn.cursor()

# Выполнение запроса
cursor.execute('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)')

# Добавление данных
cursor.execute('INSERT INTO users (name, email) VALUES (?, ?)', ('John Doe', 'john@example.com'))

# Применение изменений
conn.commit()

# Получение данных
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
print(users)

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

import psycopg2

conn = psycopg2.connect(
dbname="testdb",
user="postgres",
password="postgres",
host="localhost",
port="5432"
)

cursor = conn.cursor()
cursor.execute("SELECT version();")
db_version = cursor.fetchone()
print(f"PostgreSQL version: {db_version}")

conn.close()

import mysql.connector

conn = mysql.connector.connect(
host="localhost",
user="root",
password="root",
database="testdb"
)

cursor = conn.cursor()
cursor.execute("SELECT VERSION()")
data = cursor.fetchone()
print(f"MySQL version: {data[0]}")

conn.close()

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['testdb']
collection = db['users']

user = {"name": "John Doe", "email": "john@example.com"}
user_id = collection.insert_one(user).inserted_id
print(f"Inserted user with ID: {user_id}")

users = collection.find()
for user in users:
print(user)

client.close()

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

# Создание движка для подключения к БД
engine = create_engine('sqlite:///example.db')

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

# Определение моделей (классов)
class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String, unique=True)
posts = relationship("Post", back_populates="author")

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

class Post(Base):
__tablename__ = 'posts'

id = Column(Integer, primary_key=True)
title = Column(String)
content = Column(String)
user_id = Column(Integer, ForeignKey('users.id'))
author = relationship("User", back_populates="posts")

def __repr__(self):
return f"<Post(title='{self.title}')>"

# Создание таблиц в БД
Base.metadata.create_all(engine)

# Создание сессии для работы с БД
Session = sessionmaker(bind=engine)
session = Session()

# Создание нового пользователя
new_user = User(name="Alice", email="alice@example.com")
session.add(new_user)
session.commit()

# Создание поста для пользователя
new_post = Post(title="My First Post", content="Hello, World!", author=new_user)
session.add(new_post)
session.commit()

# Получение всех пользователей
users = session.query(User).all()
for user in users:
print(user)
print("Posts:", user.posts)

import sqlite3

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

# Создание таблицы
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
category TEXT
)
''')

# Вставка данных
cursor.execute('''
INSERT INTO products (name, price, category) VALUES (?, ?, ?)
''', ('Laptop', 999.99, 'Electronics'))

# Вставка множества записей
products = [
('Smartphone', 499.99, 'Electronics'),
('Headphones', 99.99, 'Accessories'),
('Mouse', 29.99, 'Accessories')
]
cursor.executemany('INSERT INTO products (name, price, category) VALUES (?, ?, ?)', products)

conn.commit()
conn.close()

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

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

class Product(Base):
__tablename__ = 'products'

id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
price = Column(Float, nullable=False)
category = Column(String)

def __repr__(self):
return f"<Product(name='{self.name}', price={self.price})>"

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Добавление одной записи
laptop = Product(name='Laptop', price=999.99, category='Electronics')
session.add(laptop)

# Добавление множества записей
products = [
Product(name='Smartphone', price=499.99, category='Electronics'),
Product(name='Headphones', price=99.99, category='Accessories'),
Product(name='Mouse', price=29.99, category='Accessories')
]
session.add_all(products)

session.commit()

import sqlite3

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

# Чтение всех записей
cursor.execute('SELECT * FROM products')
all_products = cursor.fetchall()
for product in all_products:
print(product)

# Чтение с фильтрацией
cursor.execute('SELECT * FROM products WHERE category = ?', ('Electronics',))
electronics = cursor.fetchall()
print("\nElectronics products:")
for product in electronics:
print(product)

# Чтение с сортировкой
cursor.execute('SELECT * FROM products ORDER BY price DESC')
sorted_products = cursor.fetchall()
print("\nProducts sorted by price (descending):")
for product in sorted_products:
print(product)

conn.close()

from sqlalchemy import create_engine, desc
from sqlalchemy.orm import sessionmaker

# Используем модель Product из предыдущего примера
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# Чтение всех записей
all_products = session.query(Product).all()
for product in all_products:
print(product)

# Чтение с фильтрацией
electronics = session.query(Product).filter(Product.category == 'Electronics').all()
print("\nElectronics products:")
for product in electronics:
print(product)

# Чтение с сортировкой
sorted_products = session.query(Product).order_by(desc(Product.price)).all()
print("\nProducts sorted by price (descending):")
for product in sorted_products:
print(product)

import sqlite3

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

# Обновление одной записи
cursor.execute('''
UPDATE products SET price = ? WHERE name = ?
''', (899.99, 'Laptop'))

# Обновление нескольких записей
cursor.execute('''
UPDATE products SET category = ? WHERE category = ?
''', ('Computer Accessories', 'Accessories'))

conn.commit()

# Проверка результатов
cursor.execute('SELECT * FROM products')
updated_products = cursor.fetchall()
for product in updated_products:
print(product)

conn.close()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# Обновление одной записи
laptop = session.query(Product).filter(Product.name == 'Laptop').first()
if laptop:
laptop.price = 899.99
session.commit()

# Обновление нескольких записей
accessories = session.query(Product).filter(Product.category == 'Accessories').all()
for product in accessories:
product.category = 'Computer Accessories'
session.commit()

# Проверка результатов
updated_products = session.query(Product).all()
for product in updated_products:
print(product)

import sqlite3

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

# Удаление одной записи
cursor.execute('DELETE FROM products WHERE name = ?', ('Mouse',))

# Удаление нескольких записей
cursor.execute('DELETE FROM products WHERE category = ?', ('Computer Accessories',))

conn.commit()

# Проверка результатов
cursor.execute('SELECT * FROM products')
remaining_products = cursor.fetchall()
for product in remaining_products:
print(product)

conn.close()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# Удаление одной записи
mouse = session.query(Product).filter(Product.name == 'Mouse').first()
if mouse:
session.delete(mouse)
session.commit()

# Удаление нескольких записей
accessories = session.query(Product).filter(
Product.category == 'Computer Accessories'
).all()
for product in accessories:
session.delete(product)
session.commit()

# Проверка результатов
remaining_products = session.query(Product).all()
for product in remaining_products:
print(product)

# Установка
# pip install alembic

# Инициализация
# alembic init migrations

# В alembic.ini настройте URL базы данных
# sqlalchemy.url = sqlite:///example.db

# Создание миграции
# alembic revision --autogenerate -m "Create users table"

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

# Пример файла миграции (versions/xxx_create_users_table.py)
from alembic import op
import sqlalchemy as sa

def upgrade():
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)

def downgrade():
op.drop_table('users')

from sqlalchemy import create_engine, func, text
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)
session = Session()

# Использование индексов (при создании таблицы)
# Index('idx_user_email', User.email)

# Выборочное получение столбцов
users_emails = session.query(User.id, User.email).all()

# Использование пагинации
page = 2
page_size = 10
users_page = session.query(User).limit(page_size).offset((page – 1) * page_size).all()

# Отложенная загрузка связанных объектов
from sqlalchemy.orm import joinedload
users_with_posts = session.query(User).options(joinedload(User.posts)).all()

# Агрегатные функции
post_count_by_user = session.query(
User.name, func.count(Post.id).label('post_count')
).join(Post).group_by(User.id).all()

# Сырой SQL для сложных запросов
complex_query = session.execute(text('''
SELECT u.name, COUNT(p.id) as post_count
FROM users u
LEFT JOIN posts p ON u.id = p.user_id
GROUP BY u.id
HAVING COUNT(p.id) > 5
ORDER BY post_count DESC
'''))
for row in complex_query:
print(row)

# Использование асинхронного драйвера
# pip install asyncpg

import asyncio
import asyncpg

async def get_users():
conn = await asyncpg.connect(
user='postgres',
password='postgres',
database='testdb',
host='localhost'
)

# Выполнение запроса
users = await conn.fetch('SELECT * FROM users')

# Выполнение транзакции
async with conn.transaction():
await conn.execute(
'INSERT INTO users(name, email) VALUES($1, $2)',
'John Doe', 'john@example.com'
)

await conn.close()
return users

# Запуск асинхронной функции
users = asyncio.run(get_users())
print(users)

import pandas as pd
from sqlalchemy import create_engine

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

# Чтение данных пакетами
chunksize = 10000
for chunk in pd.read_sql('SELECT * FROM large_table', engine, chunksize=chunksize):
# Обработка каждого пакета данных
processed_data = process_chunk(chunk)
# Сохранение результатов
processed_data.to_sql('results_table', engine, if_exists='append', index=False)

# Использование генераторов для экономии памяти
def stream_results(conn, query):
cursor = conn.cursor()
cursor.execute(query)
while True:
records = cursor.fetchmany(1000)
if not records:
break
for record in records:
yield record

for record in stream_results(engine.raw_connection(), 'SELECT * FROM large_table'):
process_record(record)

# Пример интеграции с Flask
# pip install flask flask-sqlalchemy

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)

with app.app_context():
db.create_all()

@app.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
result = [{'id': user.id, 'name': user.name, 'email': user.email} for user in users]
return jsonify(result)

@app.route('/users', methods=['POST'])
def create_user():
data = request.json
new_user = User(name=data['name'], email=data['email'])
db.session.add(new_user)
db.session.commit()
return jsonify({'id': new_user.id, 'name': new_user.name, 'email': new_user.email}), 201

if __name__ == '__main__':
app.run(debug=True)