How AsyncMigrationManager Handles Database Schema Migrations During Startup

The AsyncMigrationManager automatically checks the SurrealDB schema version and applies pending migrations on every FastAPI startup through the lifespan context manager.

The lfnovo/open-notebook repository implements an automatic database migration system for its SurrealDB backend. When the FastAPI application launches, the AsyncMigrationManager ensures the database schema stays synchronized with the application code without requiring manual intervention. This self-healing migration process runs entirely during the API startup sequence, preventing the server from running against an outdated schema.

The Migration Flow on API Startup

The migration process triggers automatically when the FastAPI lifespan context manager executes. Located in api/main.py, this startup hook orchestrates the entire schema synchronization workflow.

Triggering the Migration Check

Inside the lifespan context manager at lines 118-125 of api/main.py, the application instantiates the migration manager and immediately checks for pending updates:

# api/main.py – inside the lifespan context manager

migration_manager = AsyncMigrationManager()
if await migration_manager.needs_migration():
    await migration_manager.run_migration_up()

This code creates a fresh AsyncMigrationManager instance, queries the current schema version, and conditionally applies any missing migrations before the API begins accepting requests.

Determining the Current Schema Version

The manager reads the existing database state through get_current_version() defined in open_notebook/database/async_migrate.py. This method queries the _sbl_migrations table to retrieve the highest applied version number. If the table does not exist—such as during a fresh installation—the function returns 0, signaling that all migrations must run.

Executing Pending Migrations

When needs_migration() detects that the current version is lower than the latest available migration, the system enters the upgrade phase. The run_migration_up() method awaits AsyncMigrationRunner.run_all(), which iterates through the up_migrations list starting from the current version index.

For each pending migration, the runner:

1. Executes the SQL via await connection.query(self.sql)
2. Records the successful application via bump_version()
3. Updates the _sbl_migrations table with a new version entry

If any step fails, the startup aborts with a RuntimeError to prevent the API from running against an incomplete schema.

Core Architecture of AsyncMigrationManager

The migration subsystem in open_notebook/database/async_migrate.py organizes schema changes into discrete, reversible units with strict ordering guarantees.

Migration Lists and Ordering

The AsyncMigrationManager initializes two ordered lists during instantiation:

• up_migrations: Contains AsyncMigration objects representing forward schema changes, loaded from SQL files in open_notebook/database/migrations/
• down_migrations: Contains rollback operations for reversing schema changes

Each AsyncMigration object reads its corresponding SQL file, strips comments, and stores a clean SQL string for execution. The manager exposes get_latest_version() to determine the total count of available migrations, enabling the needs_migration() comparison.

AsyncMigrationRunner Execution Loop

The run_all() method at lines 66-73 of async_migrate.py handles the actual execution flow:

# From open_notebook/database/async_migrate.py

async def run_all(self, connection, current_version, migrations):
    for i in range(current_version, len(migrations)):
        migration = migrations[i]
        await migration.run(connection, bump=True)

When bump=True, the run method automatically increments the version in the tracking table after successful SQL execution.

Version Tracking in SurrealDB

The _sbl_migrations table serves as the authoritative registry of applied schema versions. The helper functions bump_version() and lower_version() manage this table through explicit SQL INSERT and DELETE operations. The bump_version() implementation at lines 220-228 inserts a new record marking the successful application of a specific migration version, while get_all_versions() retrieves the complete history for debugging or verification purposes.

Implementation Examples and Usage Patterns

Beyond the automatic startup behavior, the AsyncMigrationManager provides granular control for maintenance tasks and custom scripts.

Running a Single Migration Manually

To apply only the next pending migration from a standalone script:

import asyncio
from open_notebook.database.async_migrate import AsyncMigrationManager

async def migrate_one():
    manager = AsyncMigrationManager()
    # Apply the next pending migration only

    await manager.run_one_up()

if __name__ == "__main__":
    asyncio.run(migrate_one())

This pattern is useful for staged deployments or testing specific schema changes in isolation.

Rolling Back the Last Migration

The down-migration capability allows reverting the most recent schema change:

import asyncio
from open_notebook.database.async_migrate import AsyncMigrationManager

async def rollback():
    manager = AsyncMigrationManager()
    await manager.run_one_down()   # removes the latest version entry

if __name__ == "__main__":
    asyncio.run(rollback())

The run_one_down() method executes the corresponding SQL from the down_migrations list and calls lower_version() to remove the version record from _sbl_migrations.

Summary

• AsyncMigrationManager automatically triggers during FastAPI startup via the lifespan context manager in api/main.py.
• The system queries the _sbl_migrations table to determine current schema version, defaulting to 0 for fresh databases.
• Pending migrations execute sequentially through AsyncMigrationRunner, with each SQL file applied transactionally and version numbers recorded immediately after success.
• Migration files reside in open_notebook/database/migrations/ and load into ordered lists of AsyncMigration objects.
• Startup aborts with a RuntimeError if migrations fail, ensuring the API never runs against an outdated schema.
• Manual operations like run_one_up() and run_one_down() support maintenance tasks outside the automatic flow.

Frequently Asked Questions

How does the migration manager know which schema version the database currently has?

The manager calls get_current_version(), which queries the _sbl_migrations table in SurrealDB to find the highest applied version number. If the table does not exist, the function returns 0, indicating a fresh database that requires all migrations.

What happens if a migration fails during API startup?

The startup process aborts immediately with a RuntimeError. According to the implementation in open_notebook/database/async_migrate.py, the exception propagates up through the lifespan context manager, preventing the FastAPI application from fully initializing and protecting against running code against an incomplete or corrupted schema.

Can I run migrations manually without starting the full API?

Yes. Instantiate AsyncMigrationManager and call run_one_up() to apply the next pending migration, or run_migration_up() to apply all pending migrations. These methods operate independently of the FastAPI lifespan cycle and can run from standalone scripts or maintenance commands.

Where does AsyncMigrationManager load the migration SQL files from?

The manager discovers migration files in the open_notebook/database/migrations/ directory. During initialization, it scans this location and creates ordered lists of AsyncMigration objects—one for upward migrations and one for downward migrations—ensuring SQL executes in the correct sequence based on filename ordering.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}