При обновлении версии приложения важным показателем является обеспечение нулевого времени простоя (zero-downtime). Чтобы его достичь, новую версию выпускают параллельно с текущей рабочей версией (rolling updates, canary deployment, blue-green deployment).

Примерная схема процесса обновления:

• Применение миграций

• Поочерёдное отключение экземпляров от балансировщика нагрузки

• Перезапуск и возврат к балансировщику нагрузки

Поэтому применяемые миграции должны быть обратно совместимыми:

В противном случае старая версия приложения может попытаться обратиться к столбцу или таблице, которых уже нет в базе данных.

Пример py-миграции:

# file: migrations/0001.initial.py
from yoyo import step

steps = [
   step(
       "CREATE TABLE foo (id INT, bar VARCHAR(20), PRIMARY KEY (id))",
       "DROP TABLE foo"
   )
]

Здесь step — это шаг миграции, содержащий код применения миграции (forward) и её отката (backward).

В случае с SQL-миграциями у нас было бы два файла: 0001.initial.sql и 0001.initial.rollback.sql:

--
-- file: migrations/0001.initial.sql
--
CREATE TABLE foo (id INT, bar VARCHAR(20), PRIMARY KEY (id));

--
-- file: migrations/0001.initial.rollback.sql
--
DROP TABLE foo;

from pglast import ast
from pglast.ast import RangeVar
from pglast.enums.parsenodes import ConstrType
from pglast.visitors import Visitor, Ancestor


class PKViolation:
	"""Класс ошибки отсутствия PRIMARY KEY."""
    error_template: str = "Table {table_name!r} has no PRIMARY KEY"
    code: str = "001"

    def __init__(self, **context) -> None:
        self._context= context

    @property
    def message(self) -> str:
        error = self.error_template.format(**self._context)

        return f"M{self.code} {error}"


class PrimaryKeyChecker(Visitor):
    """
    Класс, проверяющий наличие PRIMARY KEY в CREATE TABLE.
    """

    def __init__(self) -> None:
        super().__init__()
        self.violations: list[PKViolation] = []

    def visit_CreateStmt(self, ancestors: Ancestor, node: ast.CreateStmt):
	    """
	    Обработка узла `CreateStmt` AST-дерева.
	    """
        rel: RangeVar = node.relation
        schema = rel.schemaname + "." if getattr(rel, "schemaname", None) else ""
        table_name = f"{schema}{rel.relname}"

        if not self._has_primary_key(node):
            self.violations.append(
                PKViolation(table_name=table_name)
            )

    def _has_primary_key(self, node: ast.CreateStmt) -> bool:
        """
        Проверяем оба варианта указания PK:

        1) на уровне столбца:  id bigint PRIMARY KEY
           => ColumnDef + Constraint(CONSTR_PRIMARY) в ColumnDef.constraints

        2) на уровне таблицы:   PRIMARY KEY (id, ...)
           => Constraint(CONSTR_PRIMARY) напрямую в tableElts
        """
        for elt in node.tableElts or ():
            # Вариант 1: колонка с ограничениями
            if isinstance(elt, ast.ColumnDef):
                for con in elt.constraints or ():
                    if (
                        isinstance(con, ast.Constraint)
                        and con.contype == ConstrType.CONSTR_PRIMARY
                    ):
                        return True

            # Вариант 2: табличное ограничение
            elif isinstance(elt, ast.Constraint):
                if elt.contype == ConstrType.CONSTR_PRIMARY:
                    return True

        return False

import sys

from pglast import ast, parse_sql


def check_sql(sql: str) -> list[PKViolation]:
    """
    Выполняет парсинг SQL и анализ выражений CREATE TABLE.
    """
    stmts: tuple[ast.RawStmt, ...] = parse_sql(sql)
    visitor = PrimaryKeyChecker()
    visitor(stmts)
    return visitor.violations


def main() -> None:
    sql = """
    --  Таблица без ПК
    CREATE TABLE public.users (
        id bigint GENERATED BY DEFAULT AS IDENTITY,
        email text NOT NULL,
        created_at timestamptz NOT NULL DEFAULT now()
    );
	
	--  ПК на уровне столбца
    CREATE TABLE public.orders (
        id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
        user_id bigint NOT NULL,
        created_at timestamptz NOT NULL DEFAULT now()
    );
	
	--  ПК на уровне таблицы
    CREATE TABLE logs (
        ts timestamptz NOT NULL,
        payload jsonb NOT NULL,
        PRIMARY KEY (ts)  
    );
    """

    violations = check_sql(sql)

    if violations:
        for v in violations:
            print(v.message)

        sys.exit(1)


if __name__ == "__main__":
    main()

@dataclass
class MigrationStep:
    """
    Шаг миграции.

    Содержит информацию о конкретном шаге миграции, включая SQL-запросы
    и их AST-представления для выполнения и отката изменений.

    Attributes:
        forward_sql: SQL-запрос для применения миграции.
            Содержит команды для перехода на новую версию схемы данных.

        forward_ast: AST-представление forward запроса.
            Кортеж абстрактных синтаксических деревьев, полученных
            в результате парсинга forward_sql. Используется для анализа
            и валидации SQL-команд.
        backward_sql: SQL-запрос для отката миграции.
            Содержит команды для возврата к предыдущей версии схемы данных.
            Может быть None, если откат миграции не предусмотрен.
        backward_ast: AST-представление backward запроса.
            Кортеж абстрактных синтаксических деревьев, полученных
            в результате парсинга backward_sql. None, если миграция
            отката отсутствует.
    """

    forward_sql: str
    forward_ast: tuple[AST, ...]
    backward_sql: str | None
    backward_ast: tuple[AST, ...] | None


@dataclass
class Migration:
    """Полное описание миграции.

    Содержит метаинформацию и последовательность шагов для выполнения миграции.

    Attributes:
        transactional: Флаг выполнения в транзакции.
            Если True, все шаги миграции выполняются в одной транзакции.
            Если False, каждый шаг выполняется автономно.

        path: Путь к файлу миграции.
        steps: Упорядоченный список шагов миграции.
    """
    transactional: bool
    path: str
    steps: list[MigrationStep]


def read_migrations(
    path: Path,
    config: MLintConfig,
) -> list[Migration]:
    # yoyo-migrations и т.д.
    migrations_backend = get_migrations_backend(config)
    # PostgreSQL и другие СУБД
    database_backend = get_database_backend(config)

    raw_migrations = migrations_backend.load_migrations(
        path, config.exclude
    )

    return database_backend.parse_migrations(raw_migrations)

class BaseViolation:
    """
    Базовый класс для всех нарушений.

    Attributes:
        error_template: Шаблон сообщения об ошибке.
        code: Уникальный код ошибки.
        message: Форматированное сообщение об ошибке.
        filename: Путь к файлу с ошибкой.
    """

    error_template: ClassVar[str]
    code: ClassVar[int]
    message: str
    filename: str | None
    
    def __init__(
        self,
        filename: str | None = None,
        error_prefix: str = "mlint",
        **context: Any,
    ) -> None:
	    """Инициализация, форматирование сообщения об ошибке и т.д."""
	    
	    message = self.format_message(**self.get_context(**context))
	    
	    full_code = str(self.code).zfill(3)  # 1 -> 001
	    self.message = f"{error_prefix}{full_code} {message}"
	
	@classmethod
    def format_message(cls, **context: Any) -> str:
        """Форматирование сообщения об ошибке."""
        return cls.error_template.format(**context)

    def get_context(self, **context: Any) -> GenericContext:
        """Хук для изменения контекста."""
        return context

	
class Application:
    """
    Основной класс приложения MLint, отвечающий за организацию 
    процесса линтинга.

    Attributes:
        _migrations: Список миграций для проверки.
        _violations: Список найденных нарушений.
        _plugins: Плагины.
        _config: Конфигурация линтера.
        _formatter: Форматтер вывода для отображения результатов.
    """

    _migrations: list[Migration]
    _violations: list[BaseViolation]
    _plugins: Plugins
    _config: MLintConfig
    _formatter: BaseFormatter

    ...

    def run(self, argv: list[str]) -> None:
        """
        Запускает основной процесс линтинга.

        Args:
            argv: Список аргументов командной строки.
        """
        try:
            self._initialize(argv)
            self._run_checks()
            self._report()
        except KeyboardInterrupt:
            self._execution_failure = True
        except MLintCriticalError:
            self._execution_failure = True

    def exit(self) -> None:
        """
        Завершает работу приложения с корректным кодом возврата:

        Код 0 при успехе, 1 при наличии нарушений или ошибок.
        """
        raise SystemExit((self._result_count > 0) or self._execution_failure)

# pyproject.toml
[tool.mlint]  
baseline-path = "mlint-baseline.txt"  
  
ignore = ["M009"]  
exclude = ["migrations/20250603-add-users-table.py"]  
  
[tool.mlint.per-file-ignores]  
"**add-users**" = ["M002"]

# setup.cfg
[mlint]
baseline-path = mlint-baseline.txt

ignore = 
    M009

exclude =
    migrations/20250603-add-users-table.py

per-file-ignores =
    **add-users**: M002

mlint \
  --baseline-path mlint-baseline.txt \
  --ignore M009 \
  --exclude migrations/20250603-add-users-table.py
  --config pyproject.toml

_NOT_PROVIDED = object()  
  
  
class _Option:  
    """
    Обёртка над опциями из argparse.ArgumentParser.add_argument.
    
    Позволяет сопоставить короткие и длинные имена аргументов (-f, -foo)
    с атрибутами конфигураций.
    """
      
    def __init__(  
        self,  
        short_option_name: str = _NOT_PROVIDED,  
        long_option_name: str = _NOT_PROVIDED,  
        parse_from_config: bool = False,  
        # Kwargs из argparse.ArgumentParser.add_argument  
        **kwargs: Any,  
    ) -> None:  
        # если указана только --опция, то это long_option_name
        if (  
            long_option_name is _NOT_PROVIDED  
            and short_option_name is not _NOT_PROVIDED  
            and short_option_name.startswith("--")  
        ):  
            short_option_name, long_option_name = (  
                _NOT_PROVIDED,  
                short_option_name,  
            )
  
        self.option_args = [  
            opt for opt in (short_option_name, long_option_name)  
            if opt is not _NOT_PROVIDED  
        ]  
        self.option_kwargs: dict[str, Any] = kwargs  
        self.dest = kwargs.get("dest", None)  
        self.parse_from_config = parse_from_config  
        self.config_name: str | None = None  
        if self.parse_from_config:  
            if long_option_name is _NOT_PROVIDED:  
                raise ValueError(  
                    "When specifying parse_from_config=True, "  
                    "a long_option_name must also be specified."               
                )  
            # Преобразуем --option-name в option_name для конфига
            self.config_name = long_option_name[2:].replace("-", "_")  
  
    def to_argparse(self) -> tuple[list[str], dict[str, Any]]:  
        return self.option_args, self.option_kwargs

class OptionManager:
    """Менеджер опций команной строки."""

    def __init__(self, parents: list[argparse.ArgumentParser]) -> None:
        self._parser = argparse.ArgumentParser(
            prog="mlint",
            usage="%(prog)s [options] file or dir ...",
            parents=parents,
        )
        self._parser.add_argument(
            "filenames",
            nargs="*",
            metavar="filename",
        )
		
		"""Словарь для связи опций и атрибутов конфигурации"""
        self.config_options_dict: dict[str, _Option] = {}
        self.options: list[_Option] = []

    def add(self, *args: Any, **kwargs: Any) -> None:
        """Добавляет CLI-аргумент."""
        option = _Option(*args, **kwargs)
        option_args, option_kwargs = option.to_argparse()

        action = self._parser.add_argument(*option_args, **option_kwargs)
        option.dest = action.dest

        self.options.append(option)
        if option.parse_from_config and option.config_name:
            name = option.config_name
            self.config_options_dict[name] = option
            self.config_options_dict[name.replace("_", "-")] = option

    def register_plugins(self, *plugins: LoadedPlugin) -> None:
        """Регистрация опций из плагинов."""
        for plugin in plugins:
            plugin.add_options(self)

    def parse_args(
        self,
        args: Sequence[str] | None = None,
        values: argparse.Namespace | None = None,
    ) -> argparse.Namespace:
        """
        Парсинг CLI-аргументов со значениями по умолчанию из конфигурации.
        """
        if values:
            self._parser.set_defaults(**vars(values))
        return self._parser.parse_args(args)

import dataclasses    
from pathlib import Path  
from typing import Any, Literal, Self  
  
from mlint.options.manager import OptionManager  
from mlint.options.utils import comma_separated_list  
  
  
@dataclasses.dataclass  
class BaseConfig:  
    """Базовый класс конфигурации для линтера и плагинов."""  
  
    @classmethod  
    def from_args(cls, args: dict[str, Any]) -> Self:  
        """Инициализирует конфигурацию данными CLI-аргументов."""
        raise NotImplementedError()  
  
    @classmethod  
    def add_options(cls, manager: OptionManager) -> None:  
        """Регистрирует CLI-аргументы для атрибутов конфигурации."""
        raise NotImplementedError()  
  
    def read_args(self, args: dict[str, Any]) -> Self:  
        """Возвращает копию конфигурации с обновлёнными данными."""  
        config = dataclasses.replace(self)  
        for field in dataclasses.fields(self):  
            value = args.get(field.name)  
            if value is not None:  
                setattr(config, field.name, value)  
        return config  
  
  
@dataclasses.dataclass  
class MLintConfig(BaseConfig):  
    """Класс конфигурации линтера."""  
  
    baseline_path: Path = Path("mlint-baseline.txt")  
    format: str = "default"  
    exclude: list[str] = dataclasses.field(default_factory=list)  
    ignore: list[str] = dataclasses.field(default_factory=list)  
    per_file_ignores: dict[str, list[str]] = dataclasses.field(  
        default_factory=dict  
    )  
    ...  # остальные аргументы  
  
    @classmethod  
    def from_args(cls, args: dict[str, Any]) -> MLintConfig:  
        """Инициализирует конфигурацию данными CLI-аргументов."""  
        config = cls()  
        config = config.read_args(args)  
  
        if isinstance(config.baseline_path, str):  
            config.baseline_path = Path(config.baseline_path) 
        
		...  # нормализация аргументов, приведение типов и т.д.
  
        return config  
  
    @classmethod  
    def add_options(cls, manager: OptionManager) -> None:  
        """Регистрирует CLI-аргументы для атрибутов конфигурации."""  
        manager.add(  
            "--baseline-path",  
            type=Path,  
            parse_from_config=True,  
            help="Путь к baseline-файлу.",  
        )  
        manager.add(  
            "--ignore",  
            metavar="errors",  
            parse_from_config=True,  
            type=comma_separated_list(str),  
            help=(  
                "Список кодов ошибок через запятую, "  
                "которые должны быть проигнорированы. "  
                "Например, ``--ignore=M001,M103``."            
            ),  
        )  
        ...  # остальные аргументы

import argparse  
from pathlib import Path  
from typing import Any, Sequence  
  
from mlint.config import MLintConfig  
from mlint.plugins.finder import Plugins, load_plugins


def parse_base_args(  
    argv: Sequence[str],  
) -> tuple[argparse.ArgumentParser, argparse.Namespace, list[str]]:
	"""
	Создаёт базовый парсер и возвращает базовые и оставшиеся аргументы
	для последующей обработки.
	""" 
    base_parser = argparse.ArgumentParser(add_help=False)  
    base_parser.add_argument(  
        "--config", default=None, help="Путь к файлу конфигурации",  
    )  
    base_parser.add_argument(  
        "--output-file", default=None, help="Перенаправить вывод в файл."  
    )  
    base_options, remaining_args = base_parser.parse_known_args(argv)  
    if base_options.output_file:  
        remaining_args.extend(("--output-file", base_options.output_file))  
      
    return base_parser, base_options, remaining_args  


def parse_args(  
    argv: Sequence[str],  
) -> tuple[MLintConfig, Plugins, argparse.Namespace]:  
    base_parser, base_options, remaining_args = parse_base_args(argv)  
    
    # дополняем базовый парсер аргументами OptionManager и конфигурации
    option_manager = OptionManager(parents=[base_parser])  
    MLintConfig.add_options(option_manager)  
	
	# загружаем файлы конфигурации  
    config, base_dir = load_config(base_options.config)  
    
    # загружаем локальные и глобальные плагины
    plugins = load_plugins(config, base_dir)  
    
    # дополняем базовый парсер CLI-аргументами плагинов
    option_manager.register_plugins(*plugins.all_plugins())  
	
	# парсим оставшиеся аргументы и объединяем конфигурацию
    opts = _aggregate_options(option_manager, config, remaining_args)  
    
    # инициализируем классы конфигурации линтера и плагинов
    mlint_config = MLintConfig.from_args(vars(opts))  
	plugins.parse_options(opts)
  
    return mlint_config, plugins, opts
    
    
def _aggregate_options(  
    manager: OptionManager,  
    config: dict[str, Any],  
    argv: list[str],  
) -> argparse.Namespace:  
	# получаем дефолтные значения CLI-аргументов
    default_values = manager.parse_args([])  
  
	# заменяем дефолтные значения на значения из файлов конфигурации
    for config_name, value in config.items():  
        dest_name = config_name  
        
        # если имя CLI-аргумента отличается от атрибута конфигурации,
        # то достаём соответствующую опцию из OptionManager
        # и получаем dest - имя, по которому значение
        # CLI-аргумента будет доступно после парсинга
        if not hasattr(default_values, config_name):  
            option = manager.config_options_dict.get(config_name)  
            if not option:  
                continue  
  
            if option.dest:  
                dest_name = option.dest  
  
        setattr(default_values, dest_name, value)  
    
    # парсим оригинальные CLI-аргументы с учётом дефолтных значений
    # из CLI-аргументов и конфигурации
    return manager.parse_args(argv, default_values)

from abc import ABC, abstractmethod
from typing import Any, final, Generic

import pglast.visitors

from mlint.core.models import Migration
from mlint.core.violations import BaseViolation
from mlint.types import TConfig


class BaseVisitor(ABC, Generic[TConfig]):
    """
    Базовый класс для различных типов visitor-ов.

    Type Parameters:
        TConfig: Тип объекта конфигурации (линтера или плагина).

    Attributes:
        _migration: Объект миграции для анализа.
        _config: Объект конфигурации.
        _violations: Список найденных нарушений.
    """

    _migration: Migration
    _config: TConfig | None
    _violations: list[BaseViolation]

    def __init__(
        self,
        migration: Migration,
        config: TConfig | None = None,
        error_prefix: str = "mlint",
    ) -> None:
        """Инициализация visitor-а."""
        self._migration = migration
        self._config = config
        self._violations = []
        self._error_prefix = error_prefix

    @abstractmethod
    def run(self) -> None:
        """Запуск анализа миграции."""
        raise NotImplementedError()

    @final
    def add_violation(
        self,
        cls: type[BaseViolation],
        **kwargs: Any,
    ) -> None:
        """Добавляет найденное нарушение."""
        violation = cls(
            filename=self._migration.path,
            error_prefix=self._error_prefix,
            **kwargs,
        )
        self._violations.append(violation)

    @property
    @final
    def violations(self) -> list[BaseViolation]:
        """Возвращает список найденных нарушений."""
        return self._violations


class BaseNodeVisitor(BaseVisitor, pglast.visitors.Visitor, Generic[TConfig]):
    """
    Базовый класс, реализующий паттерн Visitor.

    Type Parameters:
        TConfig: Тип объекта конфигурации (линтера или плагина).
    """
    def run(self) -> None:
        nodes = []
        
        for step in self._migration.steps:
            nodes.append(step.forward_ast)
            if step.backward_ast:
                nodes.append(step.backward_ast)
        
        return self(nodes)

import os  
import sys  
from pathlib import Path  
from typing import IO  
  
from mlint.config import MLintConfig  
from mlint.core.violations import BaseViolation  
  
  
class BaseFormatter:  
    """Базовый класс форматера."""  
  
    def __init__(  
        self,   
        config: MLintConfig,   
        output_file: Path | None = None,  
    ) -> None:  
        self.config = config  
        self.filename = output_file  
        self.output_fd: IO[str] | None = None  
        self.newline = os.linesep  
  
    def start(self) -> None:  
        """Хук, вызывающийся до обработки всех найденных ошибок."""  
        if self.filename:  
            dirname = os.path.dirname(os.path.abspath(self.filename))  
            os.makedirs(dirname, exist_ok=True)  
            self.output_fd = open(self.filename, "a")  
  
    def handle(self, error: BaseViolation) -> None:  
        """Обрабатывает найденную ошибку."""  
        line = self.format(error)  
        self.write(line)  
  
    def format(self, error: BaseViolation) -> str | None:  
        """Возвращает строковое представление ошибки."""  
        raise NotImplementedError()  
  
    def _write(self, output: str) -> None:  
        """Записывает строку в файл или stdout."""  
        if self.output_fd is not None:  
            self.output_fd.write(output + self.newline)  
        else:  
            sys.stdout.buffer.write(output.encode() + self.newline.encode())  
  
    def write(self, line: str | None) -> None:  
        """Записывает непустую строку в файл или stdout."""  
        if line:  
            self._write(line)  
  
    def stop(self) -> None:  
        """Хук, вызывающийся после обработки всех найденных ошибок."""  
        if self.output_fd is not None:  
            self.output_fd.close()  
            self.output_fd = None  
  
  
class DefaultFormatter(BaseFormatter):  
    """Форматтер по умолчанию, выводящий ошибки в plain-text формате."""  
  
    # Ошибки в формате "migrations/0001.initial.py: M001 missing primary key  
    error_format = "{path}: {code} {message}"  
  
    def format(self, error: BaseViolation) -> str | None:  
        """Возвращает строковое представление ошибки."""  
        return self.error_format.format(  
            path=error.filename,  
            message=error.message,  
            code=error.formatted_code,  
        )

import traceback
from argparse import Namespace
from collections.abc import Generator
from typing import ClassVar, final, Generic

from mlint.core.models import Migration
from mlint.core.violations import BaseViolation, UnhandledExceptionViolation
from mlint.formatters.base import BaseFormatter
from mlint.types import TConfig
from mlint.options.manager import OptionManager
from mlint.visitors.base import BaseVisitor


class BasePlugin(Generic[TConfig]):
    """
    Базовый класс для всех типов плагинов MLint.

    Attributes:
        name: Уникальный идентификатор плагина.
        config_cls: Опциональный класс конфигурации.
        _config: Инициализированный объект конфигурации.
    """

    name: ClassVar[str]
    config_cls: type[TConfig] | None = None
    _config: TConfig | None = None

    @classmethod
    @final
    def add_options(cls, manager: OptionManager) -> None:
        """Регистрация CLI-аргументов плагина."""
        if cls.config_cls:
            cls.config_cls.add_options(manager)

    @classmethod
    @final
    def parse_options(cls, options: Namespace) -> None:
        """Парсинг CLI-аргументов и инициализация конфигурации."""
        if cls.config_cls:
            cls._config = cls.config_cls.from_args(vars(options))


class BaseCheckerPlugin(BasePlugin[TConfig], Generic[TConfig]):
    """
    Базовый класс для checker-плагинов, выполняющих анализ миграций.

    Attributes:
        mlint_extension: Уникальный код плагина, используемый 
                         как префикс кодов ошибок.
        visitors: Список классов visitor-ов, которые плагин 
                  применяет к каждой миграции.
    """

    mlint_extension: ClassVar[str]
    visitors: list[type[BaseVisitor]]

    def __init__(self, migration: Migration) -> None:
        self._migration = migration

    @final
    def run(self) -> Generator[BaseViolation, None, None]:
        """Запуск анализа миграции."""
        for visitor_class in self.visitors:
            visitor = visitor_class(
                migration=self._migration,
                config=self._config,
                error_prefix=self.mlint_extension,
            )

            try:
                visitor.run()
            except Exception:
                print(traceback.format_exc())

                # Сохраняем системную ошибку
                visitor.add_violation(UnhandledExceptionViolation)

            yield from visitor.violations


class BaseReporterPlugin(BasePlugin[TConfig]):
    """
    Базовый класс для reporter-плагинов, формирующих отчёт об ошибках.

    Attributes:
        formatter_cls: Класс форматтера.
    """

    formatter_cls: type[BaseFormatter]

Для настройки локальных плагинов определим секцию local-plugins:

[tool.mlint]
# ...

[tool.mlint.local-plugins]
# список путей, в которых будут искаться плогины относительно
# корня проекта/файла конфигурации
paths = ["."] 
# checker-плагины в формате "уникальный код" = "импортируемый путь"
extension = { LPE = "mlint_plugin:LocalPluginExample" }
# reporter-плагины в формате "формат" = "импортируемый путь"
report = {json = "mlint_plugin:LocalJSONReporterExample"}

В данном случае модуль mlint_plugin должен находиться в корне проекта.

Если плагины оформлены в виде библиотеки, то в pyproject.toml должны быть следующие entry points:

# checker-плагины в формате "уникальный код" = "импортируемый путь"
[project.entry-points.'mlint.extension']
PE = "mlint_plugin_package.plugin:PluginExample"

# reporter-плагины в формате "формат" = "импортируемый путь"
[project.entry-points.'mlint.report']
json = "mlint_plugin_package.plugin:JSONReporterExample"

import sys  
from collections.abc import Generator  
from dataclasses import dataclass  
from importlib.metadata import distributions, EntryPoint  
from operator import attrgetter  
from pathlib import Path  
from typing import Any  
  
from mlint.exceptions import PluginLoadingError  
from mlint.options.manager import OptionManager  
from mlint.plugins.base import (  
    BaseCheckerPlugin,  
    BasePlugin,  
    BaseReporterPlugin,  
)  
from mlint.types import PluginType  # EXTENSION, REPORT  
  
  
PKG_NAME = "mlint"  
LOCAL_PLUGINS_CONFIG_SECTION = "local-plugins"  
MLINT_GROUPS = frozenset(("mlint.extension", "mlint.report"))  
PLUGIN_TYPES = [plugin_type.value for plugin_type in list(PluginType)]  
CheckerPluginClass = type[BaseCheckerPlugin]  
ReporterPluginClass = type[BaseReporterPlugin]  
BasePluginClass = type[BasePlugin]  
  
  
@dataclass  
class PluginMeta:  
    """Метаданные плагина, содержащие информации о типе и источнике."""  
  
    package: str  
    type: PluginType  
    entry_point: EntryPoint  
  
  
class LoadedPlugin:  
    """Базовый класс-контейнер, представляющий загруженный плагин."""  
  
    cls: BasePluginClass  
  
    @classmethod  
    def add_options(cls, manager: OptionManager) -> None:
	    """Регистрация CLI-аргументов плагина.""" 
        return cls.cls.add_options(manager)  
  
  
@dataclass  
class CheckerPlugin(LoadedPlugin):  
    """Класс-контейнер, представляющий загруженный checker-плагин."""  
  
    cls: CheckerPluginClass  
  
  
@dataclass  
class ReporterPlugin(LoadedPlugin):  
    """Класс-контейнер, представляющий загруженный reporter-плагин."""  
  
    name: str  # название формата  
    cls: ReporterPluginClass
    

@dataclass(frozen=True)
class LocalPluginOptions:
    """
    Контейнер для конфигурации локальных плагинов, полученных из 
    секции local-plugins файла конфигурации.
    """

    paths: list[str]
    extension: dict[str, str]
    report: dict[str, str]

    @classmethod
    def empty(cls) -> PluginOptions:
        return cls(paths=[], extension={}, report={})
        

@dataclass
class Plugins:
    """
    Контейнер для всех найденных и загруженных плагинов.

    Attributes:
        checkers: Список загруженных checker-плагинов.
        reporters: Список загруженных reporter-плагинов.
    """

    checkers: list[CheckerPlugin]
    reporters: list[ReporterPlugin]

    def all_plugins(self) -> Generator[LoadedPlugin, None, None]:
        """Возвращает все плагины независимо от типа."""
        yield from self.checkers
        yield from self.reporters

from importlib.metadata import distributions, EntryPoint


def _parse_local_plugin_options(config: dict[str, Any]) -> LocalPluginOptions:
    """Поиск локальных плагинов в файле конфигурации."""
    local_plugins = config.get(LOCAL_PLUGINS_CONFIG_SECTION)
    if not local_plugins:
        return LocalPluginOptions.empty()

    plugins_by_type: dict[str, dict[str, str]] = {
        "extension": {},
        "report": {},
    }

    for plugin_type in PLUGIN_TYPES:
        local_plugins_paths = local_plugins.get(plugin_type, {})
        if not local_plugins_paths:
            continue

        plugins_by_type[plugin_type].update(local_plugins_paths)

    return LocalPluginOptions(
        paths=local_plugins.get("paths", []),
        extension=plugins_by_type["extension"],
        report=plugins_by_type["report"],
    )


def _normalize_path(path: str | Path, parent: str | Path = Path.cwd()) -> Path:
    """Нормализация пути относительно родительской директории."""
    p = Path(path)
    if not p.is_absolute():
        p = Path(parent) / p
    return p.resolve()


def _find_local_plugins(
    opts: LocalPluginOptions,
    base_dir: Path,
) -> Generator[PluginMeta, None, None]:
    """Поиск и загрузка локальных плагинов."""
    for path in opts.paths:
	    # нормализуем путь относительно корня проекта
        normalized_path = _normalize_path(path, base_dir)
        if normalized_path.exists():
	        # добавляем путь к плагину для возможности загрузки через EntryPoint
            sys.path.insert(0, str(normalized_path))
	
	# добавляем путь к корню проекта для возможности загрузки через EntryPoint
    sys.path.insert(0, str(base_dir))

    for plugin_type in PLUGIN_TYPES:
        group = f"{PKG_NAME}.{plugin_type}"  # например, mlint.report
        for name, entry_str in getattr(opts, plugin_type, {}).items():
			# name: LPE
			# entry_str: mlint_plugin:LocalPluginExample
            parts = entry_str.split(":")
            if len(parts) != 2:
                raise PluginLoadingError(
                    f"Plugin spec must be 'path.to.module:PluginClassName', "
                    f"got {entry_str!r}"
                )
			
			# создаём фейковый EntryPoint для последующей загрузки
            ep = EntryPoint(name, entry_str, group)
            yield PluginMeta(
                package=f"local-plugins.{plugin_type}.{name}",
                type=PluginType(plugin_type),
                entry_point=ep,
            )

from importlib.metadata import distributions, EntryPoint


def _find_importlib_plugins() -> Generator[PluginMeta, None, None]:
	"""
	Поиск плагинов в метаданных установленных пакетов.
	
	Функция выполняет проверку всех установленных пакетов:
	 - если для пакета определены entry points, то выполняется поиск
	   групп из MLINT_GROUPS.
    """
    seen = set()
    for dist in distributions():  # обход пакетов из site-packages
    
    	# [project.entry-points.'mlint.extension']
		# PE = "mlint_plugin_package.plugin:PluginExample"
        eps = dist.entry_points
		
        if not any(ep.group in MLINT_GROUPS for ep in eps):
            continue

        meta = dist.metadata

        if meta["name"] in seen:
            continue
        else:
            seen.add(meta["name"])

        for ep in eps:
            if ep.group in MLINT_GROUPS:
	            # ep.group: "mlint.extension" или "mlint.report"
                _, plugin_type = ep.group.split(".")

                yield PluginMeta(
                    package=meta["name"],
                    type=PluginType(plugin_type),
                    entry_point=ep,
                )

from typing import ClassVar

from mlint.visitors.base import BaseVisitor
from mlint.formatters.default import DefaultFormatter
from mlint.plugins.base import BaseCheckerPlugin, BaseReporterPlugin
from mlint.plugins.core.config import CorePluginConfig
from mlint.plugins.core.visitors import (
    BackwardMigrationVisitor,
    CreateIndexConcurrentlyVisitor,
    CreateIndexStatementTimeoutVisitor,
    CreateIndexTransactionalMigrationVisitor,
    DropIndexConcurrentlyVisitor,
    DropIndexTransactionalMigrationVisitor,
    GeneratedAlwaysVisitor,
    PrimaryKeyVisitor,
    RequiredColumnsVisitor,
    RequiredCommentsVisitor,
)


class CoreCheckerPlugin(BaseCheckerPlugin[CorePluginConfig]):
    """Checker-плагин с базовыми правилами валидации."""

    name: ClassVar[str] = "mlint-core-checker"
    mlint_extension: ClassVar[str] = "M"
    config_cls: type[CorePluginConfig] | None = CorePluginConfig
    
    # набор базовых правил
    visitors: list[type[BaseVisitor]] = [
        BackwardMigrationVisitor,
        CreateIndexConcurrentlyVisitor,
        DropIndexConcurrentlyVisitor,
        GeneratedAlwaysVisitor,
        PrimaryKeyVisitor,
        RequiredColumnsVisitor,
        RequiredCommentsVisitor,
        ...
    ]


class CoreReporterPlugin(BaseReporterPlugin):
    """
    Reporter-плагин по умолчанию.
    
    Выводит ошибки в plain-text формате.
    """

    name: ClassVar[str] = "default"

    formatter_cls = DefaultFormatter

def load_plugins(config: dict[str, Any], base_dir: Path) -> Plugins:
    """
    Загружает локальные и глобальные плагины.

    Основной функционал загрузки плагинов, включающий:
    1. Поиск плагинов (в директории с проектом и в site-packages).
    2. Загрузку плагинов из entry points.
    3. Группировку плагинов по типам (checker, reporter).
    4. Загрузку core-плагинов с базовыми проверками mlint.
    """
    # поиск локальных плагинов в файле конфигурации
    local_plugin_opts = _parse_local_plugin_options(config)
    
    plugins_meta = [
        *_find_importlib_plugins(),
        *_find_local_plugins(local_plugin_opts, base_dir),
    ]

    plugins_meta.sort(key=attrgetter("package"))

    checkers = []
    reporters = []

    for plugin_meta in plugins_meta:
        try:
	        # импорт плагина
            plugin_cls = plugin_meta.entry_point.load()
        except Exception as exc:
            raise PluginLoadingError(
                entry_point=plugin_meta.entry_point.value
            ) from exc
		
		# группировка плагинов по типам
        if plugin_meta.type == PluginType.EXTENSION:
            checkers.append(CheckerPlugin(cls=plugin_cls))
        else:
            reporters.append(
                ReporterPlugin(
                    name=plugin_meta.entry_point.name, 
                    cls=plugin_cls,
                )
            )

    return Plugins(checkers=checkers, reporters=reporters)

@dataclasses.dataclass  
class MLintConfig(BaseConfig):  
    """Класс конфигурации линтера."""  
	  
	should_create_baseline: bool = False
    baseline_path: Path = Path("mlint-baseline.txt")  
    ...  # остальные аргументы  
	
    @classmethod  
    def from_args(cls, args: dict[str, Any]) -> MLintConfig:  
        """Инициализирует конфигурацию данными CLI-аргументов."""  
        config = cls()  
        config = config.read_args(args)  
  
        if isinstance(config.baseline_path, str):  
            config.baseline_path = Path(config.baseline_path) 
            
        if config.should_create_baseline:  
		    config.format = "baseline"  # устанавливаем новый формат отчёта   
        
		...  # нормализация аргументов, приведение типов и т.д.
  
        return config  
	
    @classmethod  
    def add_options(cls, manager: OptionManager) -> None:  
        """Регистрирует CLI-аргументы для атрибутов конфигурации."""  
        manager.add(
            "--baseline",
            action="store_true",
            dest="should_create_baseline",
            help="Создание baseline-файла.",
        )
        manager.add(  
            "--baseline-path",  
            type=Path,  
            parse_from_config=True,  
            help="Путь к baseline-файлу.",  
        ) 
        ...  # остальные аргументы

from pglast import ast, visitors, enums

from mlint.core.models import Column
from mlint.plugins.core.violations import RequiredColumnsViolation
from mlint.visitors.base import BaseNodeVisitor, CorePluginConfig


class RequiredColumnsVisitor(BaseNodeVisitor[CorePluginConfig]):
    """Класс для проверки обязательных столбцов в таблице."""
    
    def visit_CreateStmt(
        self,
        ancestors: visitors.Ancestor,
        node: ast.CreateStmt,
    ) -> None:
        """Посещение узла CreateStmt - создание таблицы."""
        # получаем обязательные таблицы из конфигурации
        required_table_columns = self._config.required_table_columns
        required_table_column_names = {
	        column.name for column in required_table_columns
	    }

        table_columns = set()
        # получаем имя таблицы
        table_name = node.stmt.relation.relname

        for elem in node.stmt.tableElts:
            if isinstance(elem, ast.ColumnDef):
                # получаем имя столбца таблицы
                table_columns.add(elem.colname)

        missing_columns = required_table_column_names - table_columns

        if missing_columns:
            self.add_violation(
                RequiredColumnsViolation,
                table_name=table_name,
                missing_columns=missing_columns,
                # указываем, что ошибка может быть исправлена автоматически
                is_auto_fixable=True,
            )
            
            # если линтер запущен с опцией --auto-fix 
            if self._config.is_auto_fix_enabled:
                columns_to_add = [
                    column.name for column in required_table_columns
                    if column.name in missing_columns
                ]
                
                self._apply_auto_fix(node, columns_to_add)

    def _apply_auto_fix(
	    self, 
	    node: ast.CreateStmt, 
	    columns_to_add: list[Column]
	) -> None:
        """Исправление ошибки: добавление недостающих столбцов."""
        original_columns = node.tableElts
        
        new_columns = []
        
        for column in columns_to_add:
            new_column = ast.ColumnDef(
                colname=column.name,  # название столбца
                typeName=ast.TypeName(  # тип данных
                    names=({"@": "String", "sval": column.data_type},),
                ),
                # ограничения: CONSTR_NOTNULL, CONSTR_UNIQUE и т.д.
                constraints=column.constraints,
            )
            new_columns.append(new_column)
        
        # добавляем оригинальные и новые столбцы
        node.tableElts = (*original_columns, *new_columns)