Skip to content
Documentation
April Robots Documentation
Database Migrations

Database Migrations

How it works (Online App)

  1. On every app start (db_lifespan) the runner:
  • Scans app/database/migrations/versions/ for files named vX_Y_Z.py
  • Compares each version with the one stored in migrations collection
  • Executes INIT migrations for every missing version ≤ current APP_VERSION
  1. On every cloud sync (run_all_pending(data=…)) the runner:
  • Copies incoming sessions & children into tmp_sessions / tmp_children
  • Executes SYNC migrations on the tmp collections
  • Replaces the real collections with the migrated tmp data
  1. Tracking
  • Collection migrations stores {version, status, started_at, completed_at, error}

Create a new migration

1. Create the file 

app/database/migrations/
├── __init__.py
├── runner.py
├── registry.py
├── types.py
├── version_tracker.py
└── versions/        <-- create new files here
   ├── v0_0_0.py
   ├── v1_0_0.py
   └── v1_1_0.py

Make sure to follow the vX_Y_Z.py naming convention (corresponding to an app version) and bump APP_VERSION in app.config to make sure migrations are executed.

2. Write the code

A sample migration to add:

from motor.motor_asyncio import AsyncIOMotorDatabase
from app.common.utils import get_now_utc


# INIT migrations run once on app launch
async def add_default_language(db: AsyncIOMotorDatabase):
   await db["children"].update_many(
       {"language": {"$exists": False}},
       {"$set": {"language": "en"}}
   )


# SYNC migrations run on every cloud → device sync
async def migrate_sessions_add_language(db: AsyncIOMotorDatabase):
   async for session in db["tmp_sessions"].find():
       if "language" not in session:
           session["language"] = "en"
           await db["tmp_sessions"].replace_one(
               {"_id": session["_id"]}, session, upsert=True
           )


# Export these two lists, either can be skipped, with the exact names
INIT_MIGRATIONS = [
   add_default_language,
]


SYNC_MIGRATIONS = [
   migrate_sessions_add_language,
]

3. Rules & best practices

RuleWhy
Never touch the real collection in SYNC migrationsUse only tmp_sessions, tmp_children
Keep migrations idempotentRunner may retry on failure
Never delete data without a backup planAdd a new field first, then drop the old one in a later migration
One logical change = one migrationMakes rollbacks trivial
Test locallyRUN_MIGRATIONS_UPON_LAUNCH=True in app.config + fresh DB

FAQ

Q: Can I run migrations manually? Yes:

from app.database.migrations.runner import AsyncMigrationRunner
runner = AsyncMigrationRunner(db, "99.99.99")
await runner.run_all_pending() # For init migrations
await runner.run_all_pending( data: SyncDataDict ) # For cloud migrations

Q: What if an init migration fails?

  • Record goes to status: failed
  • App crashes with SystemError
  • Fix the code → bump version → redeploy

Q: Do init migrations always run on launch?

Set RUN_MIGRATIONS_UPON_LAUNCH=False in app.config. This will prevent INIT_MIGRATIONS from executing immediately.

Offline App

Offline App does not use SYNC_MIGRATIONS, so only INIT_MIGRATIONS should be declared. SYNC_MIGRATIONS will be ignored. The rest is almost identical to the Online App, but the module is adapted to work with DictDataBase, and all database files are backed up before running migrations in case a crash happens. To restore from backups, use the MigrationRunner.restore_db( app_version: str ) method.

Adding New Action ScriptsRelease Docs & Versioning