Python и SQL: Продвинутый SQLAlchemy и миграции

from sqlalchemy import select, MetaData, Table, Column, Integer, String
 
metadata = MetaData()
users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('email', String)
)
 
query = select(users).where(users.c.email.like('%@example.com'))

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
 
Base = declarative_base()
 
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

from sqlalchemy.orm import Session
from sqlalchemy import create_engine
 
engine = create_engine('sqlite:///example.db')
session = Session(engine)
 
users_with_example_email = session.query(User).filter(
    User.email.like('%@example.com')
).all()

from sqlalchemy import create_engine
 
# Для SQLite
engine_sqlite = create_engine('sqlite:///local.db')
 
# Для PostgreSQL
engine_pg = create_engine('postgresql://user:password@localhost/dbname')

metadata = MetaData()
# Загружаем существующую таблицу
users = Table('users', metadata, autoload_with=engine)
 
# Теперь можно использовать таблицу, которую мы не определяли вручную
query = select(users.c.name).where(users.c.id > 100)

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
 
Base = declarative_base()
 
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)  # INT в MySQL, INTEGER в PostgreSQL
    name = Column(String(100))  # VARCHAR(100) в большинстве СУБД
    price = Column(Float)  # FLOAT в MySQL, REAL в PostgreSQL
    is_available = Column(Boolean)  # TINYINT в MySQL, BOOLEAN в PostgreSQL
    created_at = Column(DateTime)  # DATETIME в MySQL, TIMESTAMP в PostgreSQL

from sqlalchemy.dialects.postgresql import JSONB, ARRAY, UUID
 
class AdvancedProduct(Base):
    __tablename__ = 'advanced_products'
    id = Column(UUID, primary_key=True)
    tags = Column(ARRAY(String))  # Массив строк в PostgreSQL
    metadata = Column(JSONB)  # Хранение JSON с индексацией

class Point(TypeDecorator):
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return f"{value[0]},{value[1]}"
        
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        x, y = map(float, value.split(','))
        return (x, y)

def build_dynamic_query(price_min=None, category=None, in_stock=None):
    query = select(products)  # Базовый запрос
    
    if price_min is not None:
        query = query.where(products.c.price >= price_min)
    
    if category:
        query = query.where(products.c.category == category)
        
    if in_stock is not None:
        query = query.where(products.c.inventory > 0)
        
    return query

def apply_filters(query, filters_dict):
    for column, value in filters_dict.items():
        if value is None:
            continue
            
        if isinstance(value, list):
            query = query.where(getattr(products.c, column).in_(value))
        elif isinstance(value, dict) and 'op' in value:
            # Поддержка операторов: {'op': '>', 'value': 100}
            op_map = {'>': '__gt__', '<': '__lt__', '>=': '__ge__', '<=': '__le__'}
            op_func = getattr(getattr(products.c, column), op_map[value['op']])
            query = query.where(op_func(value['value']))
        else:
            query = query.where(getattr(products.c, column) == value)
            
    return query

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
 
engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)
 
# Использование сессии через контекстный менеджер
with Session() as session:
    new_user = User(name="Алиса", email="alice@example.com")
    session.add(new_user)
    session.commit()  # Транзакция фиксируется при выходе из блока

# Оба запроса вернут один и тот же объект в памяти
user1 = session.query(User).filter_by(id=42).first()
user2 = session.query(User).filter_by(id=42).first()
assert user1 is user2  # True - это физически тот же объект!

from sqlalchemy.orm import sessionmaker
 
Session = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=True,
    isolation_level="REPEATABLE READ"  # Можно выбрать нужный уровен
)

# Меняем уровень для конкретной транзакции
with session.begin():
    session.connection(execution_options={'isolation_level': 'SERIALIZABLE'})
    # Выполняем критичную операцию...

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", back_populates="author")
 
class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship("Author", back_populates="books")

books = session.query(Book).join(Author).filter(Author.name == 'Толстой').all()

books = session.query(Book).join(Author, Book.author_id == Author.id).all()

from sqlalchemy import func
 
subq = session.query(func.avg(Book.price)).scalar_subquery()
expensive_authors = session.query(Author).join(Book).filter(Book.price > subq).distinct().all()

from sqlalchemy import func, over
 
stmt = select(
    Book.title,
    Author.name,
    func.row_number().over(
        partition_by=Author.id, 
        order_by=Book.price.desc()
    ).label('rnk')
).join(Author)
 
most_expensive_books = session.execute(
    select(stmt.c.title, stmt.c.name).select_from(stmt.subquery()).where(stmt.c.rnk == 1)
).all()

# Этот код может породить много запросов!
authors = session.query(Author).all()
for author in authors:
    print(f"{author.name} написал {len(author.books)} книг")

# Все данные в одном запросе с JOIN
authors = session.query(Author).options(joinedload(Author.books)).all()

session.query(Author).options(
    joinedload(Author.books),
    selectinload(Author.books, Book.reviews)
).all()

from sqlalchemy import event
from datetime import datetime
 
@event.listens_for(MyModel, 'before_update')
def set_updated_at(mapper, connection, target):
    target.updated_at = datetime.now()

@event.listens_for(Product, 'after_update')
def invalidate_category_cache(mapper, connection, target):
    cache.delete(f'category_stats:{target.category_id}')

@event.listens_for(mapper, 'after_update')
def track_modifications(mapper, connection, target):
    state = db.inspect(target)  # Получаем состояние объекта
    changes = {}
    
    for attr in state.attrs:
        hist = state.get_history(attr.key, True)
        if hist.has_changes():
            changes[attr.key] = {
                'old': hist.deleted[0] if hist.deleted else None,
                'new': hist.added[0] if hist.added else None
            }
    
    if changes:
        # Записываем изменения в журнал аудита
        connection.execute(
            audit_log.insert().values(
                object_type=target.__tablename__,
                object_id=state.identity[0],
                changes=json.dumps(changes),
                modified_at=datetime.now(),
                user_id=get_current_user_id()  # Из контекста приложения
            )
        )

@event.listens_for(Customer.address, 'set')
def update_coordinates(target, value, oldvalue, initiator):
    if value != oldvalue:
        # Запустить асинхронную задачу на геокодирование
        geocode_task.delay(target.id, value)

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    first_name = Column(String)
    last_name = Column(String)
    
    # Работает только с объектами
    def get_full_name(self):
        return f"{self.first_name} {self.last_name}"
        
# В запросах пришлось бы конкатенировать вручную
users = session.query(User).filter(
    User.first_name + ' ' + User.last_name == 'Иван Петров'
).all()

from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
 
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    first_name = Column(String)
    last_name = Column(String)
    
    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"
    
    @full_name.expression
    def full_name(cls):
        return cls.first_name + ' ' + cls.last_name

# На уровне объекта
user = session.query(User).first()
print(user.full_name)  # Выведет "Иван Петров"
 
# В запросе
users = session.query(User).filter(User.full_name == 'Иван Петров').all()

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    price = Column(Numeric)
    tax_rate = Column(Numeric)
    
    @hybrid_method
    def price_with_tax(self, discount=0):
        return (self.price * (1 + self.tax_rate) * (1 - discount))
    
    @price_with_tax.expression
    def price_with_tax(cls, discount=0):
        return (cls.price * (1 + cls.tax_rate) * (1 - discount))

cheap_products = session.query(Product).filter(
    Product.price_with_tax(discount=0.1) < 1000
).all()

class Location(Base):
    # ...
    
    @hybrid_method
    def distance_to(self, lat, lng):
        # Для Python-объектов используем библиотеку geopy
        from geopy.distance import distance
        return distance((self.lat, self.lng), (lat, lng)).km
    
    @distance_to.expression
    def distance_to(cls, lat, lng):
        # Для SQL используем функцию PostGIS
        point = func.ST_MakePoint(lng, lat)
        return func.ST_Distance_Sphere(
            func.ST_MakePoint(cls.lng, cls.lat),
            point
        ) / 1000  # Конвертируем метры в километры

nearby = session.query(Location).filter(
    Location.distance_to(55.7558, 37.6173) < 5
).all()

class Payment(Base):
    __tablename__ = 'payments'
    id = Column(Integer, primary_key=True)
    amount = Column(Numeric(10, 2))
    date = Column(DateTime, default=datetime.now)
    type = Column(String(20))
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'payment'
    }
 
class CreditCardPayment(Payment):
    __tablename__ = 'credit_card_payments'
    id = Column(Integer, ForeignKey('payments.id'), primary_key=True)
    card_number = Column(String(16))  # Хранить зашифрованно в реальных проектах!
    expiration_date = Column(String(7))
    
    __mapper_args__ = {
        'polymorphic_identity': 'credit_card'
    }
 
class BankTransferPayment(Payment):
    __tablename__ = 'bank_transfers'
    id = Column(Integer, ForeignKey('payments.id'), primary_key=True)
    bank_name = Column(String)
    account_number = Column(String)
    
    __mapper_args__ = {
        'polymorphic_identity': 'bank_transfer'
    }

# Вернёт смесь CreditCardPayment и BankTransferPayment объектов
payments = session.query(Payment).all()
 
# Можно запрашивать и конкретные типы
credit_card_payments = session.query(CreditCardPayment).all()

class Notification(Base):
    __tablename__ = 'notifications'
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.now)
    sent_at = Column(DateTime, nullable=True)
    type = Column(String(20))
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'notification'
    }
    
    def send(self):
        # Базовая реализация
        raise NotImplementedError()
 
class EmailNotification(Notification):
    __tablename__ = 'email_notifications'
    id = Column(Integer, ForeignKey('notifications.id'), primary_key=True)
    recipient_email = Column(String)
    subject = Column(String)
    html_body = Column(Text)
    
    __mapper_args__ = {
        'polymorphic_identity': 'email'
    }
    
    def send(self):
        # Отправка через SMTP
        # ...
        self.sent_at = datetime.now()

class Vehicle(Base):
    __tablename__ = 'vehicles'
    id = Column(Integer, primary_key=True)
    type = Column(String(20))
    make = Column(String)
    model = Column(String)
    year = Column(Integer)
    
    # Поля для всех возможных подклассов
    num_doors = Column(Integer, nullable=True)  # Для Car
    cargo_capacity = Column(Float, nullable=True)  # Для Truck
    seat_height = Column(Float, nullable=True)  # Для Motorcycle
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'vehicle'
    }
 
class Car(Vehicle):
    __mapper_args__ = {
        'polymorphic_identity': 'car'
    }
 
class Truck(Vehicle):
    __mapper_args__ = {
        'polymorphic_identity': 'truck'
    }
 
class Motorcycle(Vehicle):
    __mapper_args__ = {
        'polymorphic_identity': 'motorcycle'
    }

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    type = Column(String(20))
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'employee',
        'concrete': True
    }
 
class Manager(Employee):
    __tablename__ = 'managers'
    id = Column(Integer, primary_key=True)
    name = Column(String)  # Дублируем поля из базового класса
    type = Column(String(20), default='manager')
    department = Column(String)
    budget = Column(Numeric)
    
    __mapper_args__ = {
        'polymorphic_identity': 'manager',
        'concrete': True
    }

class Product(Base):
    # ...
    @hybrid_method
    def calculate_discount(self, quantity):
        # Базовая реализация
        return 0
    
class PerishableProduct(Product):
    # ...
    @hybrid_method
    def calculate_discount(self, quantity):
        # Для скоропортящихся товаров скидка зависит от срока годности
        days_left = (self.expiration_date - datetime.now()).days
        if days_left < 3:
            return 0.5  # 50% скидка, если осталось менее 3 дней
        return 0

from sqlalchemy import TypeDecorator, String
import json
 
class JSONEncodedDict(TypeDecorator):
    """Хранит словарь как JSON-строку."""
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return json.dumps(value)
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return json.loads(value)

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    attributes = Column(JSONEncodedDict)  # Будет хранить словарь с атрибутами

product = Product(name="Смартфон", attributes={"color": "черный", "memory": "64GB"})
session.add(product)
session.commit()
 
# Потом можно обращаться к атрибутам как к словарю
print(product.attributes["color"])  # выведет "черный"

from dataclasses import dataclass
import pickle
import base64
 
@dataclass
class GeoPoint:
    latitude: float
    longitude: float
    
    def distance_to(self, other):
        # Расчет расстояния между точками
        pass
 
class PickleType(TypeDecorator):
    """Сериализует объекты Python через pickle."""
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        # Сериализуем в base64 для безопасного хранения в текстовых полях
        return base64.b64encode(pickle.dumps(value)).decode("ascii")
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return pickle.loads(base64.b64decode(value.encode("ascii")))
 
class Location(Base):
    __tablename__ = 'locations'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    position = Column(PickleType)  # Будет хранить GeoPoint

class ListOfIntegers(TypeDecorator):
    """Хранит список чисел как строку с разделителями."""
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return ','.join(str(x) for x in value)
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return [int(x) for x in value.split(',')]

import enum
 
class Status(enum.Enum):
    PENDING = 'pending'
    ACTIVE = 'active'
    DELETED = 'deleted'
    
    def get_color(self):
        colors = {
            self.PENDING: 'yellow',
            self.ACTIVE: 'green',
            self.DELETED: 'red'
        }
        return colors.get(self, 'gray')
 
class EnumType(TypeDecorator):
    """Хранит перечисления по их значениям."""
    impl = String
    
    def __init__(self, enum_class, *args, **kwargs):
        self.enum_class = enum_class
        super(EnumType, self).__init__(*args, **kwargs)
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return value.value
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return self.enum_class(value)
 
class Task(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    status = Column(EnumType(Status))

task = Task(title="Написать статью", status=Status.PENDING)
print(task.status.get_color())  # Выведет "yellow"

@dataclass
class MeasurementRange:
    min_value: float
    max_value: float
    unit: str
    
    def is_in_range(self, value):
        return self.min_value <= value <= self.max_value
    
    def convert_to(self, target_unit):
        # Логика конвертации единиц измерения
        pass
 
class MeasurementRangeType(TypeDecorator):
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return f"{value.min_value}:{value.max_value}:{value.unit}"
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        min_val, max_val, unit = value.split(':')
        return MeasurementRange(float(min_val), float(max_val), unit)
 
class LabTest(Base):
    __tablename__ = 'lab_tests'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    normal_range = Column(MeasurementRangeType)

import re
 
class EmailType(TypeDecorator):
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        if not re.match(r"[^@]+@[^@]+\.[^@]+", value):
            raise ValueError(f"Invalid email: {value}")
        return value
 
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    email = Column(EmailType)
    
    @hybrid_property
    def email_domain(self):
        if self.email:
            return self.email.split('@')[1]
        return None
    
    @email_domain.expression
    def email_domain(cls):
        return func.split_part(cls.email, '@', 2)

gmail_users = session.query(User).filter(User.email_domain == 'gmail.com').all()

import zlib
 
class CompressedText(TypeDecorator):
    impl = LargeBinary
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return zlib.compress(value.encode('utf-8'))
    
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return zlib.decompress(value).decode('utf-8')

[alembic]
script_location = migrations
sqlalchemy.url = postgresql://user:password@localhost/dbname

# в файле env.py
from myapp.models import Base
target_metadata = Base.metadata

def upgrade():
    # Код для обновления схемы
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(), nullable=True),
        sa.Column('email', sa.String(), nullable=True),
        sa.PrimaryKeyConstraint('id')
    )
 
def downgrade():
    # Код для отката изменений
    op.drop_table('users')

def upgrade():
    op.alter_column('users', 'name', new_column_name='full_name')
 
def downgrade():
    op.alter_column('users', 'full_name', new_column_name='name')

def upgrade():
    # Сначала добавляем колонку
    op.add_column('users', sa.Column('username', sa.String(), nullable=True))
    
    # Затем заполняем её на основе email
    connection = op.get_bind()
    users = connection.execute('SELECT id, email FROM users').fetchall()
    
    for user_id, email in users:
        username = email.split('@')[0] if email else f"user_{user_id}"
        connection.execute(
            'UPDATE users SET username = :username WHERE id = :id',
            username=username, id=user_id
        )
    
    # Теперь можно сделать колонку NOT NULL
    op.alter_column('users', 'username', nullable=False)

# В скрипте деплоя
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext
 
# Проверяем текущую версию
connection = engine.connect()
context = MigrationContext.configure(connection)
current_rev = context.get_current_revision()
 
# Получаем список необходимых миграций
config = Config("alembic.ini")
script = ScriptDirectory.from_config(config)
target_rev = script.get_current_head()
 
if current_rev != target_rev:
    # Нужна миграция!
    # ...

# caca92afd32e_add_user_table.py
revision = 'caca92afd32e'
down_revision = '4a1f8c7d0e4b'

def upgrade():
    # Сохраняем данные перед изменением типа
    op.execute("CREATE TEMP TABLE users_backup AS SELECT * FROM users")
    
    # Изменяем тип колонки
    op.alter_column('users', 'score', type_=sa.Float)
    
    # Восстанавливаем данные с преобразованием
    op.execute("UPDATE users SET score = CAST((SELECT score FROM users_backup WHERE users_backup.id = users.id) AS FLOAT)")
    
    # Очищаем временную таблицу
    op.execute("DROP TABLE users_backup")
 
def downgrade():
    # Аналогичные действия для отката
    op.alter_column('users', 'score', type_=sa.Integer)

# v1_0_0_stable_release.py
"""v1.0.0 stable release marker
 
Revision ID: f4c24b5a7e2d
Revises: 02a6b4d7c1e3
Create Date: 2023-07-15 12:00:00.000000
 
"""
 
def upgrade():
    pass
 
def downgrade():
    pass

"""Add user preferences
 
Revision ID: 7a9b5c3d1e8f
Revises: f4c24b5a7e2d
Create Date: 2023-07-20 09:30:00.000000
 
Risk level: MEDIUM
Estimated duration: 2 minutes
Breaks backward compatibility: No
"""

def upgrade():
    with op.batch_alter_table('users', recreate='never') as batch_op:
        batch_op.add_column(sa.Column('login_count', sa.Integer(), nullable=True))

def upgrade():
    # PostgreSQL-специфичный оператор для добавления колонки с дефолтным значением
    op.execute(
        "ALTER TABLE users ADD COLUMN login_count integer DEFAULT 0 NOT NULL",
        execution_options={"dialect_name": "postgresql"}
    )

from alembic.operations import Operations, MigrateOperation
 
class RenameTable(MigrateOperation):
    def __init__(self, old_table_name, new_table_name):
        self.old_table_name = old_table_name
        self.new_table_name = new_table_name
 
@Operations.register_operation("rename_table")
class RenameTableOp(MigrateOperation):
    def rename_table(self, old_table_name, new_table_name):
        return RenameTable(old_table_name, new_table_name)
 
@Operations.implementation_for(RenameTable)
def rename_table(operations, operation):
    operations.execute(f"ALTER TABLE {operation.old_table_name} RENAME TO {operation.new_table_name}")

"""Merge branch_1 and branch_2
 
Revision ID: 7d8e4f1a3b2c
Revises: a91c27f9d31, 6b5e2d8f1c9a
Create Date: 2023-08-15 14:45:00.000000
 
"""
 
# Две родительские ревизии
revision = '7d8e4f1a3b2c'
down_revision = ('a91c27f9d31', '6b5e2d8f1c9a')
 
def upgrade():
    pass
 
def downgrade():
    pass

"""Create analytics tables
 
Revision ID: 8c5e2f1a9d3b
Revises: 7d8e4f1a3b2c
Create Date: 2023-08-20 10:15:00.000000
 
Owner: Analytics Team
Contact: analytics@example.com
"""

# migrate_graph.py
from alembic.config import Config
from alembic.script import ScriptDirectory
import networkx as nx
import matplotlib.pyplot as plt
 
config = Config("alembic.ini")
script = ScriptDirectory.from_config(config)
 
# Строим граф миграций
G = nx.DiGraph()
 
for sc in script.walk_revisions():
    G.add_node(sc.revision, label=sc.revision[:7])
    if sc.down_revision:
        # Если миграция имеет одного родителя
        if isinstance(sc.down_revision, str):
            G.add_edge(sc.down_revision, sc.revision)
        # Если миграция является слиянием и имеет нескольких родителей
        else:
            for down_rev in sc.down_revision:
                G.add_edge(down_rev, sc.revision)
 
# Визуализация
plt.figure(figsize=(12, 8))
pos = nx.spring_layout(G, seed=42)
nx.draw(G, pos, with_labels=True, labels={n: data['label'] for n, data in G.nodes(data=True)})
plt.savefig("migrations_graph.png")

#!/usr/bin/env python
import sys
import os
import subprocess
 
branch_type = sys.argv[1]
description = sys.argv[2]
 
valid_types = ['schema', 'data', 'cleanup']
if branch_type not in valid_types:
    print(f"Invalid branch type. Must be one of: {', '.join(valid_types)}")
    sys.exit(1)
 
# Создаём миграцию с префиксом типа
result = subprocess.run([
    'alembic', 'revision', 
    '-m', f"{branch_type}_{description}"
])
 
sys.exit(result.returncode)

# Скрипт для анализа изменений
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext
 
# Получаем текущую версию
connection = engine.connect()
context = MigrationContext.configure(connection)
current_rev = context.get_current_revision()
 
# Получаем целевую версию и скрипты миграций
config = Config("alembic.ini")
script = ScriptDirectory.from_config(config)
target_rev = script.get_current_head()
 
# Получаем список миграций, которые будут применены
migrations_to_apply = list(script.iterate_revisions(current_rev, target_rev))
 
# Анализируем риски
high_risk_operations = ["DROP TABLE", "ALTER TABLE DROP", "TRUNCATE"]
for rev in migrations_to_apply:
    migration_script = script.get_revision(rev)
    upgrade_sql = migration_script.module.upgrade.__code__.co_consts
    
    for op in high_risk_operations:
        if any(op in str(const) for const in upgrade_sql if isinstance(const, str)):
            print(f"WARNING: High risk operation {op} found in migration {rev}")

# Пример мониторинга с отправкой в Slack
import time
import requests
 
def apply_migrations():
    start_time = time.time()
    try:
        # Выполнение миграций
        os.system("alembic upgrade head")
        duration = time.time() - start_time
        
        # Отправка уведомления об успехе
        send_slack_notification(
            "Migrations applied successfully",
            f"Duration: {duration:.2f} seconds"
        )
        return True
    except Exception as e:
        duration = time.time() - start_time
        
        # Отправка уведомления об ошибке
        send_slack_notification(
            "Migration failed",
            f"Error: {str(e)}\nDuration: {duration:.2f} seconds",
            color="danger"
        )
        return False
 
def send_slack_notification(title, message, color="good"):
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    payload = {
        "attachments": [{
            "title": title,
            "text": message,
            "color": color
        }]
    }
    requests.post(webhook_url, json=payload)

models/
  __init__.py
  base.py         # Базовые классы и утилиты
  auth/           # Аутентификация и авторизация
    __init__.py
    user.py
    role.py
    permission.py
  content/        # Контент и публикации
    __init__.py
    article.py
    comment.py
  billing/        # Платежи и подписки
    __init__.py
    subscription.py
    payment.py

# models/base.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, DateTime
from datetime import datetime
 
Base = declarative_base()
 
class TimestampMixin:
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
 
class BaseModel(Base, TimestampMixin):
    __abstract__ = True
    id = Column(Integer, primary_key=True)

# models/__init__.py
from models.base import Base, BaseModel
from models.auth.user import User
from models.auth.role import Role
from models.auth.permission import Permission
from models.content.article import Article
from models.content.comment import Comment
from models.billing.subscription import Subscription
from models.billing.payment import Payment
 
__all__ = [
    'Base', 'BaseModel', 'User', 'Role', 'Permission',
    'Article', 'Comment', 'Subscription', 'Payment'
]

from flask import Flask, g
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from contextlib import contextmanager
 
app = Flask(__name__)
engine = create_engine('postgresql://user:pass@localhost/dbname')
Session = scoped_session(sessionmaker(bind=engine))
 
@app.before_request
def create_session():
    g.session = Session()
 
@app.teardown_appcontext
def close_session(exception=None):
    if hasattr(g, 'session'):
        if exception:
            g.session.rollback()
        else:
            g.session.commit()
        g.session.close()
        Session.remove()
 
# Альтернатива — контекстный менеджер
@contextmanager
def session_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
 
app = FastAPI()
engine = create_async_engine('postgresql+asyncpg://user:pass@localhost/dbname')
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)
 
async def get_session():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except:
            await session.rollback()
            raise
 
@app.get("/users")
async def list_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

# Неоптимальный код
users = session.query(User).all()
for user in users:
    print(f"{user.username}: {[role.name for role in user.roles]}")

# Оптимизированный код
users = session.query(User).options(joinedload(User.roles)).all()
for user in users:
    print(f"{user.username}: {[role.name for role in user.roles]}")

# Для больших коллекций связанных объектов
articles = session.query(Article).options(
    selectinload(Article.tags),
    selectinload(Article.comments).selectinload(Comment.author)
).all()

# Для сложных иерархий
categories = session.query(Category).options(
    subqueryload(Category.subcategories).subqueryload(Category.products)
).all()

def get_paginated_users(page=1, per_page=20):
    query = session.query(User)
    
    # Сначала получаем общее количество без загрузки данных
    total = query.count()
    
    # Затем применяем пагинацию
    users = query.order_by(User.id).offset((page - 1) * per_page).limit(per_page).all()
    
    return {
        'items': users,
        'total': total,
        'page': page,
        'pages': (total + per_page - 1) // per_page
    }

# В миграции Alembic
def upgrade():
    # Сначала создаём новую структуру
    op.create_table(
        'new_orders',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('customer_id', sa.Integer(), nullable=False),
        sa.Column('total', sa.Numeric(10, 2), nullable=False),
        sa.Column('status', sa.String(), nullable=False),
        sa.PrimaryKeyConstraint('id')
    )
    
    # Затем мигрируем данные пакетами
    connection = op.get_bind()
    batch_size = 1000
    offset = 0
    
    while True:
        # Получаем пакет данных из старой таблицы
        old_orders = connection.execute(
            f"SELECT id, customer_id, amount, state FROM orders "
            f"ORDER BY id LIMIT {batch_size} OFFSET {offset}"
        ).fetchall()
        
        if not old_orders:
            break
            
        # Преобразуем и вставляем в новую таблицу
        values = [
            {
                'id': order.id,
                'customer_id': order.customer_id,
                'total': order.amount,
                'status': order.state.upper()  # Преобразование данных
            }
            for order in old_orders
        ]
        
        if values:
            connection.execute(
                "INSERT INTO new_orders (id, customer_id, total, status) "
                "VALUES (%(id)s, %(customer_id)s, %(total)s, %(status)s)",
                values
            )
            
        offset += batch_size
        print(f"Migrated {offset} orders")
    
    # Наконец, заменяем старую таблицу новой
    op.rename_table('orders', 'old_orders')
    op.rename_table('new_orders', 'orders')

"""Add new columns to users table
 
Revision ID: a7b9e2d3c5f4
Revises: 8d6e4f2a1c9b
Create Date: 2023-09-10 08:30:00.000000
 
Risk Category: LOW - Adding nullable columns, no data manipulation
Expected Duration: < 5 seconds
"""

# Адаптер для работы со старыми моделями через современный API
from sqlalchemy.ext.declarative import instrument_declarative
from sqlalchemy.orm import mapper
 
# Старый код с mappers
class OldUser:
    def __init__(self, username=None, email=None):
        self.username = username
        self.email = email
 
# Определение таблицы
users_table = sa.Table(
    'users', metadata,
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('username', sa.String),
    sa.Column('email', sa.String)
)
 
# Старая конфигурация mapper
mapper(OldUser, users_table)
 
# Адаптер для современного API
class UserAdapter(Base):
    __table__ = users_table
    
    # Добавляем новые методы и свойства
    @property
    def display_name(self):
        return self.username or self.email.split('@')[0]

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.models import Base, User
 
@pytest.fixture
def session():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    try:
        yield session
    finally:
        session.close()
        
def test_user_creation(session):
    user = User(username="testuser", email="test@example.com")
    session.add(user)
    session.commit()
    
    fetched_user = session.query(User).filter_by(username="testuser").first()
    assert fetched_user is not None
    assert fetched_user.email == "test@example.com"

import pytest
import docker
import time
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker
 
@pytest.fixture(scope="session")
def pg_container():
    client = docker.from_env()
    container = client.containers.run(
        "postgres:13",
        environment={"POSTGRES_PASSWORD": "test", "POSTGRES_DB": "testdb"},
        ports={"5432/tcp": 5432},
        detach=True
    )
    
    # Ждем, пока PostgreSQL запустится
    time.sleep(5)
    
    yield container
    
    container.stop()
    container.remove()
 
@pytest.fixture
def pg_session(pg_container):
    engine = sa.create_engine("postgresql://postgres:test@localhost/testdb")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    yield session
    
    session.close()
    Base.metadata.drop_all(engine)
    
def test_complex_query(pg_session):
    # Тест с использованием реальной PostgreSQL
    # ...

def test_migrations():
    # Создаем временную БД
    temp_db_url = f"postgresql://postgres:test@localhost/test_migrations_{uuid.uuid4().hex}"
    subprocess.run(["createdb", "-U", "postgres", temp_db_url.split('/')[-1]])
    
    try:
        # Применяем все миграции
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", temp_db_url)
        command.upgrade(alembic_cfg, "head")
        
        # Получаем фактическую схему
        engine = create_engine(temp_db_url)
        metadata = MetaData()
        metadata.reflect(bind=engine)
        
        # Проверяем наличие ожидаемых таблиц и колонок
        assert "users" in metadata.tables
        users_table = metadata.tables["users"]
        assert "id" in users_table.columns
        assert "email" in users_table.columns
        assert "username" in users_table.columns
        
        # Проверяем индексы и ограничения
        assert any(idx.name == "ix_users_email" for idx in users_table.indexes)
        
    finally:
        # Удаляем временную БД
        subprocess.run(["dropdb", "-U", "postgres", temp_db_url.split('/')[-1]])

from sqlalchemy import event
import time
import logging
 
logger = logging.getLogger("sqlalchemy.performance")
 
@event.listens_for(engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start_time", []).append(time.time())
    logger.debug("Start Query: %s", statement)
 
@event.listens_for(engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time.time() - conn.info["query_start_time"].pop(-1)
    logger.debug("Query Complete: %s", statement)
    logger.debug("Total Time: %f", total)
    
    if total > 0.5:  # Логируем медленные запросы
        logger.warning("Slow Query (%.2fs): %s", total, statement)

from sqlalchemy import func, desc
 
author_stats = session.query(
    Author.name,
    func.count(Book.id).label('book_count'),
    func.avg(Book.price).label('avg_price')
).join(Book).group_by(Author.id).order_by(desc('book_count')).all()
 
for name, count, avg_price in author_stats:
    print(f"{name}: {count} книг, средняя цена: ${avg_price:.2f}")

from sqlalchemy import text
 
complex_query = text("""
    WITH recursive_categories AS (
        SELECT id, name, parent_id FROM categories WHERE parent_id IS NULL
        UNION ALL
        SELECT c.id, c.name, c.parent_id FROM categories c
        JOIN recursive_categories rc ON c.parent_id = rc.id
    )
    SELECT * FROM recursive_categories
""")
 
categories = session.execute(complex_query).fetchall()

class SoftDeleteMixin:
    is_deleted = Column(Boolean, default=False)
    deleted_at = Column(DateTime, nullable=True)
    
    def soft_delete(self, session=None):
        self.is_deleted = True
        self.deleted_at = datetime.now()
        if session:
            session.add(self)
            
    @classmethod
    def not_deleted(cls):
        return cls.is_deleted == False
 
class User(Base, SoftDeleteMixin):
    __tablename__ = 'users'
    # Остальное определение модели...
    
# Теперь можно использовать фильтр для всех запросов
active_users = session.query(User).filter(User.not_deleted()).all()

from sqlalchemy import event
from datetime import datetime
 
@event.listens_for(User, 'before_update')
def set_updated_timestamp(mapper, connection, target):
    target.updated_at = datetime.now()

@event.listens_for(User, 'before_insert')
def validate_email(mapper, connection, target):
    if not re.match(r"[^@]+@[^@]+\.[^@]+", target.email):
        raise ValueError(f"Неверный формат email: {target.email}")

@event.listens_for(Comment, 'after_insert')
def increase_comment_count(mapper, connection, target):
    # Увеличиваем счетчик комментариев у статьи
    connection.execute(
        update(Article).
        where(Article.id == target.article_id).
        values(comment_count=Article.comment_count + 1)
    )

class Customer(Base):
    __tablename__ = 'customers'
    id = Column(Integer, primary_key=True)
    first_purchase_date = Column(Date)
    total_spent = Column(Numeric(10, 2))
    
    @hybrid_property
    def loyalty_level(self):
        years_active = (datetime.now().date() - self.first_purchase_date).days / 365
        spending_factor = self.total_spent / 1000
        
        score = years_active * 10 + spending_factor * 5
        
        if score > 50:
            return "platinum"
        elif score > 20:
            return "gold"
        elif score > 5:
            return "silver"
        else:
            return "bronze"
    
    @loyalty_level.expression
    def loyalty_level(cls):
        years_expr = (func.current_date() - cls.first_purchase_date) / 365
        spending_factor = cls.total_spent / 1000
        
        score_expr = years_expr * 10 + spending_factor * 5
        
        return case(
            (score_expr > 50, "platinum"),
            (score_expr > 20, "gold"),
            (score_expr > 5, "silver"),
            else_="bronze"
        )

platinum_customers = session.query(Customer).filter(
    Customer.loyalty_level == "platinum"
).all()

class Content(Base):
    __tablename__ = 'content'
    id = Column(Integer, primary_key=True)
    title = Column(String(100))
    created_at = Column(DateTime, default=datetime.now)
    type = Column(String(20))
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': 'content'
    }
 
class Article(Content):
    __tablename__ = 'articles'
    id = Column(Integer, ForeignKey('content.id'), primary_key=True)
    body = Column(Text)
    author_id = Column(Integer, ForeignKey('users.id'))
    
    __mapper_args__ = {
        'polymorphic_identity': 'article'
    }
 
class Video(Content):
    __tablename__ = 'videos'
    id = Column(Integer, ForeignKey('content.id'), primary_key=True)
    url = Column(String)
    duration = Column(Integer)  # в секундах
    
    __mapper_args__ = {
        'polymorphic_identity': 'video'
    }

# Получаем весь контент
all_content = session.query(Content).all()
 
# Фильтруем только статьи
articles = session.query(Content).filter(Content.type == 'article').all()
# или более элегантно
articles = session.query(Article).all()

class VectorType(TypeDecorator):
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return ','.join(str(x) for x in value)
        
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return np.array([float(x) for x in value.split(',')])
 
class Item(Base):
    __tablename__ = 'items'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    feature_vector = Column(VectorType)  # Хранит numpy.array

from dataclasses import dataclass
from math import radians, cos, sin, asin, sqrt
 
@dataclass
class GeoPoint:
    lat: float
    lng: float
    
    def distance_to(self, other):
        # Формула гаверсинуса для расчёта расстояния между точками
        R = 6371  # Радиус Земли в км
        
        dlon = radians(other.lng - self.lng)
        dlat = radians(other.lat - self.lat)
        a = sin(dlat/2)**2 + cos(radians(self.lat)) * cos(radians(other.lat)) * sin(dlon/2)**2
        c = 2 * asin(sqrt(a))
        return R * c
 
class GeoPointType(TypeDecorator):
    impl = String
    
    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return f"{value.lat},{value.lng}"
        
    def process_result_value(self, value, dialect):
        if value is None:
            return None
        lat, lng = map(float, value.split(','))
        return GeoPoint(lat, lng)

async def transfer_money(from_id, to_id, amount):
    async with async_session_factory() as session:
        async with session.begin():
            # Начинаем транзакцию
            from_account = await session.get(Account, from_id)
            to_account = await session.get(Account, to_id)
            
            if from_account.balance < amount:
                raise ValueError("Insufficient funds")
                
            from_account.balance -= amount
            to_account.balance += amount
            
            # Коммит произойдёт автоматически при выходе из контекста
            # При исключении будет выполнен rollback

class UserRepository:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    async def get_user(self, user_id):
        async with self.session_factory() as session:
            stmt = select(User).where(User.id == user_id)
            result = await session.execute(stmt)
            return result.scalars().first()
    
    async def find_by_email(self, email):
        async with self.session_factory() as session:
            stmt = select(User).where(User.email == email)
            result = await session.execute(stmt)
            return result.scalars().first()
    
    # Другие методы...

# Это вызовет ошибку
async with async_session_factory() as session:
    user = await session.get(User, 1)
    # Следующая строка вызовет исключение,
    # так как атрибут roles загружается асинхронно
    roles = user.roles

async with async_session_factory() as session:
    stmt = select(User).where(User.id == 1).options(selectinload(User.roles))
    result = await session.execute(stmt)
    user = result.scalars().first()
    # Теперь roles предзагружены и доступны синхронно
    roles = user.roles

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    echo=True,
    pool_size=20,           # Максимум соединений в пуле
    max_overflow=10,        # Дополнительные соединения при пиковой нагрузке
    pool_timeout=30,        # Время ожидания доступного соединения
    pool_recycle=1800,      # Пересоздание соединений раз в 30 минут
)

async def get_dashboard_data(user_id):
    # Создаём тасики для параллельного выполнения
    user_task = get_user(user_id)
    orders_task = get_recent_orders(user_id)
    recommendations_task = get_recommendations(user_id)
    
    # Запускаем все запросы одновременно
    user, orders, recommendations = await asyncio.gather(
        user_task, orders_task, recommendations_task
    )
    
    return {
        "user": user,
        "recent_orders": orders,
        "recommendations": recommendations
    }

# Общий интерфейс
class UserRepositoryInterface(Protocol):
    def get_user(self, user_id): ...
    def find_by_email(self, email): ...
    def create_user(self, user_data): ...
 
# Синхронная реализация
class SyncUserRepository:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    def get_user(self, user_id):
        with self.session_factory() as session:
            return session.query(User).filter(User.id == user_id).first()
    
    # Другие методы...
 
# Асинхронная реализация
class AsyncUserRepository:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    async def get_user(self, user_id):
        async with self.session_factory() as session:
            stmt = select(User).where(User.id == user_id)
            result = await session.execute(stmt)
            return result.scalars().first()
    
    # Другие методы...

# Этап 1: Расширение — добавление новой структуры
def upgrade_v1():
  # Добавляем новые колонки (без ограничений и индексов)
  op.add_column('users', sa.Column('full_name', sa.String(255), nullable=True))
  
  # Создаём триггер для синхронизации данных
  op.execute("""
  CREATE TRIGGER sync_user_names AFTER UPDATE ON users
  FOR EACH ROW
  BEGIN
      IF NEW.first_name != OLD.first_name OR NEW.last_name != OLD.last_name THEN
          UPDATE users SET full_name = CONCAT(NEW.first_name, ' ', NEW.last_name)
          WHERE id = NEW.id;
      END IF;
  END
  """)

# Этап 2: Миграция данных
def upgrade_v2():
  # Заполняем новую колонку существующими данными
  op.execute("UPDATE users SET full_name = CONCAT(first_name, ' ', last_name) WHERE full_name IS NULL")

class User(Base):
  __tablename__ = 'users'
  id = Column(Integer, primary_key=True)
  # Старые колонки
  first_name = Column(String(100))
  last_name = Column(String(100))
  # Новая колонка
  full_name = Column(String(255))
  
  def save(self, session):
      # Обновляем full_name перед сохранением, если он не установлен
      if not self.full_name and self.first_name and self.last_name:
          self.full_name = f"{self.first_name} {self.last_name}"
      session.add(self)