Skip to content

Upgrader

Class to handle the upgrade of a module. This class is used to install a new instance or to upgrade an existing instance of a module. It stores the info about the upgrade in a table on the database.

Source code in pum/upgrader.py 
 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class Upgrader:
    """Class to handle the upgrade of a module.
    This class is used to install a new instance or to upgrade an existing instance of a module.
    It stores the info about the upgrade in a table on the database.
    """

    def __init__(
        self,
        config: PumConfig,
        max_version: packaging.version.Version | str | None = None,
    ) -> None:
        """Initialize the Upgrader class.
        This class is used to install a new instance or to upgrade an existing instance of a module.
        Stores the info about the upgrade in a table on the database.
        The table is created in the schema defined in the config file if it does not exist.

        Args:
            connection:
                The database connection to use for the upgrade.
            config:
                The configuration object
            max_version:
                Maximum (including) version to run the deltas up to.

        """
        self.config = config
        self.max_version = packaging.parse(max_version) if max_version else None
        self.schema_migrations = SchemaMigrations(self.config)

    def install(
        self,
        connection: psycopg.Connection = None,
        *,
        parameters: dict | None = None,
        max_version: str | packaging.version.Version | None = None,
        roles: bool = False,
        grant: bool = False,
        beta_testing: bool = False,
        commit: bool = False,
    ) -> None:
        """Installs the given module
        This will create the schema_migrations table if it does not exist.
        The changelogs are applied in the order they are found in the directory.
        It will also set the baseline version to the current version of the module.

        Args:
            connection:
                The database connection to use for the upgrade.
            parameters:
                The parameters to pass for the migration.
            max_version:
                The maximum version to apply. If None, all versions are applied.
            roles:
                If True, roles will be created.
            grant:
                If True, permissions will be granted to the roles.
            beta_testing:
                If True, the module is installed in beta testing mode.
                This means that the module will not be able to receive any future updates.
                We strongly discourage using this for production.
            commit:
                If True, the changes will be committed to the database.
        """
        if self.schema_migrations.exists(connection):
            msg = (
                f"Schema migrations table {self.config.config.pum.migration_table_schema}.pum_migrations already exists. "
                "This means that the module is already installed or the database is not empty. "
                "Use upgrade() to upgrade the db or start with a clean db."
            )
            raise PumException(msg)
        self.schema_migrations.create(connection, commit=False)

        if roles or grant:
            self.config.role_manager().create_roles(
                connection=connection, grant=False, commit=False
            )

        for pre_hook in self.config.pre_hook_handlers():
            pre_hook.execute(connection=connection, commit=False, parameters=parameters)

        parameters_literals = SqlContent.prepare_parameters(parameters)
        last_changelog = None
        for changelog in self.config.changelogs(max_version=max_version):
            last_changelog = changelog
            changelog_files = changelog.apply(
                connection, commit=False, parameters=parameters_literals
            )
            changelog_files = [str(f) for f in changelog_files]
            self.schema_migrations.set_baseline(
                connection=connection,
                version=changelog.version,
                beta_testing=beta_testing,
                commit=False,
                changelog_files=changelog_files,
                parameters=parameters,
            )

        for post_hook in self.config.post_hook_handlers():
            post_hook.execute(connection=connection, commit=False, parameters=parameters)

        logger.info(
            "Installed %s.pum_migrations table and applied changelogs up to version %s",
            self.config.config.pum.migration_table_schema,
            last_changelog.version,
        )

        if grant:
            self.config.role_manager().grant_permissions(connection=connection, commit=False)

        if commit:
            connection.commit()
            logger.info("Changes committed to the database.")

    def install_demo_data(
        self,
        connection: psycopg.Connection,
        name: str,
        *,
        parameters: dict | None = None,
    ) -> None:
        """Install demo data for the module.

        Args:
            connection: The database connection to use.
            name: The name of the demo data to install.
            parameters: The parameters to pass to the demo data SQL.
        """
        if name not in self.config.demo_data():
            raise PumException(f"Demo data '{name}' not found in the configuration.")

        demo_data_file = self.config.base_path / self.config.demo_data()[name]
        logger.info("Installing demo data from %s", demo_data_file)

        for pre_hook in self.config.pre_hook_handlers():
            pre_hook.execute(connection=connection, commit=False, parameters=parameters)

        connection.commit()

        parameters_literals = SqlContent.prepare_parameters(parameters)
        SqlContent(sql=demo_data_file).execute(
            connection=connection,
            commit=False,
            parameters=parameters_literals,
        )

        connection.commit()

        for post_hook in self.config.post_hook_handlers():
            post_hook.execute(connection=connection, commit=False, parameters=parameters)

        logger.info("Demo data '%s' installed successfully.", name)

__init__ 

__init__(config: PumConfig, max_version: Version | str | None = None) -> None

Initialize the Upgrader class. This class is used to install a new instance or to upgrade an existing instance of a module. Stores the info about the upgrade in a table on the database. The table is created in the schema defined in the config file if it does not exist.

Parameters:

Name Type Description Default
connection

The database connection to use for the upgrade.

 required
config PumConfig

The configuration object

 required
max_version Version | str | None

Maximum (including) version to run the deltas up to.

 None
Source code in pum/upgrader.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    config: PumConfig,
    max_version: packaging.version.Version | str | None = None,
) -> None:
    """Initialize the Upgrader class.
    This class is used to install a new instance or to upgrade an existing instance of a module.
    Stores the info about the upgrade in a table on the database.
    The table is created in the schema defined in the config file if it does not exist.

    Args:
        connection:
            The database connection to use for the upgrade.
        config:
            The configuration object
        max_version:
            Maximum (including) version to run the deltas up to.

    """
    self.config = config
    self.max_version = packaging.parse(max_version) if max_version else None
    self.schema_migrations = SchemaMigrations(self.config)

install 

install(connection: Connection = None, *, parameters: dict | None = None, max_version: str | Version | None = None, roles: bool = False, grant: bool = False, beta_testing: bool = False, commit: bool = False) -> None

Installs the given module This will create the schema_migrations table if it does not exist. The changelogs are applied in the order they are found in the directory. It will also set the baseline version to the current version of the module.

Parameters:

Name Type Description Default
connection Connection

The database connection to use for the upgrade.

 None
parameters dict | None

The parameters to pass for the migration.

 None
max_version str | Version | None

The maximum version to apply. If None, all versions are applied.

 None
roles bool

If True, roles will be created.

 False
grant bool

If True, permissions will be granted to the roles.

 False
beta_testing bool

If True, the module is installed in beta testing mode. This means that the module will not be able to receive any future updates. We strongly discourage using this for production.

 False
commit bool

If True, the changes will be committed to the database.

 False
Source code in pum/upgrader.py 
 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
def install(
    self,
    connection: psycopg.Connection = None,
    *,
    parameters: dict | None = None,
    max_version: str | packaging.version.Version | None = None,
    roles: bool = False,
    grant: bool = False,
    beta_testing: bool = False,
    commit: bool = False,
) -> None:
    """Installs the given module
    This will create the schema_migrations table if it does not exist.
    The changelogs are applied in the order they are found in the directory.
    It will also set the baseline version to the current version of the module.

    Args:
        connection:
            The database connection to use for the upgrade.
        parameters:
            The parameters to pass for the migration.
        max_version:
            The maximum version to apply. If None, all versions are applied.
        roles:
            If True, roles will be created.
        grant:
            If True, permissions will be granted to the roles.
        beta_testing:
            If True, the module is installed in beta testing mode.
            This means that the module will not be able to receive any future updates.
            We strongly discourage using this for production.
        commit:
            If True, the changes will be committed to the database.
    """
    if self.schema_migrations.exists(connection):
        msg = (
            f"Schema migrations table {self.config.config.pum.migration_table_schema}.pum_migrations already exists. "
            "This means that the module is already installed or the database is not empty. "
            "Use upgrade() to upgrade the db or start with a clean db."
        )
        raise PumException(msg)
    self.schema_migrations.create(connection, commit=False)

    if roles or grant:
        self.config.role_manager().create_roles(
            connection=connection, grant=False, commit=False
        )

    for pre_hook in self.config.pre_hook_handlers():
        pre_hook.execute(connection=connection, commit=False, parameters=parameters)

    parameters_literals = SqlContent.prepare_parameters(parameters)
    last_changelog = None
    for changelog in self.config.changelogs(max_version=max_version):
        last_changelog = changelog
        changelog_files = changelog.apply(
            connection, commit=False, parameters=parameters_literals
        )
        changelog_files = [str(f) for f in changelog_files]
        self.schema_migrations.set_baseline(
            connection=connection,
            version=changelog.version,
            beta_testing=beta_testing,
            commit=False,
            changelog_files=changelog_files,
            parameters=parameters,
        )

    for post_hook in self.config.post_hook_handlers():
        post_hook.execute(connection=connection, commit=False, parameters=parameters)

    logger.info(
        "Installed %s.pum_migrations table and applied changelogs up to version %s",
        self.config.config.pum.migration_table_schema,
        last_changelog.version,
    )

    if grant:
        self.config.role_manager().grant_permissions(connection=connection, commit=False)

    if commit:
        connection.commit()
        logger.info("Changes committed to the database.")

install_demo_data 

install_demo_data(connection: Connection, name: str, *, parameters: dict | None = None) -> None

Install demo data for the module.

Parameters:

Name Type Description Default
connection Connection

The database connection to use.

 required
name str

The name of the demo data to install.

 required
parameters dict | None

The parameters to pass to the demo data SQL.

 None
Source code in pum/upgrader.py 
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
def install_demo_data(
    self,
    connection: psycopg.Connection,
    name: str,
    *,
    parameters: dict | None = None,
) -> None:
    """Install demo data for the module.

    Args:
        connection: The database connection to use.
        name: The name of the demo data to install.
        parameters: The parameters to pass to the demo data SQL.
    """
    if name not in self.config.demo_data():
        raise PumException(f"Demo data '{name}' not found in the configuration.")

    demo_data_file = self.config.base_path / self.config.demo_data()[name]
    logger.info("Installing demo data from %s", demo_data_file)

    for pre_hook in self.config.pre_hook_handlers():
        pre_hook.execute(connection=connection, commit=False, parameters=parameters)

    connection.commit()

    parameters_literals = SqlContent.prepare_parameters(parameters)
    SqlContent(sql=demo_data_file).execute(
        connection=connection,
        commit=False,
        parameters=parameters_literals,
    )

    connection.commit()

    for post_hook in self.config.post_hook_handlers():
        post_hook.execute(connection=connection, commit=False, parameters=parameters)

    logger.info("Demo data '%s' installed successfully.", name)