Skip to content

PumConfig

A class to hold configuration settings.

Source code in pum/pum_config.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
class PumConfig:
    """A class to hold configuration settings."""

    def __init__(self, base_path: str | Path, validate: bool = True, **kwargs: dict) -> None:
        """Initialize the configuration with key-value pairs.

        Args:
            base_path: The directory where the changelogs are located.
            validate: Whether to validate the changelogs and hooks.
            **kwargs: Key-value pairs representing configuration settings.

        Raises:
            PumConfigError: If the configuration is invalid.

        """
        # self.pg_restore_exe: str | None = kwargs.get("pg_restore_exe") or os.getenv(
        #     "PG_RESTORE_EXE"
        # )
        # self.pg_dump_exe: str | None = kwargs.get("pg_dump_exe") or os.getenv("PG_DUMP_EXE")

        if not isinstance(base_path, Path):
            base_path = Path(base_path)
        if not base_path.is_dir():
            raise PumConfigError(f"Directory `{base_path}` does not exist.")
        self._base_path = base_path

        try:
            self.config = ConfigModel(**kwargs)
        except ValidationError as e:
            logger.error("Config validation error: %s", e)
            raise PumConfigError(e) from e

        if validate:
            if self.config.pum.minimum_version and PUM_VERSION < self.config.pum.minimum_version:
                raise PumConfigError(
                    f"Minimum required version of pum is {self.config.pum.minimum_version}, but the current version is {PUM_VERSION}. Please upgrade pum."
                )
            try:
                self.validate()
            except (PumInvalidChangelog, PumHookError) as e:
                raise PumConfigError(
                    f"Configuration is invalid: {e}. You can disable the validation when constructing the config."
                ) from e

    @classmethod
    def from_yaml(cls, file_path: str | Path, *, validate: bool = True) -> "PumConfig":
        """Create a PumConfig instance from a YAML file.

        Args:
            file_path: The path to the YAML file.
            validate: Whether to validate the changelogs and hooks.

        Returns:
            PumConfig: An instance of the PumConfig class.

        Raises:
            FileNotFoundError: If the file does not exist.
            yaml.YAMLError: If there is an error parsing the YAML file.

        """
        with Path.open(file_path) as file:
            data = yaml.safe_load(file)

        if "base_path" in data:
            raise PumConfigError("base_path not allowed in configuration instead.")

        base_path = Path(file_path).parent
        return cls(base_path=base_path, validate=validate, **data)

    def parameter(self, name: str) -> ParameterDefinition:
        """Get a specific migration parameter by name.

        Args:
            name: The name of the parameter.

        Returns:
            ParameterDefintion: The migration parameter definition.

        Raises:
            PumConfigError: If the parameter name does not exist.

        """
        for parameter in self.config.parameters:
            if parameter.name == name:
                return ParameterDefinition(**parameter.model_dump())
        raise PumConfigError(f"Parameter '{name}' not found in configuration.") from KeyError

    def last_version(
        self, min_version: str | None = None, max_version: str | None = None
    ) -> str | None:
        """Return the last version of the changelogs.
        The changelogs are sorted by version.

        Args:
            min_version (str | None): The version to start from (inclusive).
            max_version (str | None): The version to end at (inclusive).

        Returns:
            str | None: The last version of the changelogs. If no changelogs are found, None is returned.

        """
        changelogs = self.changelogs(min_version, max_version)
        if not changelogs:
            return None
        if min_version:
            changelogs = [
                c for c in changelogs if c.version >= packaging.version.parse(min_version)
            ]
        if max_version:
            changelogs = [
                c for c in changelogs if c.version <= packaging.version.parse(max_version)
            ]
        if not changelogs:
            return None
        return changelogs[-1].version

    def changelogs(self, min_version: str | None = None, max_version: str | None = None) -> list:
        """Return a list of changelogs.
        The changelogs are sorted by version.

        Args:
            min_version (str | None): The version to start from (inclusive).
            max_version (str | None): The version to end at (inclusive).

        Returns:
            list: A list of changelogs. Each changelog is represented by a Changelog object.

        """
        path = self._base_path / self.config.changelogs_directory
        if not path.is_dir():
            raise PumException(f"Changelogs directory `{path}` does not exist.")
        if not path.iterdir():
            raise PumException(f"Changelogs directory `{path}` is empty.")

        changelogs = [Changelog(d) for d in path.iterdir() if d.is_dir()]

        if min_version:
            changelogs = [
                c for c in changelogs if c.version >= packaging.version.parse(min_version)
            ]
        if max_version:
            changelogs = [
                c for c in changelogs if c.version <= packaging.version.parse(max_version)
            ]

        changelogs.sort(key=lambda c: c.version)
        return changelogs

    def role_manager(self) -> RoleManager:
        """Return a RoleManager instance based on the roles defined in the configuration."""
        if not self.config.roles:
            logger.warning("No roles defined in the configuration. Returning an empty RoleManager.")
            return RoleManager()
        return RoleManager([role.model_dump() for role in self.config.roles])

    def pre_hook_handlers(self) -> list[HookHandler]:
        """Return the list of pre-migration hook handlers."""
        return (
            [
                HookHandler(base_path=self._base_path, **hook.model_dump())
                for hook in self.config.migration_hooks.pre
            ]
            if self.config.migration_hooks.pre
            else []
        )

    def post_hook_handlers(self) -> list[HookHandler]:
        """Return the list of post-migration hook handlers."""
        return (
            [
                HookHandler(base_path=self._base_path, **hook.model_dump())
                for hook in self.config.migration_hooks.post
            ]
            if self.config.migration_hooks.post
            else []
        )

    def validate(self) -> None:
        """Validate the chanbgelogs and hooks."""

        parameter_defaults = {}
        for parameter in self.config.parameters:
            parameter_defaults[parameter.name] = psycopg.sql.Literal(parameter.default)

        for changelog in self.changelogs():
            try:
                changelog.validate(parameters=parameter_defaults)
            except (PumInvalidChangelog, PumSqlError) as e:
                raise PumInvalidChangelog(f"Changelog `{changelog}` is invalid.") from e

        hook_handlers = []
        if self.config.migration_hooks.pre:
            hook_handlers.extend(self.pre_hook_handlers())
        if self.config.migration_hooks.post:
            hook_handlers.extend(self.post_hook_handlers())
        for hook_handler in hook_handlers:
            try:
                hook_handler.validate(parameter_defaults)
            except PumHookError as e:
                raise PumHookError(f"Hook `{hook_handler}` is invalid.") from e

__init__ 

__init__(base_path: str | Path, validate: bool = True, **kwargs: dict) -> None

Initialize the configuration with key-value pairs.

Parameters:

Name Type Description Default
base_path str | Path

The directory where the changelogs are located.

 required
validate bool

Whether to validate the changelogs and hooks.

 True
**kwargs dict

Key-value pairs representing configuration settings.

 {}

Raises:

Type Description
PumConfigError

If the configuration is invalid.
Source code in pum/pum_config.py 
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
def __init__(self, base_path: str | Path, validate: bool = True, **kwargs: dict) -> None:
    """Initialize the configuration with key-value pairs.

    Args:
        base_path: The directory where the changelogs are located.
        validate: Whether to validate the changelogs and hooks.
        **kwargs: Key-value pairs representing configuration settings.

    Raises:
        PumConfigError: If the configuration is invalid.

    """
    # self.pg_restore_exe: str | None = kwargs.get("pg_restore_exe") or os.getenv(
    #     "PG_RESTORE_EXE"
    # )
    # self.pg_dump_exe: str | None = kwargs.get("pg_dump_exe") or os.getenv("PG_DUMP_EXE")

    if not isinstance(base_path, Path):
        base_path = Path(base_path)
    if not base_path.is_dir():
        raise PumConfigError(f"Directory `{base_path}` does not exist.")
    self._base_path = base_path

    try:
        self.config = ConfigModel(**kwargs)
    except ValidationError as e:
        logger.error("Config validation error: %s", e)
        raise PumConfigError(e) from e

    if validate:
        if self.config.pum.minimum_version and PUM_VERSION < self.config.pum.minimum_version:
            raise PumConfigError(
                f"Minimum required version of pum is {self.config.pum.minimum_version}, but the current version is {PUM_VERSION}. Please upgrade pum."
            )
        try:
            self.validate()
        except (PumInvalidChangelog, PumHookError) as e:
            raise PumConfigError(
                f"Configuration is invalid: {e}. You can disable the validation when constructing the config."
            ) from e

changelogs 

changelogs(min_version: str | None = None, max_version: str | None = None) -> list

Return a list of changelogs. The changelogs are sorted by version.

Parameters:

Name Type Description Default
min_version str | None

The version to start from (inclusive).

 None
max_version str | None

The version to end at (inclusive).

 None

Returns:

Name Type Description
list list

A list of changelogs. Each changelog is represented by a Changelog object.
Source code in pum/pum_config.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def changelogs(self, min_version: str | None = None, max_version: str | None = None) -> list:
    """Return a list of changelogs.
    The changelogs are sorted by version.

    Args:
        min_version (str | None): The version to start from (inclusive).
        max_version (str | None): The version to end at (inclusive).

    Returns:
        list: A list of changelogs. Each changelog is represented by a Changelog object.

    """
    path = self._base_path / self.config.changelogs_directory
    if not path.is_dir():
        raise PumException(f"Changelogs directory `{path}` does not exist.")
    if not path.iterdir():
        raise PumException(f"Changelogs directory `{path}` is empty.")

    changelogs = [Changelog(d) for d in path.iterdir() if d.is_dir()]

    if min_version:
        changelogs = [
            c for c in changelogs if c.version >= packaging.version.parse(min_version)
        ]
    if max_version:
        changelogs = [
            c for c in changelogs if c.version <= packaging.version.parse(max_version)
        ]

    changelogs.sort(key=lambda c: c.version)
    return changelogs

from_yaml classmethod 

from_yaml(file_path: str | Path, *, validate: bool = True) -> PumConfig

Create a PumConfig instance from a YAML file.

Parameters:

Name Type Description Default
file_path str | Path

The path to the YAML file.

 required
validate bool

Whether to validate the changelogs and hooks.

 True

Returns:

Name Type Description
PumConfig PumConfig

An instance of the PumConfig class.

Raises:

Type Description
FileNotFoundError

If the file does not exist.
YAMLError

If there is an error parsing the YAML file.
Source code in pum/pum_config.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def from_yaml(cls, file_path: str | Path, *, validate: bool = True) -> "PumConfig":
    """Create a PumConfig instance from a YAML file.

    Args:
        file_path: The path to the YAML file.
        validate: Whether to validate the changelogs and hooks.

    Returns:
        PumConfig: An instance of the PumConfig class.

    Raises:
        FileNotFoundError: If the file does not exist.
        yaml.YAMLError: If there is an error parsing the YAML file.

    """
    with Path.open(file_path) as file:
        data = yaml.safe_load(file)

    if "base_path" in data:
        raise PumConfigError("base_path not allowed in configuration instead.")

    base_path = Path(file_path).parent
    return cls(base_path=base_path, validate=validate, **data)

last_version 

last_version(min_version: str | None = None, max_version: str | None = None) -> str | None

Return the last version of the changelogs. The changelogs are sorted by version.

Parameters:

Name Type Description Default
min_version str | None

The version to start from (inclusive).

 None
max_version str | None

The version to end at (inclusive).

 None

Returns:

Type Description
str | None

str | None: The last version of the changelogs. If no changelogs are found, None is returned.
Source code in pum/pum_config.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def last_version(
    self, min_version: str | None = None, max_version: str | None = None
) -> str | None:
    """Return the last version of the changelogs.
    The changelogs are sorted by version.

    Args:
        min_version (str | None): The version to start from (inclusive).
        max_version (str | None): The version to end at (inclusive).

    Returns:
        str | None: The last version of the changelogs. If no changelogs are found, None is returned.

    """
    changelogs = self.changelogs(min_version, max_version)
    if not changelogs:
        return None
    if min_version:
        changelogs = [
            c for c in changelogs if c.version >= packaging.version.parse(min_version)
        ]
    if max_version:
        changelogs = [
            c for c in changelogs if c.version <= packaging.version.parse(max_version)
        ]
    if not changelogs:
        return None
    return changelogs[-1].version

parameter 

parameter(name: str) -> ParameterDefinition

Get a specific migration parameter by name.

Parameters:

Name Type Description Default
name str

The name of the parameter.

 required

Returns:

Name Type Description
ParameterDefintion ParameterDefinition

The migration parameter definition.

Raises:

Type Description
PumConfigError

If the parameter name does not exist.
Source code in pum/pum_config.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def parameter(self, name: str) -> ParameterDefinition:
    """Get a specific migration parameter by name.

    Args:
        name: The name of the parameter.

    Returns:
        ParameterDefintion: The migration parameter definition.

    Raises:
        PumConfigError: If the parameter name does not exist.

    """
    for parameter in self.config.parameters:
        if parameter.name == name:
            return ParameterDefinition(**parameter.model_dump())
    raise PumConfigError(f"Parameter '{name}' not found in configuration.") from KeyError

post_hook_handlers 

post_hook_handlers() -> list[HookHandler]

Return the list of post-migration hook handlers.

Source code in pum/pum_config.py 
193
194
195
196
197
198
199
200
201
202
def post_hook_handlers(self) -> list[HookHandler]:
    """Return the list of post-migration hook handlers."""
    return (
        [
            HookHandler(base_path=self._base_path, **hook.model_dump())
            for hook in self.config.migration_hooks.post
        ]
        if self.config.migration_hooks.post
        else []
    )

pre_hook_handlers 

pre_hook_handlers() -> list[HookHandler]

Return the list of pre-migration hook handlers.

Source code in pum/pum_config.py 
182
183
184
185
186
187
188
189
190
191
def pre_hook_handlers(self) -> list[HookHandler]:
    """Return the list of pre-migration hook handlers."""
    return (
        [
            HookHandler(base_path=self._base_path, **hook.model_dump())
            for hook in self.config.migration_hooks.pre
        ]
        if self.config.migration_hooks.pre
        else []
    )

role_manager 

role_manager() -> RoleManager

Return a RoleManager instance based on the roles defined in the configuration.

Source code in pum/pum_config.py 
175
176
177
178
179
180
def role_manager(self) -> RoleManager:
    """Return a RoleManager instance based on the roles defined in the configuration."""
    if not self.config.roles:
        logger.warning("No roles defined in the configuration. Returning an empty RoleManager.")
        return RoleManager()
    return RoleManager([role.model_dump() for role in self.config.roles])

validate 

validate() -> None

Validate the chanbgelogs and hooks.

Source code in pum/pum_config.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def validate(self) -> None:
    """Validate the chanbgelogs and hooks."""

    parameter_defaults = {}
    for parameter in self.config.parameters:
        parameter_defaults[parameter.name] = psycopg.sql.Literal(parameter.default)

    for changelog in self.changelogs():
        try:
            changelog.validate(parameters=parameter_defaults)
        except (PumInvalidChangelog, PumSqlError) as e:
            raise PumInvalidChangelog(f"Changelog `{changelog}` is invalid.") from e

    hook_handlers = []
    if self.config.migration_hooks.pre:
        hook_handlers.extend(self.pre_hook_handlers())
    if self.config.migration_hooks.post:
        hook_handlers.extend(self.post_hook_handlers())
    for hook_handler in hook_handlers:
        try:
            hook_handler.validate(parameter_defaults)
        except PumHookError as e:
            raise PumHookError(f"Hook `{hook_handler}` is invalid.") from e