tests exactly like code-blocks

Author: euri10

docs/examples/usage/usage_migrations_1.py

@@ -0,0 +1,24 @@
+# start-example
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+__all__ = ("test_async_methods",)
+
+
+config = AsyncpgConfig(
+    pool_config={"dsn": "postgresql://user:pass@localhost/mydb"},
+    migration_config={"enabled": True, "script_location": "migrations"},
+)
+# end-example
+
+
+def test_async_methods() -> None:
+    # These are just smoke tests for method presence, not actual DB calls
+    assert hasattr(config, "migrate_up")
+    assert hasattr(config, "upgrade")
+    assert hasattr(config, "migrate_down")
+    assert hasattr(config, "downgrade")
+    assert hasattr(config, "get_current_migration")
+    assert hasattr(config, "create_migration")
+    assert hasattr(config, "init_migrations")
+    assert hasattr(config, "stamp_migration")
+    assert hasattr(config, "fix_migrations")

docs/examples/usage/usage_migrations_2.py

@@ -0,0 +1,49 @@
+# start-example
+from sqlspec.adapters.sqlite import SqliteConfig
+
+__all__ = ("test_sync_methods",)
+
+
+config = SqliteConfig(
+    pool_config={"database": "myapp.db"}, migration_config={"enabled": True, "script_location": "migrations"}
+)
+
+# Apply migrations (no await needed)
+config.migrate_up("head")
+# Or use the alias
+config.upgrade("head")
+
+# Rollback one revision
+config.migrate_down("-1")
+# Or use the alias
+config.downgrade("-1")
+
+# Check current version
+current = config.get_current_migration(verbose=True)
+print(current)
+
+# Create new migration
+config.create_migration("add users table", file_type="sql")
+
+# Initialize migrations directory
+config.init_migrations()
+
+# Stamp database to specific revision
+config.stamp_migration("0003")
+
+# Convert timestamp to sequential migrations
+config.fix_migrations(dry_run=False, update_database=True, yes=True)
+# end-example
+
+
+def test_sync_methods() -> None:
+    # Smoke tests for method presence, not actual DB calls
+    assert hasattr(config, "migrate_up")
+    assert hasattr(config, "upgrade")
+    assert hasattr(config, "migrate_down")
+    assert hasattr(config, "downgrade")
+    assert hasattr(config, "get_current_migration")
+    assert hasattr(config, "create_migration")
+    assert hasattr(config, "init_migrations")
+    assert hasattr(config, "stamp_migration")
+    assert hasattr(config, "fix_migrations")

docs/examples/usage/usage_migrations_3.py

@@ -0,0 +1,31 @@
+# start-example
+__all__ = ("test_template_config",)
+
+
+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,23 @@
+__all__ = ("test_async_command_class_methods",)
+
+
+async def test_async_command_class_methods() -> None:
+
+    # start-example
+    from sqlspec.adapters.asyncpg import AsyncpgConfig
+    from sqlspec.migrations.commands import AsyncMigrationCommands
+
+    config = AsyncpgConfig(pool_config={"dsn": "postgresql://..."}, migration_config={"script_location": "migrations"})
+
+    # Create commands instance
+    commands = AsyncMigrationCommands(config)
+
+    # Use commands directly
+    await commands.upgrade("head")
+    # end-example
+
+    # Smoke test for AsyncMigrationCommands method presence
+    commands = AsyncMigrationCommands(config)
+    assert hasattr(commands, "upgrade")
+    assert hasattr(commands, "downgrade")
+    assert hasattr(commands, "get_current_migration")

docs/examples/usage/usage_migrations_5.py

@@ -0,0 +1,26 @@
+# start-example
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+__all__ = ("test_config_structure",)
+
+
+config = AsyncpgConfig(
+    pool_config={"dsn": "postgresql://user:pass@localhost/mydb"},
+    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,37 @@
+# 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
+"""
+
+__all__ = ("downgrade", "test_upgrade_and_downgrade_strings", "upgrade")
+
+
+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 @@
+# start-example
+__all__ = ("test_upgrade_returns_list", "upgrade")
+
+
+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)

docs/examples/usage/usage_migrations_8.py

@@ -0,0 +1,33 @@
+__all__ = ("test_tracker_instance",)
+
+
+from pytest_databases.docker.postgres import PostgresService
+
+
+async def test_tracker_instance(postgres_service: PostgresService) -> None:
+
+    # start-example
+    from sqlspec.adapters.asyncpg import AsyncpgConfig
+    from sqlspec.migrations.tracker import AsyncMigrationTracker
+
+    tracker = AsyncMigrationTracker()
+
+    config = AsyncpgConfig(
+        pool_config={
+            "dsn": f"postgres://{postgres_service.user}:{postgres_service.password}@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
+        },
+        migration_config={
+            "enabled": True,
+            "script_location": "migrations",
+            "version_table_name": "ddl_migrations",
+            "auto_sync": True,  # Enable automatic version reconciliation
+        },
+    )
+    async with config.provide_session() as session:
+        driver = session._driver
+
+        # Update version record
+        await tracker.update_version_record(driver, old_version="20251018120000", new_version="0003")
+    # end-example
+    # Just check that tracker is an instance of AsyncMigrationTracker
+    assert isinstance(tracker, AsyncMigrationTracker)