Как понять что база данных находится в консистентном состоянии?

Как понять, что база данных находится в консистентном состоянии

Консистентность базы данных — это одно из ключевых свойств ACID моделей. Понимание и проверка консистентности важны для надёжности приложения и целостности данных.

Что такое консистентность

База данных находится в консистентном (согласованном) состоянии, когда все ограничения целостности данных соблюдаются:

• Первичные ключи уникальны
• Внешние ключи указывают на существующие строки
• Проверяются ограничения (constraints)
• Данные соответствуют бизнес-правилам

Ограничения целостности (Constraints)

# PRIMARY KEY — каждая строка имеет уникальный идентификатор
CREATE TABLE users (
    id INT PRIMARY KEY,
    email VARCHAR(255) UNIQUE,
    age INT CHECK (age >= 18)
);

# FOREIGN KEY — связь между таблицами
CREATE TABLE orders (
    id INT PRIMARY KEY,
    user_id INT,
    FOREIGN KEY (user_id) REFERENCES users(id)
);

Проверка консистентности в Python

1. Проверка при подключении к БД

import sqlite3

# SQLite
connection = sqlite3.connect('database.db')
cursor = connection.cursor()

# Включить проверку внешних ключей
cursor.execute('PRAGMA foreign_keys = ON')

# Проверить консистентность БД
cursor.execute('PRAGMA integrity_check')
result = cursor.fetchall()
if result[0][0] == 'ok':
    print('База данных консистентна')
else:
    print(f'Ошибка консистентности: {result}')

2. PostgreSQL проверка

import psycopg2

connection = psycopg2.connect(
    'dbname=mydb user=postgres password=pass host=localhost'
)
cursor = connection.cursor()

# Проверить внешние ключи
cursor.execute("""
    SELECT constraint_name
    FROM information_schema.table_constraints
    WHERE constraint_type = 'FOREIGN KEY'
"")

foreign_keys = cursor.fetchall()
print(f'Найдено внешних ключей: {len(foreign_keys)}')

# Проверить нарушения внешних ключей
cursor.execute("""
    SELECT orders.id, orders.user_id
    FROM orders
    LEFT JOIN users ON orders.user_id = users.id
    WHERE users.id IS NULL
"")

orphan_orders = cursor.fetchall()
if orphan_orders:
    print(f'Найдены сирротствующие заказы: {orphan_orders}')

3. SQLAlchemy с проверкой

from sqlalchemy import create_engine, event, text

engine = create_engine('postgresql://user:password@localhost/mydb')

# Включить проверку внешних ключей при подключении
@event.listens_for(engine, 'connect')
def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute('PRAGMA foreign_keys=ON')
    cursor.close()

with engine.connect() as conn:
    # Проверить целостность
    result = conn.execute(text('PRAGMA integrity_check'))
    integrity = result.fetchone()
    print(f'Целостность БД: {integrity}')

Проверка данных на уровне приложения

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import IntegrityError

engine = create_engine('postgresql://user:password@localhost/mydb')
Session = sessionmaker(bind=engine)

def verify_database_consistency():
    """Проверить консистентность на уровне приложения"""
    session = Session()
    errors = []
    
    # Проверка 1: Нет сирротствующих заказов
    orphan_orders = session.query(Order).filter(
        ~Order.user_id.in_(
            session.query(User.id)
        )
    ).count()
    
    if orphan_orders > 0:
        errors.append(f'Найдено {orphan_orders} сирротствующих заказов')
    
    # Проверка 2: Уникальность email
    duplicate_emails = session.query(User.email).group_by(
        User.email
    ).having(func.count(User.id) > 1).all()
    
    if duplicate_emails:
        errors.append(f'Найдены дублирующиеся emails: {duplicate_emails}')
    
    # Проверка 3: Баланс в пределах разумного
    negative_balance = session.query(Account).filter(
        Account.balance < 0
    ).count()
    
    if negative_balance > 0:
        errors.append(f'Найдены счета с отрицательным балансом')
    
    session.close()
    return errors if errors else ['База данных консистентна']

result = verify_database_consistency()
for msg in result:
    print(msg)

Транзакции и консистентность

from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import IntegrityError

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

try:
    user = User(name='John', email='john@example.com')
    session.add(user)
    session.flush()
    
    order = Order(user_id=user.id, amount=100)
    session.add(order)
    
    session.commit()
    print('Транзакция успешна, БД консистентна')
    
except IntegrityError as e:
    session.rollback()
    print(f'Нарушение целостности: {e}')
except Exception as e:
    session.rollback()
    print(f'Ошибка: {e}')
finally:
    session.close()

Основные способы проверки

1. PRAGMA integrity_check (SQLite)
2. PRAGMA foreign_keys (SQLite)
3. Валидация на уровне приложения
4. Логирование и мониторинг
5. Регулярные бэкапы
6. Тесты данных