OperationalError

Исключение OperationalError относится к исключениям базы данных в Django и наследуется от DatabaseError. Оно возникает в ситуациях, связанных с работой базы данных, которые не являются ошибками программирования: проблемы с подключением, потеря соединения, таймауты, недоступность сервера базы данных, превышение лимитов соединений или блокировки таблиц. Это исключение оборачивает соответствующие ошибки драйвера базы данных (например, psycopg2.OperationalError для PostgreSQL или mysql.connector.OperationalError для MySQL) и предоставляет единообразный интерфейс для их обработки в приложении.

Синтаксис

pythonfrom django.db import OperationalError

try:
  # Код, работающий с базой данных
  pass
except OperationalError as e:
  # Обработка ошибки
  print(f"Operational error: {e}")

Пример . Перехват исключения при подключении

Рассмотрим попытку выполнить запрос при недоступной базе данных:

pythonfrom django.db import OperationalError
from myapp.models import Article

try:
  articles = Article.objects.all()
  print(f"Found {articles.count()} articles")
except OperationalError as e:
  print(f"Database is unavailable: {e}")

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

python"Database is unavailable: could not connect to server: Connection refused"

Пример . Обработка потери соединения

Обработаем ситуацию, когда соединение с базой данных было потеряно во время выполнения длительной операции:

pythonfrom django.db import OperationalError, connection
from myapp.models import Article
import time

def process_articles():
  try:
    # Длительная операция, которая может вызвать потерю соединения
    for article in Article.objects.all():
      time.sleep(0.1)  # Имитация длительной обработки
      process_article(article)
  except OperationalError as e:
    print(f"Connection lost during processing: {e}")
    # Попытка переподключения
    connection.close()
    connection.ensure_connection()
    print("Reconnected successfully")
    
def process_article(article):
  print(f"Processing: {article.title}")

Результат выполнения кода при потере соединения:

python"Processing: article 1"
"Processing: article 2"
"Connection lost during processing: server closed the connection unexpectedly"
"Reconnected successfully"

Пример . Обработка таймаута запроса

Обработаем ситуацию превышения времени выполнения запроса:

pythonfrom django.db import OperationalError
from myapp.models import Article
from django.db import connection

try:
  with connection.cursor() as cursor:
    cursor.execute("SELECT * FROM myapp_article WHERE id > 1000")
    results = cursor.fetchall()
    print(f"Found {len(results)} articles")
except OperationalError as e:
  if "timeout" in str(e).lower():
    print("Query timeout exceeded, please try again later")
  else:
    print(f"Database operation failed: {e}")

Результат выполнения кода при таймауте:

python"Query timeout exceeded, please try again later"

Пример . Обработка блокировки таблицы

Обработаем ситуацию, когда таблица заблокирована другой транзакцией:

pythonfrom django.db import OperationalError, transaction
from myapp.models import Article

def update_articles():
  try:
    with transaction.atomic():
      Article.objects.update(status='processed')
  except OperationalError as e:
    if "deadlock" in str(e).lower() or "lock" in str(e).lower():
      print("Table is locked, please try again later")
    else:
      print(f"Update failed: {e}")

Пример . Повторных попыток при ошибке

Реализуем механизм повторных попыток при возникновении OperationalError:

pythonfrom django.db import OperationalError
from myapp.models import Article
import time

def get_articles_with_retry(max_retries=3, delay=1):
  for attempt in range(max_retries):
    try:
      return list(Article.objects.all()[:10])
    except OperationalError as e:
      print(f"Attempt {attempt + 1} failed: {e}")
      if attempt < max_retries - 1:
        print(f"Retrying in {delay} seconds...")
        time.sleep(delay)
      else:
        print("All retry attempts failed")
        raise
  return []

try:
  articles = get_articles_with_retry()
  print(f"Retrieved {len(articles)} articles")
except OperationalError:
  print("Unable to retrieve articles after multiple attempts")

Результат выполнения кода при временных проблемах с БД:

python"Attempt 1 failed: could not connect to server"
"Retrying in 1 seconds..."
"Attempt 2 failed: could not connect to server"
"Retrying in 1 seconds..."
"Attempt 3 succeeded"
"Retrieved 10 articles"

Смотрите также