Skip to content

Changelog

This class represent a changelog directory. The directory name is the version of the changelog.

Source code in pum/changelog.py 
 12
 13
 14
 15
 16
 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
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class Changelog:
    """This class represent a changelog directory.
    The directory name is the version of the changelog.
    """

    def __init__(self, dir):
        """Args:
        dir (str): The directory where the changelog is located.

        """
        self.dir = dir
        self.version = parse_version(basename(dir))

    def __repr__(self):
        return f"<dir: {self.dir} (v: {self.version})>"

    def files(self) -> list[Path]:
        """Get the ordered list of SQL files in the changelog directory.
        This is not recursive, it only returns the files in the given changelog directory.

        Returns:
            list[Path]: A list of paths to the changelog files.

        """
        files = [
            self.dir / f
            for f in listdir(self.dir)
            if (self.dir / f).is_file() and f.endswith(".sql")
        ]
        files.sort()
        return files

    def validate(self, parameters: dict | None = None) -> bool:
        """Validate the changelog directory.
        This is done by checking if the directory exists and if it contains at least one SQL file.

        Args:
            parameters: The parameters to pass to the SQL files.

        Raises:
            PumInvalidChangelog: If the changelog directory does not exist or does not contain any SQL files.

        """
        if not self.dir.is_dir():
            raise PumInvalidChangelog(f"Changelog directory `{self.dir}` does not exist.")
        files = self.files()
        if not files:
            raise PumInvalidChangelog(
                f"Changelog directory `{self.dir}` does not contain any SQL files."
            )
        for file in files:
            if not file.is_file():
                raise PumInvalidChangelog(f"Changelog file `{file}` does not exist.")
            if not file.suffix == ".sql":
                raise PumInvalidChangelog(f"Changelog file `{file}` is not a SQL file.")
            try:
                SqlContent(file).validate(parameters=parameters)
            except PumSqlError as e:
                raise PumInvalidChangelog(
                    f"Changelog file `{file}` is not a valid SQL file."
                ) from e

        if not self.version:
            raise PumInvalidChangelog(
                f"Changelog directory `{self.dir}` does not have a valid version."
            )
        return True

    def apply(
        self,
        connection: psycopg.Connection,
        parameters: dict | None = None,
        commit: bool = True,
    ) -> list[Path]:
        """Apply a changelog
        This will execute all the files in the changelog directory.
        The changelog directory is the one that contains the delta files.

        Args:
            connection: Connection
                The connection to the database
            parameters: dict
                The parameters to pass to the SQL files
            commit: bool
                If true, the transaction is committed. The default is true.

        Returns:
            list[Path]
                The list of changelogs that were executed

        """
        files = self.files()
        for file in files:
            try:
                SqlContent(file).execute(
                    connection=connection, commit=commit, parameters=parameters
                )
            except PumSqlError as e:
                raise PumSqlError(f"Error applying changelog {file}: {e}") from e
        return files

__init__ 

__init__(dir)

Args: dir (str): The directory where the changelog is located.

Source code in pum/changelog.py 
17
18
19
20
21
22
23
def __init__(self, dir):
    """Args:
    dir (str): The directory where the changelog is located.

    """
    self.dir = dir
    self.version = parse_version(basename(dir))

apply 

apply(connection: Connection, parameters: dict | None = None, commit: bool = True) -> list[Path]

Apply a changelog This will execute all the files in the changelog directory. The changelog directory is the one that contains the delta files.

Parameters:

Name Type Description Default
connection Connection

Connection The connection to the database

 required
parameters dict | None

dict The parameters to pass to the SQL files

 None
commit bool

bool If true, the transaction is committed. The default is true.

 True

Returns:

Type Description
list[Path]

list[Path] The list of changelogs that were executed
Source code in pum/changelog.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def apply(
    self,
    connection: psycopg.Connection,
    parameters: dict | None = None,
    commit: bool = True,
) -> list[Path]:
    """Apply a changelog
    This will execute all the files in the changelog directory.
    The changelog directory is the one that contains the delta files.

    Args:
        connection: Connection
            The connection to the database
        parameters: dict
            The parameters to pass to the SQL files
        commit: bool
            If true, the transaction is committed. The default is true.

    Returns:
        list[Path]
            The list of changelogs that were executed

    """
    files = self.files()
    for file in files:
        try:
            SqlContent(file).execute(
                connection=connection, commit=commit, parameters=parameters
            )
        except PumSqlError as e:
            raise PumSqlError(f"Error applying changelog {file}: {e}") from e
    return files

files 

files() -> list[Path]

Get the ordered list of SQL files in the changelog directory. This is not recursive, it only returns the files in the given changelog directory.

Returns:

Type Description
list[Path]

list[Path]: A list of paths to the changelog files.
Source code in pum/changelog.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def files(self) -> list[Path]:
    """Get the ordered list of SQL files in the changelog directory.
    This is not recursive, it only returns the files in the given changelog directory.

    Returns:
        list[Path]: A list of paths to the changelog files.

    """
    files = [
        self.dir / f
        for f in listdir(self.dir)
        if (self.dir / f).is_file() and f.endswith(".sql")
    ]
    files.sort()
    return files

validate 

validate(parameters: dict | None = None) -> bool

Validate the changelog directory. This is done by checking if the directory exists and if it contains at least one SQL file.

Parameters:

Name Type Description Default
parameters dict | None

The parameters to pass to the SQL files.

 None

Raises:

Type Description
PumInvalidChangelog

If the changelog directory does not exist or does not contain any SQL files.
Source code in pum/changelog.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def validate(self, parameters: dict | None = None) -> bool:
    """Validate the changelog directory.
    This is done by checking if the directory exists and if it contains at least one SQL file.

    Args:
        parameters: The parameters to pass to the SQL files.

    Raises:
        PumInvalidChangelog: If the changelog directory does not exist or does not contain any SQL files.

    """
    if not self.dir.is_dir():
        raise PumInvalidChangelog(f"Changelog directory `{self.dir}` does not exist.")
    files = self.files()
    if not files:
        raise PumInvalidChangelog(
            f"Changelog directory `{self.dir}` does not contain any SQL files."
        )
    for file in files:
        if not file.is_file():
            raise PumInvalidChangelog(f"Changelog file `{file}` does not exist.")
        if not file.suffix == ".sql":
            raise PumInvalidChangelog(f"Changelog file `{file}` is not a SQL file.")
        try:
            SqlContent(file).validate(parameters=parameters)
        except PumSqlError as e:
            raise PumInvalidChangelog(
                f"Changelog file `{file}` is not a valid SQL file."
            ) from e

    if not self.version:
        raise PumInvalidChangelog(
            f"Changelog directory `{self.dir}` does not have a valid version."
        )
    return True