aequilibrae.project.tools.migration_manager#

Classes

`Migration`(id, name, file[, type])	Small utility class to wrap files used for database upgrades/migrations.
`MigrationManager`(migration_file)	Small utility class to manage, validate, and apply a set of `Migration`s.
`MigrationStatus`(value)	An enumeration.

class aequilibrae.project.tools.migration_manager.Migration(id: int, name: str, file: Path, type: str | None = None)[source]#

Small utility class to wrap files used for database upgrades/migrations.

Individual migrations can report their status, be marked as ‘seen’ or as another status, and applied. SQL migrations are executed using sqlite3.executescript. Python migrations are loaded as a module, they should expose a migrate function which accepts an sqlite3.Connection as a single positional argument.

Marking a migration as ‘seen’ will add it to the migrations table as MISSING if it is not already present. If it is present no change is made.

Applying a migration will update the status to ‘APPLIED’ with the current timestamp.

A migration’s status cannot be downgraded without force.

Migrations are identified based on their id attribute and the id field of the migrations table.

apply(conn: Connection, connections: dict[str, Connection])[source]#

Apply this migration.

Successful application will mark the migration as APPLIED.

Python migrations should never use executescript as it will commit the pending transaction and place SQLite in autocommit mode. If the migration then fails the database will be bad state.

Arguments:

conn (sqlite3.Connection): Main SQLite database connection. This is connection is used for the migrations table and ‘.sql’ migrations.

connections (dict[str, sqlite3.Connection]): All open SQLite connections. Passed as keyword arguments for Python migrations.

mark_as(conn: Connection, status: MigrationStatus, force: bool = False)[source]#

Update or insert this migration with the given status.

If the migration is not present in the table it will be inserted. If it is present and the new status is a ‘upgrade’ or force=True, then it will be updated. Otherwise no change will be made.

Arguments:

conn (sqlite3.Connection): SQLite database connection.

status (MigrationStatus): Migration status enum.

mark_as_seen(conn: Connection)[source]#

Mark this migration as ‘seen’.

Marking a migration as ‘seen’ will add it to the migrations table as MISSING if it is not already present. If it is present no change is made.

Arguments:: conn (sqlite3.Connection): SQLite database connection.

status(conn: Connection) → MigrationStatus[source]#

Query the database for this migrations status.

If the migrations table is not present all migrations are considered MISSING.

Arguments:: conn (sqlite3.Connection): SQLite database connection.
Returns:: status (MigrationStatus): Migration status enum.

file: Path#

id: int#

name: str#

type: str = None#

class aequilibrae.project.tools.migration_manager.MigrationManager(migration_file: Path)[source]#

Small utility class to manage, validate, and apply a set of Migrations.

Arguments:: migration_file (pathlib.Path): A path to a Python with which defines a global migrations variable as a list of pathlib.Path to migrations.

find_applicable(conn: Connection)[source]#

Find all applicable migrations.

A migration is applicable if all migrations before it (ordered by ID) have been applied or skipped.

If an out-of-order migration is detected a RuntimeError will be raised and manual intervention will be required.

Arguments:: conn (sqlite3.Connection): SQLite database connection.

mark_all_as_seen(conn: Connection)[source]#

Mark all migrations as ‘seen’.

Marking a migration as ‘seen’ will add it to the migrations table as MISSING if it is not already present. If it is present no change is made.

Arguments:: conn (sqlite3.Connection): SQLite database connection.

status(conn: Connection) → dict[int, MigrationStatus][source]#

Query the database for all migrations’ status.

If the migrations table is not present all migrations are considered MISSING.

Arguments:: conn (sqlite3.Connection): SQLite database connection.
Returns:: status (dict[int, MigrationStatus]): Migration status enums by their ID.

upgrade(main_conn: str, connections: dict[str, AequilibraEConnection | None], skip: set[int] | None = None)[source]#

Find and apply all applicable migrations.

Optionally skip some migrations. Take care when skipping migrations.

Arguments:

main_conn (str): Main SQLite database connection. This is connection is used for the migrations table. Must be a key in connections.

skip (set[int]): Set of migration IDs to skip.

connections (dict[str, Optional[AequilibraEConnection]]): Dictionary mapping connection names to AequilibraEConnection objects. These connections are used during the migration process.

network_migration_file = PosixPath('/home/runner/work/aequilibrae/aequilibrae/aequilibrae/project/database_specification/network/migrations/migrations.py')#

transit_migration_file = PosixPath('/home/runner/work/aequilibrae/aequilibrae/aequilibrae/project/database_specification/transit/migrations/migrations.py')#

class aequilibrae.project.tools.migration_manager.MigrationStatus(value)[source]#

An enumeration.

as_integer_ratio()#

Return integer ratio.

Return a pair of integers, whose ratio is exactly equal to the original int and with a positive denominator.

>>> (10).as_integer_ratio()
(10, 1)
>>> (-10).as_integer_ratio()
(-10, 1)
>>> (0).as_integer_ratio()
(0, 1)

bit_count()#

Number of ones in the binary representation of the absolute value of self.

Also known as the population count.

>>> bin(13)
'0b1101'
>>> (13).bit_count()
3

bit_length()#

Number of bits necessary to represent self in binary.

>>> bin(37)
'0b100101'
>>> (37).bit_length()
6

conjugate()#: Returns self, the complex conjugate of any int.

from_bytes(byteorder, *, signed=False)#

Return the integer represented by the given array of bytes.

bytes: Holds the array of bytes to convert. The argument must either support the buffer protocol or be an iterable object producing bytes. Bytes and bytearray are examples of built-in objects that support the buffer protocol.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Indicates whether two’s complement is used to represent the integer.

to_bytes(length, byteorder, *, signed=False)#

Return an array of bytes representing an integer.

length: Length of bytes object to use. An OverflowError is raised if the integer is not representable with the given number of bytes.
byteorder: The byte order used to represent the integer. If byteorder is ‘big’, the most significant byte is at the beginning of the byte array. If byteorder is ‘little’, the most significant byte is at the end of the byte array. To request the native byte order of the host system, use `sys.byteorder’ as the byte order value.
signed: Determines whether two’s complement is used to represent the integer. If signed is False and a negative integer is given, an OverflowError is raised.

APPLIED: int = 3#

MISSING: int = 1#

SKIPPED: int = 2#

denominator#: the denominator of a rational number in lowest terms

imag#: the imaginary part of a complex number

numerator#: the numerator of a rational number in lowest terms

real#: the real part of a complex number

aequilibrae.project.tools.migration_manager#

This Page