Skip to content

API Reference

Data migration tracking and execution module.

Provides utilities for running and tracking schema/data migrations at application startup.

Exports
  • CRUD: get_migrations_not_executed, set_migration_as_executed
  • Utils: check_migrations_not_executed

check_migrations_not_executed async 

check_migrations_not_executed(db)

Check and execute any pending migrations.

Iterates over unexecuted migrations and runs each migration handler in order. Sync handlers are called directly; async handlers are awaited.

Parameters:

Name Type Description Default
db Session

Database session.

 required

Returns:

Type Description
None

None.
Source code in backend/app/migrations/utils.py 
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
async def check_migrations_not_executed(
    db: Session,
) -> None:
    """
    Check and execute any pending migrations.

    Iterates over unexecuted migrations and runs each migration handler in
    order. Sync handlers are called directly; async handlers are awaited.

    Args:
        db: Database session.

    Returns:
        None.
    """
    migrations_not_executed = migrations_crud.get_migrations_not_executed(db)

    if not migrations_not_executed:
        return

    for migration in migrations_not_executed:
        core_logger.print_to_log(f"Migration not executed: {migration.name} - Migration will be executed")

        if migration.id in _SYNC_MIGRATIONS:
            _SYNC_MIGRATIONS[migration.id](db)
        elif migration.id in _ASYNC_MIGRATIONS:
            await _ASYNC_MIGRATIONS[migration.id](db)

get_migrations_not_executed 

get_migrations_not_executed(db)

Retrieve all unexecuted migrations.

Parameters:

Name Type Description Default
db Session

Database session.

 required

Returns:

Type Description
list[MigrationRead]

List of unexecuted Migration schemas.

Raises:

Type Description
HTTPException

500 error on database failure.
Source code in backend/app/migrations/crud.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@core_decorators.handle_db_errors
def get_migrations_not_executed(
    db: Session,
) -> list[migrations_schema.MigrationRead]:
    """
    Retrieve all unexecuted migrations.

    Args:
        db: Database session.

    Returns:
        List of unexecuted Migration schemas.

    Raises:
        HTTPException: 500 error on database failure.
    """
    stmt = select(migrations_models.Migration).where(migrations_models.Migration.executed.is_(False))
    db_migrations = db.execute(stmt).scalars().all()
    return _transform_migrations(list(db_migrations))

set_migration_as_executed 

set_migration_as_executed(migration_id, db)

Mark a migration as executed by its ID.

Parameters:

Name Type Description Default
migration_id int

ID of the migration to mark as executed.

 required
db Session

Database session.

 required

Returns:

Type Description
MigrationRead

The updated Migration schema.

Raises:

Type Description
HTTPException

404 if migration not found.
HTTPException

500 error on database failure.
Source code in backend/app/migrations/crud.py 
 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
@core_decorators.handle_db_errors
def set_migration_as_executed(
    migration_id: int,
    db: Session,
) -> migrations_schema.MigrationRead:
    """
    Mark a migration as executed by its ID.

    Args:
        migration_id: ID of the migration to mark as executed.
        db: Database session.

    Returns:
        The updated Migration schema.

    Raises:
        HTTPException: 404 if migration not found.
        HTTPException: 500 error on database failure.
    """
    db_migration = _get_migration_by_id_model_or_404(migration_id, db)

    db_migration.executed = True
    db.commit()
    db.refresh(db_migration)
    return _transform_migrations(db_migration)