mafw.db.migration.history

Per-database migration history.

The migration history is a dedicated table stored inside each target database, so history is naturally per-database: the same set of migration scripts applied to two different databases maintains two independent histories.

The history model is created with do_not_register=True so it never enters the global model register and is therefore never itself a migration subject. The reader/writer helpers record a script only after it applies successfully and expose consistency checks that compare the recorded name/checksum against the scripts present in the migration directory.

Added in version 2.4.

author:: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)

Module Attributes

MIGRATION_HISTORY_TABLE

The fixed table name of the per-database migration history.

Functions

`check_consistency`(entries, scripts)	Verify that the history and the directory are mutually consistent.
`ensure_history_table`(database)	Create the migration-history table if it does not yet exist.
`history_table_exists`(database)	Return whether the migration-history table exists in the database.
`list_scripts`(directory)	List the migration scripts in a directory, ordered by version.
`read_history`(database)	Read the applied-migration history in chronological order.
`record_applied`(database, script)	Record a successfully applied script in the history.

Classes

`MigrationHistory`(args, *kwargs)	The per-database applied-migration history table.
`MigrationHistoryEntry`(version, name, ...)	A read-only view of a single migration-history row.

class mafw.db.migration.history.MigrationHistory(*args, **kwargs)[source]

Bases: MAFwBaseModel

The per-database applied-migration history table.

Created with do_not_register=True so it never enters mafw_model_register and is excluded from discovery and diffing.

Added in version 2.4.

DoesNotExist: alias of MigrationHistoryDoesNotExist

applied_at = <DateTimeField: MigrationHistory.applied_at>: The timestamp at which the script was applied.

checksum = <CharField: MigrationHistory.checksum>: The script content checksum recorded when the script was applied.

name = <CharField: MigrationHistory.name>: The migration script file stem.

version = <CharField: MigrationHistory.version>: The migration script version identifier (primary key).

class mafw.db.migration.history.MigrationHistoryEntry(version: str, name: str, checksum: str, applied_at: datetime)[source]

Bases: object

A read-only view of a single migration-history row.

Parameters:

version (str) – The migration script version identifier.
name (str) – The migration script file stem.
checksum (str) – The recorded content checksum.
applied_at (datetime.datetime) – The application timestamp.

Added in version 2.4.

mafw.db.migration.history.check_consistency(entries: Sequence[MigrationHistoryEntry], scripts: Sequence[MigrationScript]) → None[source]

Verify that the history and the directory are mutually consistent.

A history entry must have a corresponding script in the directory with an unchanged checksum. A mismatch indicates a missing or altered script.

Parameters:

entries (Sequence[MigrationHistoryEntry]) – The applied-history entries.
scripts (Sequence[MigrationScript]) – The scripts present in the directory.

Raises:

MigrationHistoryInconsistencyError – when the state is inconsistent.

mafw.db.migration.history.ensure_history_table(database: Database) → None[source]

Create the migration-history table if it does not yet exist.

Parameters:: database (peewee.Database) – The database in which to create the table.

mafw.db.migration.history.history_table_exists(database: Database) → bool[source]

Return whether the migration-history table exists in the database.

Parameters:: database (peewee.Database) – The database to inspect.
Returns:: True when the history table is present.
Return type:: bool

mafw.db.migration.history.list_scripts(directory: Path) → list[MigrationScript][source]

List the migration scripts in a directory, ordered by version.

Parameters:: directory (Path) – The migration directory to scan.
Returns:: The scripts found, ordered by their version identifier.
Return type:: list[MigrationScript]

mafw.db.migration.history.read_history(database: Database) → list[MigrationHistoryEntry][source]

Read the applied-migration history in chronological order.

Parameters:: database (peewee.Database) – The database to read from.
Returns:: The applied entries ordered by application timestamp then version.
Return type:: list[MigrationHistoryEntry]

mafw.db.migration.history.record_applied(database: Database, script: MigrationScript) → MigrationHistoryEntry[source]

Record a successfully applied script in the history.

Parameters:

database (peewee.Database) – The database whose history to update.
script (MigrationScript) – The script that was applied.

Returns:

The recorded history entry.

Return type:

MigrationHistoryEntry

mafw.db.migration.history.MIGRATION_HISTORY_TABLE = 'mafw_migration_history': The fixed table name of the per-database migration history.