Hook base

class HookBase(abc.ABC):
    """Base class for Python application hooks.
    This class defines the interface for application hooks that can be implemented in Python.
    It requires the implementation of the `run_hook` method, which will be called during the migration process.
    It can call the execute method to run SQL statements with the provided connection and parameters.

    Version Added:
        1.0.0
    """

    def __init__(self) -> None:
        """Initialize the HookBase class."""
        self._parameters: dict | None = None

    def _prepare(self, connection: psycopg.Connection, parameters: dict | None = None) -> None:
        """Prepare the hook with the given connection and parameters.
        Args:
            connection: The database connection.
            parameters: Parameters to bind to the SQL statement. Defaults to None.
        Note:
            Parameters are stored as a deep copy, any modification will not be used when calling execute.
        """
        self._connection = connection
        self._parameters = copy.deepcopy(parameters)

    @abc.abstractmethod
    def run_hook(self, connection: psycopg.Connection, parameters: dict | None = None) -> None:
        """Run the migration hook.
        Args:
            connection: The database connection.
            parameters: Parameters to bind to the SQL statement. Defaults to None.

        Note:
            Parameters are given as a deep copy, any modification will not be used when calling execute.
        """
        raise NotImplementedError("The run_hook method must be implemented in the subclass.")

    def execute(
        self,
        sql: str | psycopg.sql.SQL | Path,
    ) -> None:
        """Execute the migration hook with the provided SQL and parameters for the migration.
        This is not committing the transaction and will be handled afterwards.

        Args:
            sql: The SQL statement to execute or a path to a SQL file.
        """
        parameters_literals = SqlContent.prepare_parameters(self._parameters)
        SqlContent(sql).execute(
            connection=self._connection, parameters=parameters_literals, commit=False
        )

    execute.__isfinal__ = True