Migrations — Aquilia Documentation
Comprehensive guide and documentation for Migrations in the Aquilia framework. View API reference, examples, and implementation patterns.
Docs / Models / Migrations Migrations Aquilia's 4-layer migration system — from high-level DSL operations to raw DDL, with auto-generation, tracking, and rollback support. Architecture The migration system is organized in four layers, from highest to lowest level: Layer Module Role Auto-Generator migration_gen.py Diff current models against DB schema → generate DSL migration files DSL migration_dsl.py High-level operations: CreateModel, AddField, RunSQL, RunPython, etc. Runner migration_runner.py Execute, track, rollback, plan, status. Manages aquilia_migrations table. DDL Ops migrations.py Raw SQL builders: create/drop/rename tables, add/alter/drop columns, indexes, constraints Migration DSL Migrations are Python files in your migrations/ directory. Each file defines a Migration class with a list of operations. from aquilia.models.migration_dsl import Migration, CreateModel, AddField, CreateIndex, C class InitialMigration(Migration): """Create users and posts tables.""" dependencies = [] # no prior migration operations = [ CreateModel( name="users", columns=[ C.bigserial("id").primary_key(), C.varchar("name", 150).not_null(), C.varchar("email", 254).not_null().unique(), C.boolean("active").default(True), C.timestamp("created_at").default("NOW()"), ], ), CreateModel( name="posts", columns=[ C.bigserial("id").primary_key(), C.varchar("title", 200).not_null(), C.text("body"), C.bigint("author_id").not_null().references("users", "id", on_delete="CASCADE"), C.timestamp("published_at").nullable(), ], ), CreateIndex( table="posts", columns=["author_id"], name="idx_posts_author", ), ] DSL Operations Operation Description Reversible ))} Column Definitions — C Namespace The C class provides a fluent builder for column definitions used in CreateModel and AddField operations: from aquilia.models.migration_dsl import C # C. (name, ...) returns a ColumnDef with chainable methods: # .not_null() — add NOT NULL constraint # .nullable() — explicitly allow NULL # .primary_key() — mark as PRIMARY KEY # .unique() — add UNIQUE constraint # .default(val) — set DEFAULT value # .references(table, column, on_delete=...) — add FOREIGN KEY # .check(expr) — add CHECK constraint # Examples C.bigserial("id").primary_key() C.varchar("name", 150).not_null() C.integer("age").nullable().default(0) C.text("bio") C.boolean("active").not_null().default(True) C.timestamp("created_at").default("NOW()") C.decimal("price", 10, 2).not_null().check("price > 0") C.bigint("user_id").references("users", "id", on_delete="CASCADE") C.jsonb("metadata").default("' '") C.uuid("public_id").not_null().unique() # Available types: # C.integer, C.bigint, C.smallint, C.serial, C.bigserial # C.varchar, C.text, C.char # C.boolean # C.real, C.double, C.decimal, C.numeric # C.date, C.time, C.timestamp, C.interval # C.blob, C.bytea # C.json, C.jsonb # C.uuid # C.inet, C.cidr, C.macaddr # C.array(base_type) # C.custom(sql_type) Data Migrations Use RunSQL and RunPython for data transformations that go beyond schema changes. from aquilia.models.migration_dsl import Migration, RunSQL, RunPython async def backfill_slugs(db): """Generate slugs for existing articles.""" rows = await db.fetch_all("SELECT id, title FROM articles WHERE slug IS NULL") for row in rows: slug = row["title"].lower().replace(" ", "-")[:50] await db.execute("UPDATE articles SET slug = ? WHERE id = ?", [slug, row["id"]]) async def reverse_slugs(db): """Clear all slugs.""" await db.execute("UPDATE articles SET slug = NULL") class BackfillSlugsMigration(Migration): dependencies = ["0002_add_slug"] operations = [ # RunPython — callable with optional reverse RunPython( forward=backfill_slugs, reverse=reverse_slugs, ), # RunSQL — raw SQL with optional reverse SQL RunSQL( sql="UPDATE articles SET status = 'draft' WHERE status IS NULL", reverse_sql="UPDATE articles SET status = NULL WHERE status = 'draft'", ), ] Migration Runner The MigrationRunner handles execution, tracking, and rollback of migrations. It uses an aquilia_migrations tracking table. from aquilia.models.migration_runner import MigrationRunner # Initialize — auto-creates tracking table runner = MigrationRunner(db, migrations_dir="migrations/") await runner.init() # Status — show all migrations and their state status = await runner.status() # [ # , # , # ] # Pending — only unapplied migrations pending = await runner.get_pending() # ["0002_add_slug", "0003_backfill_slugs"] # Applied — already executed applied = await runner.get_applied() # ["0001_initial"] # Plan — dry-run showing what would execute plan = await runner.plan() # [ , ...] # sqlmigrate — view SQL without executing sql = await runner.sqlmigrate("0002_add_slug") # "ALTER TABLE articles ADD COLUMN slug VARCHAR(50);" # Migrate — apply all pending migrations await runner.migrate() # Migrate to specific target await runner.migrate(target="0002_add_slug") # Rollback — reverse the last applied migration await runner.rollback() # Rollback to specific target await runner.rollback(target="0001_initial") Auto-Generation generate_dsl_migration() compares your current model definitions against the database schema and generates a migration file with the necessary operations. from aquilia.models.migration_gen import generate_dsl_migration # Generate migration from schema diff migration_code = await generate_dsl_migration( db=db, models=[User, Post, Comment], # current model classes migrations_dir="migrations/", # where to find existing migrations name="auto", # migration name (auto-numbered) ) # The generated code is a valid Python file with: # - CreateModel for new tables # - AddField for new columns # - AlterField for changed columns # - RemoveField for deleted columns # - CreateIndex / DropIndex for index changes # Write to file with open("migrations/0004_auto.py", "w") as f: f.write(migration_code) # Review and apply await runner.migrate() Low-Level DDL — MigrationOps For advanced or dynamic schema changes, use MigrationOps directly. This is the lowest-level API that DSL operations compile into. from aquilia.models.migrations import MigrationOps ops = MigrationOps(db) # Table operations await ops.create_table("users", ) await ops.drop_table("old_table") await ops.rename_table("users", "accounts") await ops.table_exists("users") # → True # Column operations await ops.add_column("users", "age", "INTEGER DEFAULT 0") await ops.drop_column("users", "legacy_field") await ops.rename_column("users", "name", "full_name") await ops.alter_column("users", "email", "TEXT NOT NULL") await ops.column_exists("users", "age") # → True # Index operations await ops.create_index("users", ["email"], unique=True, name="idx_email") await ops.drop_index("idx_email") # Constraint operations await ops.add_constraint("users", "ck_age", "CHECK (age >= 0)") await ops.drop_constraint("users", "ck_age") # Column type helpers ops.integer() # → "INTEGER" ops.bigint() # → "BIGINT" ops.varchar(100) # → "VARCHAR(100)" ops.text() # → "TEXT" ops.boolean() # → "BOOLEAN" / "INTEGER" (SQLite) ops.timestamp() # → "TIMESTAMP" / "TIMESTAMPTZ" ops.decimal(10, 2) # → "DECIMAL(10,2)" ops.jsonb() # → "JSONB" / "TEXT" (SQLite) ops.uuid() # → "UUID" / "VARCHAR(36)" (SQLite) ops.serial() # → "SERIAL" / "INTEGER" (SQLite) ops.bigserial() # → "BIGSERIAL" / "INTEGER" (SQLite) ops.array("TEXT") # → "TEXT[]" / "TEXT" (SQLite) # Foreign key helper await ops.add_foreign_key( table="posts", column="author_id", ref_table="users", ref_column="id", on_delete="CASCADE", ) Tracking Table The runner stores migration state in aquilia_migrations: CREATE TABLE IF NOT EXISTS aquilia_migrations ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, applied TEXT NOT NULL DEFAULT (datetime('now')) ); -- Each row = one applied migration -- name: migration filename without .py extension -- applied: ISO timestamp when it was applied Migration Signals Hook into migration execution with the pre_migrate and post_migrate signals. from aquilia.models.signals import pre_migrate, post_migrate, receiver @receiver(pre_migrate) async def before_migration(sender, migration_name, **kwargs): print(f"About to apply: ") @receiver(post_migrate) async def after_migration(sender, migration_name, **kwargs): print(f"Successfully applied: ") # Good place to seed data, clear caches, etc. Relationships Signals );
Go to Homepage