MigrationHook

Base class for migration hooks.

Source code in pum/migration_hooks.py

class MigrationHook:
    """
    Base class for migration hooks.
    """

    def __init__(self, type: str | MigrationHookType, file: str | Path | None = None):
        """
        Initialize a MigrationHook instance.

        Args:
            type (str): The type of the hook (e.g., "pre", "post").
            file (str): The file path of the hook.
        """
        self.type = type if isinstance(type, MigrationHookType) else MigrationHookType(type)
        self.file = file

    def __repr__(self):
        return f"<{self.type.value} hook: {self.file}>"

    def __eq__(self, other):
        if not isinstance(other, MigrationHook):
            return NotImplemented
        return self.type == other.type and self.file == other.file

    def execute_sql(
        self,
        conn: Connection,
        dir: str | Path = ".",
        commit: bool = False,
        parameters: dict | None = None,
    ):
        """
        Execute the SQL file associated with the migration hook.

        Args:
            conn: The database connection.
            commit: Whether to commit the transaction after executing the SQL.
            dir: The root directory of the project.
            parameters (dict, optional): Parameters to bind to the SQL statement. Defaults to ().
        """

        logger.info(
            f"Executing {self.type.value} hook from file: {self.file} with parameters: {parameters}",
        )

        if self.file is None:
            raise ValueError("No file specified for the migration hook.")

        path = Path(dir) / self.file
        execute_sql(conn=conn, sql=path, commit=False, parameters=parameters)
        if commit:
            conn.commit()

__init__

__init__(type: str | MigrationHookType, file: str | Path | None = None)

Initialize a MigrationHook instance.

Parameters:

Name	Type	Description	Default
type	str	The type of the hook (e.g., "pre", "post").	required
file	str	The file path of the hook.	None

Source code in pum/migration_hooks.py

def __init__(self, type: str | MigrationHookType, file: str | Path | None = None):
    """
    Initialize a MigrationHook instance.

    Args:
        type (str): The type of the hook (e.g., "pre", "post").
        file (str): The file path of the hook.
    """
    self.type = type if isinstance(type, MigrationHookType) else MigrationHookType(type)
    self.file = file

execute_sql

execute_sql(conn: Connection, dir: str | Path = '.', commit: bool = False, parameters: dict | None = None)

Execute the SQL file associated with the migration hook.

Parameters:

Name	Type	Description	Default
conn	Connection	The database connection.	required
commit	bool	Whether to commit the transaction after executing the SQL.	False
dir	str | Path	The root directory of the project.	'.'
parameters	dict	Parameters to bind to the SQL statement. Defaults to ().	None

Source code in pum/migration_hooks.py

def execute_sql(
    self,
    conn: Connection,
    dir: str | Path = ".",
    commit: bool = False,
    parameters: dict | None = None,
):
    """
    Execute the SQL file associated with the migration hook.

    Args:
        conn: The database connection.
        commit: Whether to commit the transaction after executing the SQL.
        dir: The root directory of the project.
        parameters (dict, optional): Parameters to bind to the SQL statement. Defaults to ().
    """

    logger.info(
        f"Executing {self.type.value} hook from file: {self.file} with parameters: {parameters}",
    )

    if self.file is None:
        raise ValueError("No file specified for the migration hook.")

    path = Path(dir) / self.file
    execute_sql(conn=conn, sql=path, commit=False, parameters=parameters)
    if commit:
        conn.commit()