docs: migration usage (#265)

Migration usage docs

Author: euri10

docs/examples/arrow/arrow_basic_usage.py

@@ -75,11 +75,14 @@ async def example_adbc_native() -> None:
 # Example 2: PostgreSQL with Conversion Path
 async def example_postgres_conversion() -> None:
     """Demonstrate PostgreSQL adapter with dict → Arrow conversion."""
+    import os
+
     from sqlspec import SQLSpec
     from sqlspec.adapters.asyncpg import AsyncpgConfig
 
+    dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
     db_manager = SQLSpec()
-    asyncpg_db = db_manager.add_config(AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/test"}))
+    asyncpg_db = db_manager.add_config(AsyncpgConfig(pool_config={"dsn": dsn}))
 
     async with db_manager.provide_session(asyncpg_db) as session:
         # Create test table with PostgreSQL-specific types

docs/examples/patterns/configs/multi_adapter_registry.py

@@ -1,5 +1,7 @@
 """Show how to register multiple adapters on a single SQLSpec instance."""
 
+import os
+
 from sqlspec import SQLSpec
 from sqlspec.adapters.aiosqlite import AiosqliteConfig
 from sqlspec.adapters.asyncpg import AsyncpgConfig, AsyncpgPoolConfig
@@ -11,15 +13,12 @@
 
 def build_registry() -> "SQLSpec":
     """Create a registry with both sync and async adapters."""
+    dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
     registry = SQLSpec()
     registry.add_config(SqliteConfig(bind_key="sync_sqlite", pool_config={"database": ":memory:"}))
     registry.add_config(AiosqliteConfig(bind_key="async_sqlite", pool_config={"database": ":memory:"}))
     registry.add_config(DuckDBConfig(bind_key="duckdb_docs", pool_config={"database": ":memory:docs_duck"}))
-    registry.add_config(
-        AsyncpgConfig(
-            bind_key="asyncpg_docs", pool_config=AsyncpgPoolConfig(dsn="postgresql://user:pass@localhost:5432/db")
-        )
-    )
+    registry.add_config(AsyncpgConfig(bind_key="asyncpg_docs", pool_config=AsyncpgPoolConfig(dsn=dsn)))
     return registry
 
 

docs/examples/usage/usage_migrations_1.py

@@ -0,0 +1,50 @@
+"""Async migration commands via config methods."""
+
+import tempfile
+from pathlib import Path
+
+import pytest
+from pytest_databases.docker.postgres import PostgresService
+
+pytestmark = pytest.mark.xdist_group("postgres")
+
+__all__ = ("test_async_methods",)
+
+
+async def test_async_methods(postgres_service: PostgresService) -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        migration_dir = Path(temp_dir) / "migrations"
+        migration_dir.mkdir()
+
+        # start-example
+        from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+        dsn = (
+            f"postgresql://{postgres_service.user}:{postgres_service.password}"
+            f"@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
+        )
+        config = AsyncpgConfig(
+            pool_config={"dsn": dsn}, migration_config={"enabled": True, "script_location": str(migration_dir)}
+        )
+
+        # Initialize migrations directory (creates __init__.py if package=True)
+        await config.init_migrations()
+
+        # Create new migration file
+        await config.create_migration("add users table", file_type="sql")
+
+        # Apply migrations to head
+        await config.migrate_up("head")
+
+        # Rollback one revision
+        await config.migrate_down("-1")
+
+        # Check current version
+        await config.get_current_migration(verbose=True)
+
+        # Stamp database to specific revision
+        await config.stamp_migration("0001")
+
+        # Convert timestamp to sequential migrations
+        await config.fix_migrations(dry_run=True, update_database=False, yes=True)
+        # end-example

docs/examples/usage/usage_migrations_10.py

@@ -0,0 +1,51 @@
+"""Using AsyncMigrationTracker for version management."""
+
+import tempfile
+from pathlib import Path
+
+import pytest
+from pytest_databases.docker.postgres import PostgresService
+
+pytestmark = pytest.mark.xdist_group("postgres")
+
+__all__ = ("test_tracker_instance",)
+
+
+async def test_tracker_instance(postgres_service: PostgresService) -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        migration_dir = Path(temp_dir) / "migrations"
+        migration_dir.mkdir()
+
+        # start-example
+        from sqlspec.adapters.asyncpg import AsyncpgConfig
+        from sqlspec.migrations.tracker import AsyncMigrationTracker
+
+        # Create tracker with custom table name
+        tracker = AsyncMigrationTracker(version_table_name="ddl_migrations")
+
+        dsn = (
+            f"postgresql://{postgres_service.user}:{postgres_service.password}"
+            f"@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
+        )
+        config = AsyncpgConfig(
+            pool_config={"dsn": dsn},
+            migration_config={
+                "enabled": True,
+                "script_location": str(migration_dir),
+                "version_table_name": "ddl_migrations",
+                "auto_sync": True,  # Enable automatic version reconciliation
+            },
+        )
+
+        # Use the session to work with migrations
+        async with config.provide_session() as session:
+            # Ensure the tracking table exists
+            await tracker.ensure_tracking_table(session)
+
+            # Get current version (None if no migrations applied)
+            current = await tracker.get_current_version(session)
+            print(f"Current version: {current}")
+        # end-example
+
+        assert isinstance(tracker, AsyncMigrationTracker)
+        assert config.migration_config["version_table_name"] == "ddl_migrations"

docs/examples/usage/usage_migrations_2.py

@@ -0,0 +1,44 @@
+"""Sync migration commands via config methods."""
+
+import tempfile
+from pathlib import Path
+
+__all__ = ("test_sync_methods",)
+
+
+def test_sync_methods() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        migration_dir = Path(temp_dir) / "migrations"
+        migration_dir.mkdir()
+        temp_db = Path(temp_dir) / "test.db"
+
+        # start-example
+        from sqlspec.adapters.sqlite import SqliteConfig
+
+        config = SqliteConfig(
+            pool_config={"database": str(temp_db)},
+            migration_config={"enabled": True, "script_location": str(migration_dir)},
+        )
+
+        # Initialize migrations directory (creates __init__.py if package=True)
+        config.init_migrations()
+
+        # Create new migration file
+        config.create_migration("add users table", file_type="sql")
+
+        # Apply migrations to head (no await needed for sync)
+        config.migrate_up("head")
+
+        # Rollback one revision
+        config.migrate_down("-1")
+
+        # Check current version
+        current = config.get_current_migration(verbose=True)
+        print(current)
+
+        # Stamp database to specific revision
+        config.stamp_migration("0001")
+
+        # Convert timestamp to sequential migrations
+        config.fix_migrations(dry_run=True, update_database=False, yes=True)
+        # end-example

docs/examples/usage/usage_migrations_3.py

@@ -0,0 +1,31 @@
+__all__ = ("test_template_config",)
+
+
+# start-example
+migration_config = {
+    "default_format": "py",  # CLI default when --format omitted
+    "title": "Acme Migration",  # Shared title for all templates
+    "author": "env:SQLSPEC_AUTHOR",  # Read from environment variable
+    "templates": {
+        "sql": {
+            "header": "-- {title} - {message}",
+            "metadata": ["-- Version: {version}", "-- Owner: {author}"],
+            "body": "-- custom SQL body",
+        },
+        "py": {
+            "docstring": """{title}\nDescription: {description}""",
+            "imports": ["from typing import Iterable"],
+            "body": """def up(context: object | None = None) -> str | Iterable[str]:\n    return \"SELECT 1\"\n\ndef down(context: object | None = None) -> str | Iterable[str]:\n    return \"DROP TABLE example;\"\n""",
+        },
+    },
+}
+# end-example
+
+
+def test_template_config() -> None:
+    # Check structure of migration_config
+    assert migration_config["default_format"] == "py"
+    assert "py" in migration_config["templates"]
+    assert "sql" in migration_config["templates"]
+    assert isinstance(migration_config["templates"]["py"], dict)
+    assert isinstance(migration_config["templates"]["sql"], dict)

docs/examples/usage/usage_migrations_4.py

@@ -0,0 +1,37 @@
+"""Using AsyncMigrationCommands directly."""
+
+import os
+import tempfile
+from pathlib import Path
+
+__all__ = ("test_async_command_class_methods",)
+
+
+async def test_async_command_class_methods() -> None:
+    with tempfile.TemporaryDirectory() as temp_dir:
+        migration_dir = Path(temp_dir) / "migrations"
+        migration_dir.mkdir()
+
+        # start-example
+        from sqlspec.adapters.asyncpg import AsyncpgConfig
+        from sqlspec.migrations.commands import AsyncMigrationCommands
+
+        dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
+        config = AsyncpgConfig(pool_config={"dsn": dsn}, migration_config={"script_location": str(migration_dir)})
+
+        # Create commands instance
+        commands = AsyncMigrationCommands(config)
+
+        # Use commands directly
+        await commands.init(str(migration_dir))
+        await commands.upgrade("head")
+        # end-example
+
+        # Smoke test for AsyncMigrationCommands method presence
+        assert hasattr(commands, "upgrade")
+        assert hasattr(commands, "downgrade")
+        assert hasattr(commands, "current")
+        assert hasattr(commands, "revision")
+        assert hasattr(commands, "stamp")
+        assert hasattr(commands, "fix")
+        assert hasattr(commands, "init")

docs/examples/usage/usage_migrations_5.py

@@ -0,0 +1,29 @@
+import os
+
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+__all__ = ("test_config_structure",)
+
+
+# start-example
+dsn = os.getenv("SQLSPEC_USAGE_PG_DSN", "postgresql://localhost/db")
+config = AsyncpgConfig(
+    pool_config={"dsn": dsn},
+    migration_config={
+        "enabled": True,
+        "script_location": "migrations",
+        "version_table_name": "ddl_migrations",
+        "auto_sync": True,  # Enable automatic version reconciliation
+    },
+)
+# end-example
+
+
+def test_config_structure() -> None:
+    # Check config attributes
+    assert hasattr(config, "pool_config")
+    assert hasattr(config, "migration_config")
+    assert config.migration_config["enabled"] is True
+    assert config.migration_config["script_location"] == "migrations"
+    assert config.migration_config["version_table_name"] == "ddl_migrations"
+    assert config.migration_config["auto_sync"] is True

docs/examples/usage/usage_migrations_6.py

@@ -0,0 +1,36 @@
+__all__ = ("downgrade", "test_upgrade_and_downgrade_strings", "upgrade")
+# start-example
+# migrations/0002_add_user_roles.py
+"""Add user roles table
+
+Revision ID: 0002_add_user_roles
+Created at: 2025-10-18 12:00:00
+"""
+
+
+def upgrade() -> str:
+    """Apply migration."""
+    return """
+    CREATE TABLE user_roles (
+        id SERIAL PRIMARY KEY,
+        user_id INTEGER REFERENCES users(id),
+        role VARCHAR(50) NOT NULL
+    );
+    """
+
+
+def downgrade() -> str:
+    """Revert migration."""
+    return """
+    DROP TABLE user_roles;
+    """
+
+
+# end-example
+
+
+def test_upgrade_and_downgrade_strings() -> None:
+    up_sql = upgrade()
+    down_sql = downgrade()
+    assert "CREATE TABLE user_roles" in up_sql
+    assert "DROP TABLE user_roles" in down_sql

docs/examples/usage/usage_migrations_7.py

@@ -0,0 +1,21 @@
+__all__ = ("test_upgrade_returns_list", "upgrade")
+
+
+# start-example
+def upgrade() -> list[str]:
+    """Apply migration in multiple steps."""
+    return [
+        "CREATE TABLE products (id SERIAL PRIMARY KEY);",
+        "CREATE TABLE orders (id SERIAL PRIMARY KEY, product_id INTEGER);",
+        "CREATE INDEX idx_orders_product ON orders(product_id);",
+    ]
+
+
+# end-example
+
+
+def test_upgrade_returns_list() -> None:
+    stmts = upgrade()
+    assert isinstance(stmts, list)
+    assert any("products" in s for s in stmts)
+    assert any("orders" in s for s in stmts)