Перейти к содержанию

Управление структурой таблиц

В Planiqum для декларативного управления структурой таблиц используется класс Table из модуля planiqum.core.libs.table. Этот инструмент позволяет описывать структуру таблицы (колонки, индексы, ограничения) и автоматически синхронизировать её с базой данных.

Основные возможности

Класс Table предоставляет следующие возможности:

  • Загрузка структуры из БД: Загрузить существующую структуру таблицы из базы данных
  • Создание структуры: Описать новую структуру таблицы программно
  • Создание таблицы: Создать таблицу в базе данных по описанию
  • Обновление структуры: Обновить существующую таблицу в соответствии с описанием
  • Синхронизация: Автоматически создать или обновить таблицу в зависимости от её существования

Базовые классы

ColumnDefinition

Файл: src/planiqum/core/libs/db/table.py

Назначение: Описание колонки таблицы.

Параметры: - name (str): Имя колонки - type (str): SQL тип колонки (например, 'INTEGER', 'VARCHAR(255)', 'TEXT', 'SERIAL', 'TIMESTAMP') - nullable (bool, default=True): Может ли колонка содержать NULL значения - default (Optional[str], default=None): SQL выражение для значения по умолчанию - generated (Optional[str], default=None): SQL выражение для generated column. Если указано, колонка будет создана как GENERATED ALWAYS AS (expression) STORED

Примеры:

from planiqum.core.libs.db import ColumnDefinition

# Простая колонка
col1 = ColumnDefinition("id", "SERIAL", nullable=False)

# Колонка с default значением
col2 = ColumnDefinition("name", "VARCHAR(255)", nullable=True, default="'unnamed'")

# Колонка с timestamp по умолчанию
col3 = ColumnDefinition("created_at", "TIMESTAMP", default="NOW()")

# Generated column
col4 = ColumnDefinition(
    "full_name", 
    "TEXT", 
    generated="first_name || ' ' || last_name"
)

IndexDefinition

Файл: src/planiqum/core/libs/db/table.py

Назначение: Описание индекса таблицы.

Параметры: - name (str): Имя индекса - columns (List[str]): Список имен колонок, по которым создается индекс - unique (bool, default=False): Является ли индекс уникальным - primary (bool, default=False): Является ли индекс первичным ключом - type (IndexType, default=IndexType.BTREE): Тип индекса PostgreSQL

Типы индексов (IndexType): - BTREE - B-дерево (по умолчанию) - GIN - Generalized Inverted Index - GIST - Generalized Search Tree - HASH - Хеш-индекс - SPGIST - Space-partitioned GiST - BRIN - Block Range Index

Примеры:

from planiqum.core.libs.db import IndexDefinition, IndexType

# Уникальный индекс
idx1 = IndexDefinition("idx_user_email", ["email"], unique=True)

# Первичный ключ
idx2 = IndexDefinition("pk_users", ["id"], primary=True)

# GIN индекс для полнотекстового поиска
idx3 = IndexDefinition("idx_user_name_gin", ["name"], type=IndexType.GIN)

# Составной индекс
idx4 = IndexDefinition("idx_user_name_email", ["name", "email"])

ConstraintDefinition

Файл: src/planiqum/core/libs/db/table.py

Назначение: Описание ограничения таблицы.

Параметры: - name (str): Имя ограничения - type (ConstraintType): Тип ограничения (CHECK, FOREIGN_KEY, UNIQUE) - definition (str): SQL выражение для ограничения

Типы ограничений (ConstraintType): - CHECK - проверочное ограничение - FOREIGN_KEY - внешний ключ - UNIQUE - уникальное ограничение (не индекс)

Примеры:

from planiqum.core.libs.db import ConstraintDefinition, ConstraintType

# CHECK constraint
check = ConstraintDefinition(
    "check_age", 
    ConstraintType.CHECK, 
    "age >= 0 AND age <= 150"
)

# FOREIGN KEY constraint
fk = ConstraintDefinition(
    "fk_user", 
    ConstraintType.FOREIGN_KEY, 
    "FOREIGN KEY (user_id) REFERENCES users(id)"
)

# UNIQUE constraint
unique = ConstraintDefinition(
    "unique_email_username", 
    ConstraintType.UNIQUE, 
    "UNIQUE (email, username)"
)

Работа с классом Table

Создание структуры таблицы

Создание новой структуры:

from planiqum.core.libs.table import Table, ColumnDefinition, IndexDefinition

# Создание пустой структуры
table = Table.new("users")

# Создание структуры с начальными данными
table = Table.new("users", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)", nullable=False),
    ColumnDefinition("email", "VARCHAR(255)", nullable=False),
], indexes=[
    IndexDefinition("pk_users", ["id"], primary=True),
])

Программное добавление элементов:

table = Table.new("users")

# Добавление колонок
table.add_column(ColumnDefinition("id", "SERIAL", nullable=False))
table.add_columns([
    ColumnDefinition("name", "VARCHAR(255)"),
    ColumnDefinition("email", "VARCHAR(255)"),
])

# Добавление индексов
table.add_index(IndexDefinition("idx_email", ["email"], unique=True))

# Добавление ограничений
from planiqum.core.libs.db import ConstraintDefinition, ConstraintType
table.add_constraint(ConstraintDefinition(
    "check_email",
    ConstraintType.CHECK,
    "email LIKE '%@%'"
))

Загрузка структуры из БД 

# Загрузка структуры существующей таблицы
table = Table.from_db("users")

# Загрузка из другой схемы
table = Table.from_db("my_table", schema="custom_schema")

# Проверка структуры
print(f"Колонок: {len(table.columns)}")
print(f"Индексов: {len(table.indexes)}")
print(f"Ограничений: {len(table.constraints)}")

# Доступ к элементам
for col in table.columns:
    print(f"{col.name}: {col.type}")

for idx in table.indexes:
    print(f"{idx.name}: {idx.columns}")

Обработка ошибок:

try:
    table = Table.from_db("non_existent_table")
except ValueError as e:
    print(f"Таблица не существует: {e}")

Создание таблицы

Базовое создание:

table = Table.new("users", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)", nullable=False),
])

# Создание таблицы (ошибка, если существует)
table.create()

Обработка существующих таблиц:

# Ошибка, если таблица существует (по умолчанию)
table.create()  # ValueError, если таблица существует

# Удалить и создать заново
table.create(if_exists="drop")

# Синхронизировать структуру
table.create(if_exists="sync", on_column_missing="add")

Обновление таблицы 

# Обновление существующей таблицы
table = Table.new("users", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)"),
    ColumnDefinition("email", "VARCHAR(255)"),  # Новая колонка
])

# Обновление (ошибка, если таблица не существует)
table.update()

# Создать, если не существует
table.update(if_not_exists="create")

Синхронизация структуры

Метод create_or_update - универсальный способ синхронизации структуры таблицы:

table = Table.new("users", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)"),
    ColumnDefinition("email", "VARCHAR(255)"),
])

# Автоматически создаст таблицу, если её нет, или обновит, если существует
table.create_or_update()

Опции синхронизации:

table.create_or_update(
    # Колонки
    allow_drop_columns=False,        # Разрешить удаление колонок
    allow_change_type=False,          # Разрешить изменение типа колонки
    on_type_mismatch="error",         # Действие при несовпадении типа: "error", "alter", "drop_and_recreate", "ignore"
    on_column_missing="add",          # Действие при отсутствии колонки: "add", "error", "ignore"

    # Индексы
    allow_drop_indexes=True,          # Разрешить удаление индексов
    allow_rename_indexes=True,        # Разрешить переименование индексов

    # Ограничения
    allow_drop_constraints=False,    # Разрешить удаление ограничений
    allow_rename_constraints=True,    # Разрешить переименование ограничений
)

Примеры использования опций:

# Простая синхронизация (только добавление)
table.create_or_update()

# Разрешить удаление колонок и изменение типов
table.create_or_update(
    allow_drop_columns=True,
    allow_change_type=True,
    on_type_mismatch="alter"
)

# Строгий режим (ошибка при любых несовпадениях)
table.create_or_update(
    on_column_missing="error",
    on_type_mismatch="error",
    allow_drop_indexes=False,
    allow_drop_constraints=False
)

Примеры использования

Пример 1: Создание простой таблицы 

from planiqum.core.libs.table import Table, ColumnDefinition, IndexDefinition

# Описываем структуру
table = Table.new("products", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)", nullable=False),
    ColumnDefinition("price", "NUMERIC(10,2)", nullable=False),
    ColumnDefinition("created_at", "TIMESTAMP", default="NOW()"),
], indexes=[
    IndexDefinition("pk_products", ["id"], primary=True),
    IndexDefinition("idx_products_name", ["name"]),
])

# Создаем таблицу
table.create()

Пример 2: Синхронизация структуры 

# Загружаем текущую структуру
table = Table.from_db("products")

# Добавляем новую колонку
table.add_column(ColumnDefinition("description", "TEXT"))

# Синхронизируем (добавит новую колонку)
table.create_or_update(on_column_missing="add")

Пример 3: Работа со схемами 

# Создание таблицы в пользовательской схеме
table = Table.new("my_table", schema="custom_schema", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
])

# Создание схемы (если нужно)
from planiqum.core.libs.db import execute_query
execute_query("CREATE SCHEMA IF NOT EXISTS custom_schema")

# Создание таблицы
table.create()

Пример 4: Generated columns 

table = Table.new("orders", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("quantity", "INTEGER", nullable=False),
    ColumnDefinition("price_per_unit", "NUMERIC(10,2)", nullable=False),
    ColumnDefinition(
        "total_price",
        "NUMERIC(10,2)",
        generated="quantity * price_per_unit"
    ),
], indexes=[
    IndexDefinition("pk_orders", ["id"], primary=True),
])

table.create()

Пример 5: Комплексная структура 

from planiqum.core.libs.db import (
    Table, ColumnDefinition, IndexDefinition, ConstraintDefinition,
    IndexType, ConstraintType
)

table = Table.new("users", columns=[
    ColumnDefinition("id", "SERIAL", nullable=False),
    ColumnDefinition("email", "VARCHAR(255)", nullable=False),
    ColumnDefinition("name", "VARCHAR(255)"),
    ColumnDefinition("age", "INTEGER"),
], indexes=[
    IndexDefinition("pk_users", ["id"], primary=True),
    IndexDefinition("idx_users_email", ["email"], unique=True),
    IndexDefinition("idx_users_name_gin", ["name"], type=IndexType.GIN),
], constraints=[
    ConstraintDefinition(
        "check_age",
        ConstraintType.CHECK,
        "age >= 0 AND age <= 150"
    ),
    ConstraintDefinition(
        "check_email",
        ConstraintType.CHECK,
        "email LIKE '%@%'"
    ),
])

table.create()

Обработка ошибок

Таблица не существует:

try:
    table = Table.from_db("non_existent_table")
except ValueError as e:
    print(f"Ошибка: {e}")

Таблица уже существует:

try:
    table.create()  # По умолчанию if_exists="error"
except ValueError as e:
    print(f"Таблица уже существует: {e}")
    # Варианты решения:
    # table.create(if_exists="drop")  # Удалить и создать
    # table.create(if_exists="sync")  # Синхронизировать

Несовпадение типов:

try:
    table.create_or_update(on_type_mismatch="error")
except ValueError as e:
    print(f"Несовпадение типа: {e}")
    # Варианты решения:
    # table.create_or_update(on_type_mismatch="alter", allow_change_type=True)
    # table.create_or_update(on_type_mismatch="drop_and_recreate", allow_drop_columns=True)

Рекомендации

Когда использовать

Используйте класс Table для: - Создания таблиц в миграциях и скриптах инициализации - Синхронизации структуры таблиц между окружениями - Программного управления структурой таблиц - Автоматизации создания таблиц в тестах

Не используйте для: - Частых изменений структуры в production (используйте миграции Django) - Создания таблиц, управляемых Django ORM (используйте модели Django)

Безопасность операций

Безопасные операции (по умолчанию): - Добавление колонок - Добавление индексов - Добавление ограничений - Изменение nullable и default

Опасные операции (требуют явного разрешения): - Удаление колонок (allow_drop_columns=True) - Изменение типа колонки (allow_change_type=True) - Удаление индексов (allow_drop_indexes=True, по умолчанию разрешено) - Удаление ограничений (allow_drop_constraints=True)

Рекомендации: - В production используйте строгий режим (on_type_mismatch="error", allow_drop_columns=False) - В development можно использовать более гибкие настройки - Всегда проверяйте изменения перед применением в production

Тестирование

Тесты для класса Table находятся в src/planiqum/core/tests/libs/db/test_table.py.

Запуск тестов:

pytest src/planiqum/core/tests/libs/db/test_table.py -v

Пример теста:

import pytest
from planiqum.core.libs.table import Table, ColumnDefinition

@pytest.mark.django_db
def test_create_table(django_db_blocker):
    django_db_blocker.unblock()

    table = Table.new("test_table", columns=[
        ColumnDefinition("id", "SERIAL", nullable=False),
    ])
    table.create()

    # Проверяем создание
    loaded_table = Table.from_db("test_table")
    assert len(loaded_table.columns) == 1

Связанные темы

Важно: описанные настройки и сценарии могут отличаться в вашей инсталляции Planiqum
За уточнениями и методологической поддержкой обращайтесь в компанию ЮНИК СОФТ

— © 2025 ООО «ЮНИК СОФТ». Planiqum — платформа интегрированного бизнес-планирования. www.uniquesoft.ru