Skip to content

PumConfig

A class to hold configuration settings.

Source code in pum/config.py 
  7
  8
  9
 10
 11
 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
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
class PumConfig:
    """
    A class to hold configuration settings.
    """

    def __init__(self, **kwargs):
        """
        Initialize the configuration with key-value pairs.

        Args:
            **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 "dir" in kwargs:
            raise PumConfigError(
                "dir not allowed in configuration, use PumConfig.from_yaml() instead."
            )

        self.pum_migrations_table: str = (
            f"{(kwargs.get('pum_migrations_schema') or 'public')}.pum_migrations"
        )
        self.changelogs_directory: str = kwargs.get("changelogs_directory") or "changelogs"

        self.parameter_definitions = dict()
        for p in kwargs.get("parameters") or ():
            if isinstance(p, dict):
                name = p.get("name")
                type_ = p.get("type")
                default = p.get("default")
                description = p.get("description")
                self.parameter_definitions[name] = MigrationParameterDefinition(
                    name=name,
                    type_=type_,
                    default=default,
                    description=description,
                )
            elif isinstance(p, MigrationParameterDefinition):
                self.parameter_definitions[p.name] = p
            else:
                raise PumConfigError(
                    "parameters must be a list of dictionaries or MigrationParameterDefintion instances"
                )

        # Migration hooks
        self.pre_hooks = []
        self.post_hooks = []
        migration_hooks = kwargs.get("migration_hooks") or dict()
        pre_hook_defintions = migration_hooks.get("pre", [])
        post_hook_defintions = migration_hooks.get("post", [])
        for hook_type, hook_definitions in (
            (MigrationHookType.PRE, pre_hook_defintions),
            (MigrationHookType.POST, post_hook_defintions),
        ):
            for hook_definition in hook_definitions:
                hook = None
                if not isinstance(hook_definition, dict):
                    raise PumConfigError("hook must be a list of key-value pairs")
                if isinstance(hook_definition.get("file"), str):
                    hook = MigrationHook(type=hook_type, file=hook_definition.get("file"))
                else:
                    raise PumConfigError("invalid hook configuration")
                assert isinstance(hook, MigrationHook)
                if hook_type == MigrationHookType.PRE:
                    self.pre_hooks.append(hook)
                elif hook_type == MigrationHookType.POST:
                    self.post_hooks.append(hook)
                else:
                    raise PumConfigError(f"Invalid hook type: {hook_type}")

    # def get(self, key, default=None) -> any:
    #     """
    #     Get a configuration value by key, with an optional default.
    #     This method allows dynamic retrieval of attributes from the PumConfig instance.
    #     Args:
    #         key (str): The name of the attribute to retrieve.
    #         default: The default value to return if the attribute does not exist.
    #     Returns:
    #         any: The value of the attribute, or the default value if the attribute does not exist.
    #     """
    #     return getattr(self, key, default)

    # def set(self, key, value):
    #     """
    #     Set a configuration value by key.
    #     This method allows dynamic setting of attributes on the PumConfig instance.

    #     Args:
    #         key (str): The name of the attribute to set.
    #         value: The value to assign to the attribute.
    #     Raises:
    #         AttributeError: If the attribute does not exist.
    #     """
    #     setattr(self, key, value)

    def parameters(self) -> dict[str, MigrationParameterDefinition]:
        """
        Get all migration parameters as a dictionary.

        Returns:
            dict[str, MigrationParameterDefintion]: A dictionary of migration parameters.
        The keys are parameter names, and the values are MigrationParameterDefintion instances.
        """
        return self.parameter_definitions

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

        Returns:
            MigrationParameterDefintion: The migration parameter definition.
        Raises:
            PumConfigError: If the parameter name does not exist.
        """
        try:
            return self.parameter_definitions[name]
        except KeyError:
            raise PumConfigError(f"Parameter '{name}' not found in configuration.")

    @classmethod
    def from_yaml(cls, file_path):
        """
        Create a PumConfig instance from a YAML file.
        Args:
            file_path (str): The path to the YAML file.
        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 open(file_path) as file:
            data = yaml.safe_load(file)
        return cls(**data)

__init__ 

__init__(**kwargs)

Initialize the configuration with key-value pairs.

Parameters:

Name Type Description Default
**kwargs

Key-value pairs representing configuration settings.

 {}

Raises:

Type Description
PumConfigError

If the configuration is invalid.
Source code in pum/config.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
def __init__(self, **kwargs):
    """
    Initialize the configuration with key-value pairs.

    Args:
        **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 "dir" in kwargs:
        raise PumConfigError(
            "dir not allowed in configuration, use PumConfig.from_yaml() instead."
        )

    self.pum_migrations_table: str = (
        f"{(kwargs.get('pum_migrations_schema') or 'public')}.pum_migrations"
    )
    self.changelogs_directory: str = kwargs.get("changelogs_directory") or "changelogs"

    self.parameter_definitions = dict()
    for p in kwargs.get("parameters") or ():
        if isinstance(p, dict):
            name = p.get("name")
            type_ = p.get("type")
            default = p.get("default")
            description = p.get("description")
            self.parameter_definitions[name] = MigrationParameterDefinition(
                name=name,
                type_=type_,
                default=default,
                description=description,
            )
        elif isinstance(p, MigrationParameterDefinition):
            self.parameter_definitions[p.name] = p
        else:
            raise PumConfigError(
                "parameters must be a list of dictionaries or MigrationParameterDefintion instances"
            )

    # Migration hooks
    self.pre_hooks = []
    self.post_hooks = []
    migration_hooks = kwargs.get("migration_hooks") or dict()
    pre_hook_defintions = migration_hooks.get("pre", [])
    post_hook_defintions = migration_hooks.get("post", [])
    for hook_type, hook_definitions in (
        (MigrationHookType.PRE, pre_hook_defintions),
        (MigrationHookType.POST, post_hook_defintions),
    ):
        for hook_definition in hook_definitions:
            hook = None
            if not isinstance(hook_definition, dict):
                raise PumConfigError("hook must be a list of key-value pairs")
            if isinstance(hook_definition.get("file"), str):
                hook = MigrationHook(type=hook_type, file=hook_definition.get("file"))
            else:
                raise PumConfigError("invalid hook configuration")
            assert isinstance(hook, MigrationHook)
            if hook_type == MigrationHookType.PRE:
                self.pre_hooks.append(hook)
            elif hook_type == MigrationHookType.POST:
                self.post_hooks.append(hook)
            else:
                raise PumConfigError(f"Invalid hook type: {hook_type}")

from_yaml classmethod 

from_yaml(file_path)

Create a PumConfig instance from a YAML file. Args: file_path (str): The path to the YAML file. 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.

Source code in pum/config.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@classmethod
def from_yaml(cls, file_path):
    """
    Create a PumConfig instance from a YAML file.
    Args:
        file_path (str): The path to the YAML file.
    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 open(file_path) as file:
        data = yaml.safe_load(file)
    return cls(**data)

parameter 

parameter(name) -> MigrationParameterDefinition

Get a specific migration parameter by name.

Returns:

Name Type Description
MigrationParameterDefintion MigrationParameterDefinition

The migration parameter definition.

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

Source code in pum/config.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
def parameter(self, name) -> MigrationParameterDefinition:
    """
    Get a specific migration parameter by name.

    Returns:
        MigrationParameterDefintion: The migration parameter definition.
    Raises:
        PumConfigError: If the parameter name does not exist.
    """
    try:
        return self.parameter_definitions[name]
    except KeyError:
        raise PumConfigError(f"Parameter '{name}' not found in configuration.")

parameters 

parameters() -> dict[str, MigrationParameterDefinition]

Get all migration parameters as a dictionary.

Returns:

Type Description
dict[str, MigrationParameterDefinition]

dict[str, MigrationParameterDefintion]: A dictionary of migration parameters.

The keys are parameter names, and the values are MigrationParameterDefintion instances.

Source code in pum/config.py 
108
109
110
111
112
113
114
115
116
def parameters(self) -> dict[str, MigrationParameterDefinition]:
    """
    Get all migration parameters as a dictionary.

    Returns:
        dict[str, MigrationParameterDefintion]: A dictionary of migration parameters.
    The keys are parameter names, and the values are MigrationParameterDefintion instances.
    """
    return self.parameter_definitions