Skip to content

PumConfig

A class to hold configuration settings.

Source code in pum/pum_config.py 
 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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
class PumConfig:
    """A class to hold configuration settings."""

    def __init__(
        self,
        base_path: str | Path,
        *,
        validate: bool = True,
        install_dependencies: bool = False,
        **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 and resolve dependencies. Defaults to True.
            install_dependencies: Whether to temporarily install dependencies.
            **kwargs: Key-value pairs representing configuration settings.

        Raises:
            PumConfigError: If the configuration is invalid.

        """

        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

        self.dependency_path = None

        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(install_dependencies=install_dependencies)
            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,
        install_dependencies: bool = False,
    ) -> "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.
            install_dependencies: Wheter to temporarily install dependencies.

        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,
            install_dependencies=install_dependencies,
            **data,
        )

    @property
    def base_path(self) -> Path:
        """Return the base path used for configuration and changelogs."""
        return self._base_path

    def parameters(self) -> list[ParameterDefinition]:
        """Return a list of migration parameters.

        Returns:
            list[ParameterDefinition]: A list of migration parameter definitions.

        """
        return [
            ParameterDefinition(**parameter.model_dump()) for parameter in self.config.parameters
        ]

    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 demo_data(self) -> dict[str, str]:
        """Return a dictionary of demo data files defined in the configuration."""
        return {dm.name: dm.file for dm in self.config.demo_data}

    def __del__(self):
        # Cleanup temporary directories and sys.path modifications
        if self.dependency_path:
            # Remove from sys.path if present
            sys.path = [p for p in sys.path if p != str(self.dependency_path)]
            # Remove the directory if it exists and is a TemporaryDirectory
            if hasattr(self, "_temp_dir") and self._temp_dir:
                self.dependency_tmp.cleanup()

    def validate(self, install_dependencies: bool = False) -> None:
        """Validate the changelogs and hooks.

        Args:
            install_dependencies (bool): Whether to temporarily install dependencies.
        """

        if install_dependencies and self.config.dependencies:
            self.dependency_tmp = tempfile.TemporaryDirectory()
            self.dependency_path = Path(self.dependency_tmp.name)
            sys.path.insert(0, str(self.dependency_path))

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

        for dependency in self.config.dependencies:
            DependencyHandler(**dependency.model_dump()).resolve(
                install_dependencies=install_dependencies, install_path=self.dependency_path
            )

        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

base_path property 

base_path: Path

Return the base path used for configuration and changelogs.

__init__ 

__init__(base_path: str | Path, *, validate: bool = True, install_dependencies: bool = False, **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 and resolve dependencies. Defaults to True.

 True
install_dependencies bool

Whether to temporarily install dependencies.

 False
**kwargs dict

Key-value pairs representing configuration settings.

 {}

Raises:

Type Description
PumConfigError

If the configuration is invalid.
Source code in pum/pum_config.py 
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
def __init__(
    self,
    base_path: str | Path,
    *,
    validate: bool = True,
    install_dependencies: bool = False,
    **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 and resolve dependencies. Defaults to True.
        install_dependencies: Whether to temporarily install dependencies.
        **kwargs: Key-value pairs representing configuration settings.

    Raises:
        PumConfigError: If the configuration is invalid.

    """

    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

    self.dependency_path = None

    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(install_dependencies=install_dependencies)
        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 
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
227
228
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

demo_data 

demo_data() -> dict[str, str]

Return a dictionary of demo data files defined in the configuration.

Source code in pum/pum_config.py 
259
260
261
def demo_data(self) -> dict[str, str]:
    """Return a dictionary of demo data files defined in the configuration."""
    return {dm.name: dm.file for dm in self.config.demo_data}

from_yaml classmethod 

from_yaml(file_path: str | Path, *, validate: bool = True, install_dependencies: bool = False) -> 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
install_dependencies bool

Wheter to temporarily install dependencies.

 False

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 
 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
@classmethod
def from_yaml(
    cls,
    file_path: str | Path,
    *,
    validate: bool = True,
    install_dependencies: bool = False,
) -> "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.
        install_dependencies: Wheter to temporarily install dependencies.

    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,
        install_dependencies=install_dependencies,
        **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 
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
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 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
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

parameters 

parameters() -> list[ParameterDefinition]

Return a list of migration parameters.

Returns:

Type Description
list[ParameterDefinition]

list[ParameterDefinition]: A list of migration parameter definitions.
Source code in pum/pum_config.py 
140
141
142
143
144
145
146
147
148
149
def parameters(self) -> list[ParameterDefinition]:
    """Return a list of migration parameters.

    Returns:
        list[ParameterDefinition]: A list of migration parameter definitions.

    """
    return [
        ParameterDefinition(**parameter.model_dump()) for parameter in self.config.parameters
    ]

post_hook_handlers 

post_hook_handlers() -> list[HookHandler]

Return the list of post-migration hook handlers.

Source code in pum/pum_config.py 
248
249
250
251
252
253
254
255
256
257
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 
237
238
239
240
241
242
243
244
245
246
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 
230
231
232
233
234
235
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(install_dependencies: bool = False) -> None

Validate the changelogs and hooks.

Parameters:

Name Type Description Default
install_dependencies bool

Whether to temporarily install dependencies.

 False
Source code in pum/pum_config.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def validate(self, install_dependencies: bool = False) -> None:
    """Validate the changelogs and hooks.

    Args:
        install_dependencies (bool): Whether to temporarily install dependencies.
    """

    if install_dependencies and self.config.dependencies:
        self.dependency_tmp = tempfile.TemporaryDirectory()
        self.dependency_path = Path(self.dependency_tmp.name)
        sys.path.insert(0, str(self.dependency_path))

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

    for dependency in self.config.dependencies:
        DependencyHandler(**dependency.model_dump()).resolve(
            install_dependencies=install_dependencies, install_path=self.dependency_path
        )

    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