Skip to content

HookBase

Bases: ABC

Base class for Python migration hooks. This class defines the interface for migration 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.

Source code in pum/hook.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class HookBase(abc.ABC):
    """Base class for Python migration hooks.
    This class defines the interface for migration 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.
    """

    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:
            connection: The database connection.
            sql: The SQL statement to execute or a path to a SQL file..
        """
        SqlContent(sql).execute(
            connection=self._connection, parameters=self._parameters, commit=False
        )

    execute.__isfinal__ = True

__init__ 

__init__() -> None

Initialize the HookBase class.

Source code in pum/hook.py 
24
25
26
def __init__(self) -> None:
    """Initialize the HookBase class."""
    self._parameters: dict | None = None

execute 

execute(sql: str | 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.

Parameters:

Name Type Description Default
connection

The database connection.

 required
sql str | SQL | Path

The SQL statement to execute or a path to a SQL file..

 required
Source code in pum/hook.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
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:
        connection: The database connection.
        sql: The SQL statement to execute or a path to a SQL file..
    """
    SqlContent(sql).execute(
        connection=self._connection, parameters=self._parameters, commit=False
    )

run_hook abstractmethod 

run_hook(connection: 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.

Source code in pum/hook.py 
39
40
41
42
43
44
45
46
47
48
49
@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.")