Migrations¶

Introduction¶

Migrations are version control for your database. They let you define and share the application's schema as code. Arvel's schema DSL — Schema plus Blueprint — describes tables in Python and compiles them to Alembic operations under the hood, while a Laravel-style runner tracks which migrations have run and supports batch rollback.

Generating Migrations¶

Use the make:migration command. The framework derives the table name from the migration name and timestamps the file so migrations run in order:

arvel make:migration create_flights_table

Migrations are written to database/migrations/. Files are discovered and applied in lexicographic order (the timestamp prefix ensures correct ordering), and files whose names start with _ are skipped.

Migration Structure¶

A migration defines two module-level coroutines, up and down, that receive the Schema. up makes the change; down reverses it:

from arvel.database import Blueprint, Schema


async def up(schema: Schema) -> None:
    def build(table: Blueprint) -> None:
        table.id()
        table.string("name", length=255)
        table.decimal("price", precision=10, scale=2)
        table.boolean("is_active").server_default("true")
        table.timestamps()

    Schema.create("flights", build)


async def down(schema: Schema) -> None:
    Schema.drop_if_exists("flights")

Running Migrations¶

arvel migrate              # apply all pending migrations
arvel migrate --dry-run    # print the SQL without applying it
arvel migrate:status       # show applied vs pending

Rolling Back¶

Each migrate run is recorded as a "batch". Rolling back reverses the most recent batch; resetting reverses everything:

arvel migrate:rollback     # undo the last batch
arvel migrate:reset        # roll back every applied migration

Refreshing & Dropping¶

arvel migrate:refresh      # reset, then re-run all migrations
arvel migrate:fresh        # drop all tables, then re-run all migrations

Both accept --seed (run the default seeder afterward) and --seeder NAME.

Tables¶

Creating Tables¶

Schema.create takes the table name and a callback that receives a Blueprint:

def build(table: Blueprint) -> None:
    table.id()
    table.string("email").unique()
    table.timestamps()

Schema.create("users", build)

Updating Tables¶

Use Schema.table to alter an existing table — add columns, drop columns, add indexes:

def build(table: Blueprint) -> None:
    table.string("phone").nullable()
    table.drop_column("legacy_field")

Schema.table("users", build)

Dropping Tables¶

Schema.drop("users")
Schema.drop_if_exists("users")

Columns¶

Available Column Types¶

Inside a build callback, declare columns with the Blueprint methods:

Column Modifiers¶

Columns are fluent — chain modifiers onto a column definition:

table.string("email", length=255).unique()
table.datetime("published_at").nullable()
table.integer("views").default(0)
table.boolean("active").server_default("true")
table.datetime("created_at").use_current()

Foreign Keys¶

Declare a foreign key with foreign_id and chain constrained plus a delete rule:

table.foreign_id("user_id").constrained()                 # references users.id
table.foreign_id("team_id").constrained("teams").cascade_on_delete()
table.foreign_id("owner_id").nullable().null_on_delete()

constrained() infers the referenced table by pluralizing the {name}_id column. Delete-rule shorthands: .cascade_on_delete(), .null_on_delete(), .restrict_on_delete(), or the explicit .on_delete(...) / .on_update(...).

Indexes¶

table.index(["last_name", "first_name"])
table.unique("email")
table.gin_index("documents", "search_vector")
table.expression_index("LOWER(email)", name="users_lower_email")

# drop helpers
table.drop_index("users_email_unique")
table.rename_column("old_name", "new_name")

Standalone (outside a build callback): Schema.create_index(...) and Schema.create_expression_index(...).

Views & Extensions¶

The schema builder can manage raw SQL, database extensions, and views:

Schema.run_sql("UPDATE settings SET value = '1' WHERE key = 'migrated'")
Schema.install_extension("pg_trgm")
Schema.create_view("active_users", "SELECT * FROM users WHERE active")
Schema.create_materialized_view("revenue", "SELECT ...", with_data=True)
Schema.refresh_materialized_view("revenue", concurrently=True)

Async introspection helpers — has_table, has_column, get_columns, has_view — let a migration branch on the current schema state.

Database Connections¶

Connections read config/database.py — default picks the active connection and connections maps each named connection to its settings. The DB_* environment variables are the fallback when a key isn't in the file (see the cascade). Set DB_CONNECTION to choose a driver:

DB_CONNECTION=postgresql
DB_HOST=127.0.0.1
DB_PORT=5432
DB_DATABASE=myapp
DB_USERNAME=postgres
DB_PASSWORD=secret

Alternatively set a full DB_URL, which wins over the composed fields. The DatabaseServiceProvider binds the engine, session maker, and session into the container, and pings the database on boot.

Seeding¶

Seeders populate your database with data — useful for development and tests.

Writing Seeders¶

Generate one with arvel make:seeder, then implement the async run method:

from arvel.database import Seeder
from app.models.user import User


class DatabaseSeeder(Seeder):
    async def run(self) -> None:
        await User.create(name="Ada", email="[email protected]")
        # run another seeder by awaiting its run()
        await ProductSeeder().run()

Running Seeders¶

arvel db:seed                      # runs DatabaseSeeder
arvel db:seed --seeder ProductSeeder

Factories¶

Factories generate model instances for tests and seeding, with sensible fake data. Define a Factory subclass with a definition that returns the default attributes. Generate one with arvel make:factory:

from typing import Any
from arvel.database import Factory
from app.models.user import User


class UserFactory(Factory[User]):
    model = User

    def definition(self) -> dict[str, Any]:
        return {"name": "Ada Lovelace", "email": "[email protected]"}

Make instances (in memory) or create them (persisted):

user = await UserFactory().create()
users = await UserFactory().count(3).create()
admin = await UserFactory().state({"is_admin": True}).create()
draft = UserFactory().make()           # not persisted

Build relationships, attach pivots, and run callbacks:

author = await UserFactory().has("posts", PostFactory(), count=3).create()
post = await PostFactory().for_("author", author).create()

await UserFactory().has_attached(
    "roles", RoleFactory(), count=2, pivot={"assigned_by": "system"}
).create()

await UserFactory().after_creating(lambda u, faker: notify(u)).create()