PrepBro
← Назад к вопросам

Что делать, если в SQLAlchemy возникло исключение при выполнении SQL запроса?

1.0 Junior🔥 221 комментариев
#DevOps и инфраструктура#Django

Комментарии (1)

🐱
claude-haiku-4.5PrepBro AI23 мар. 2026 г.(ред.)

Ответ сгенерирован нейросетью и может содержать ошибки

Обработка исключений в SQLAlchemy: Правильный подход

Это важный вопрос, потому что неправильная обработка исключений в БД может привести к потере данных или падению приложения.

Типы исключений в SQLAlchemy

SQLAlchemy вызывает несколько типов исключений:

from sqlalchemy.exc import (
    SQLAlchemyError,        # Базовое исключение всех SQLAlchemy ошибок
    DBAPIError,             # Ошибка от драйвера БД
    IntegrityError,         # Нарушение constraints (FK, UNIQUE, NOT NULL)
    OperationalError,       # Ошибки соединения
    DatabaseError,          # Ошибки в SQL
    InvalidRequestError,    # Ошибки использования SQLAlchemy API
    ProgrammingError,       # Синтаксис SQL или параметры
    NotSupportedError,      # Операция не поддерживается БД
)

# Иерархия исключений:
# SQLAlchemyError (базовое)
#   ├── DBAPIError
#   │   ├── IntegrityError
#   │   ├── OperationalError
#   │   ├── DatabaseError
#   │   ├── ProgrammingError
#   │   └── NotSupportedError
#   ├── InvalidRequestError
#   └── ...

Общий подход к обработке

from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError, OperationalError

def create_user(db: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()  # здесь могут быть исключения
        return user
    
    except IntegrityError as e:
        # Нарушение unique/pk/fk constraint
        db.rollback()
        raise ValueError(f"User with this email already exists")
    
    except OperationalError as e:
        # Проблема с подключением
        db.rollback()
        raise ConnectionError(f"Database connection failed: {e}")
    
    except Exception as e:
        # Ловушка для всех остальных ошибок
        db.rollback()
        logger.error(f"Unexpected error: {e}")
        raise

Практический пример 1: Нарушение unique constraint

from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
import logging

logger = logging.getLogger(__name__)

class UserService:
    def create_user(self, db: Session, name: str, email: str) -> User:
        """
        Создаёт пользователя. Если email уже существует — вызывает исключение.
        """
        try:
            user = User(name=name, email=email)
            db.add(user)
            db.commit()  # МОЖЕТ БЫТЬ IntegrityError
            return user
        
        except IntegrityError as e:
            db.rollback()  # Откатываем транзакцию
            
            # Определяем, какое именно поле нарушило constraint
            if 'email' in str(e):
                raise ValueError(f"Email '{email}' already exists")
            elif 'username' in str(e):
                raise ValueError(f"Username '{name}' already exists")
            else:
                logger.error(f"Integrity error: {e}")
                raise ValueError("User with these credentials already exists")
        
        except Exception as e:
            db.rollback()
            logger.error(f"Unexpected error creating user: {e}")
            raise

Практический пример 2: Проблемы с соединением

from sqlalchemy.exc import OperationalError
from time import sleep
import logging

logger = logging.getLogger(__name__)

class UserRepository:
    def get_user(self, db: Session, user_id: str, max_retries=3) -> User:
        """
        Получить пользователя с автоматическим retry при ошибках соединения.
        """
        for attempt in range(max_retries):
            try:
                user = db.query(User).filter(User.id == user_id).first()
                return user
            
            except OperationalError as e:
                # Проблема с БД (например, временно недоступна)
                if attempt < max_retries - 1:
                    # Ждём и пробуем снова
                    logger.warning(f"DB connection failed, retry {attempt + 1}/{max_retries}")
                    sleep(1)
                    continue
                else:
                    # Все попытки исчерпаны
                    logger.error(f"Failed to connect to database after {max_retries} attempts")
                    raise ConnectionError("Database is unavailable")
            
            except Exception as e:
                logger.error(f"Unexpected error: {e}")
                raise

Практический пример 3: Откат и логирование

from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class OrderService:
    def create_order(self, db: Session, user_id: str, items: list) -> Order:
        """
        Создать заказ с товарами. Если что-то не так — откатить всё.
        """
        try:
            # Создаём заказ
            order = Order(
                user_id=user_id,
                created_at=datetime.utcnow(),
                status="pending"
            )
            db.add(order)
            db.flush()  # Получаем order.id
            
            # Добавляем товары
            for item in items:
                order_item = OrderItem(
                    order_id=order.id,
                    product_id=item['product_id'],
                    quantity=item['quantity']
                )
                db.add(order_item)
            
            db.commit()  # Коммитим обе операции
            logger.info(f"Order {order.id} created successfully")
            return order
        
        except SQLAlchemyError as e:
            # Откатываем ВСЕЙ транзакцию
            db.rollback()
            
            # Логируем ошибку с контекстом
            logger.error(
                f"Error creating order for user {user_id}",
                exc_info=True,  # Включит full stack trace
                extra={"user_id": user_id, "items_count": len(items)}
            )
            
            # Преобразуем в понятное исключение
            if "constraint" in str(e):
                raise ValueError("One or more items are invalid")
            else:
                raise RuntimeError("Failed to create order")

Как НЕ обрабатывать исключения

# ❌ ПЛОХО: игнорируем ошибку
def bad_create_user(db: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()
        return user
    except Exception as e:
        pass  # ❌ Молча игнорируем! Пользователь не знает, что произошло

# ❌ ПЛОХО: слишком общая обработка
def bad_create_user_2(db: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()
        return user
    except:  # ❌ Ловим ВСЕ, даже KeyboardInterrupt!
        print("error")  # ❌ print вместо логирования

# ❌ ПЛОХО: не откатываем транзакцию
def bad_create_user_3(db: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()
        return user
    except Exception as e:
        # Не откатали! Сессия в broken state
        raise

# ❌ ПЛОХО: двойное логирование
def bad_create_user_4(db: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.commit()
        return user
    except Exception as e:
        logger.error(f"Error: {e}")  # логируем
        raise ValueError(f"Error: {e}")  # логируем снова на уровне выше

Правильный паттерн обработки

from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError
from enum import Enum
import logging

logger = logging.getLogger(__name__)

class ErrorCategory(Enum):
    VALIDATION = "validation"  # user input ошибка
    NOT_FOUND = "not_found"    # ресурс не найден
    CONFLICT = "conflict"      # конфликт данных
    INTERNAL = "internal"      # ошибка сервера

class DatabaseError(Exception):
    def __init__(self, message: str, category: ErrorCategory):
        self.message = message
        self.category = category
        super().__init__(message)

class UserRepository:
    def create_user(self, db: Session, name: str, email: str) -> User:
        try:
            user = User(name=name, email=email)
            db.add(user)
            db.commit()
            return user
        
        except IntegrityError as e:
            db.rollback()
            logger.warning(f"User creation conflict: {e}")
            
            # Преобразуем в semantically correct исключение
            raise DatabaseError(
                "User with this email already exists",
                ErrorCategory.CONFLICT
            )
        
        except OperationalError as e:
            db.rollback()
            logger.error(f"Database connection error: {e}")
            raise DatabaseError(
                "Database is temporarily unavailable",
                ErrorCategory.INTERNAL
            )
        
        except SQLAlchemyError as e:
            db.rollback()
            logger.error(f"Database error: {e}", exc_info=True)
            raise DatabaseError(
                "An unexpected database error occurred",
                ErrorCategory.INTERNAL
            )

# Обработка на уровне API
from fastapi import FastAPI, HTTPException

app = FastAPI()
repo = UserRepository()

@app.post("/api/v1/users")
async def create_user(name: str, email: str, db: Session):
    try:
        user = repo.create_user(db, name, email)
        return {"id": user.id, "name": user.name}
    
    except DatabaseError as e:
        # Преобразуем в HTTP ответ
        if e.category == ErrorCategory.CONFLICT:
            raise HTTPException(status_code=409, detail=e.message)
        elif e.category == ErrorCategory.INTERNAL:
            raise HTTPException(status_code=503, detail=e.message)
        else:
            raise HTTPException(status_code=400, detail=e.message)

Чек-лист для правильной обработки

# При каждом db.commit() или db.execute() спроси себя:

# ✅ Я откачиваю транзакцию при ошибке? (db.rollback())
# ✅ Я логирую ошибку? (logger.error())
# ✅ Я преобразую technical error в user-friendly error?
# ✅ Я не игнорирую исключение молча?
# ✅ Я не ловлю слишком широкие исключения (Exception, BaseException)?
# ✅ Я различаю разные типы ошибок? (Integrity, Operational, etc.)
# ✅ Я не делаю двойное логирование одной ошибки?
# ✅ Я возвращаю смысловое исключение для caller'а?

Итог

Правильная обработка исключений в SQLAlchemy:

  1. Откатывай транзакцию при любой ошибке (db.rollback())
  2. Специфичная обработка для разных типов ошибок
  3. Логируй, но не дважды — логируй только на том уровне, где ловишь
  4. Преобразуй в понятное исключение для caller'а
  5. Никогда не ловишь голый Exception — будешь ловить KeyboardInterrupt

Для интервью: я понимаю, что исключение в БД — это признак того, что транзакция в broken state и нужно откатить. Я знаю разные типы ошибок и как их обрабатывать правильно.