feat: migration convenience methods to config classes (#218)

Introduce new migration methods to database configuration classes, simplifying the migration process by allowing direct calls on config instances.

Author: cofin

AGENTS.md

@@ -183,6 +183,94 @@ SQLSpec is a type-safe SQL query mapper designed for minimal abstraction between
 - **Parameter Style Abstraction**: Automatically converts between different parameter styles (?, :name, $1, %s)
 - **Type Safety**: Supports mapping results to Pydantic, msgspec, attrs, and other typed models
 - **Single-Pass Processing**: Parse once → transform once → validate once - SQL object is single source of truth
+- **Abstract Methods with Concrete Implementations**: Protocol defines abstract methods, base classes provide concrete sync/async implementations
+
+### Protocol Abstract Methods Pattern
+
+When adding methods that need to support both sync and async configurations, use this pattern:
+
+**Step 1: Define abstract method in protocol**
+
+```python
+from abc import abstractmethod
+from typing import Awaitable
+
+class DatabaseConfigProtocol(Protocol):
+    is_async: ClassVar[bool]  # Set by base classes
+
+    @abstractmethod
+    def migrate_up(
+        self, revision: str = "head", allow_missing: bool = False, auto_sync: bool = True, dry_run: bool = False
+    ) -> "Awaitable[None] | None":
+        """Apply database migrations up to specified revision.
+
+        Args:
+            revision: Target revision or "head" for latest.
+            allow_missing: Allow out-of-order migrations.
+            auto_sync: Auto-reconcile renamed migrations.
+            dry_run: Show what would be done without applying.
+        """
+        raise NotImplementedError
+```
+
+**Step 2: Implement in sync base class (no async/await)**
+
+```python
+class NoPoolSyncConfig(DatabaseConfigProtocol):
+    is_async: ClassVar[bool] = False
+
+    def migrate_up(
+        self, revision: str = "head", allow_missing: bool = False, auto_sync: bool = True, dry_run: bool = False
+    ) -> None:
+        """Apply database migrations up to specified revision."""
+        commands = self._ensure_migration_commands()
+        commands.upgrade(revision, allow_missing, auto_sync, dry_run)
+```
+
+**Step 3: Implement in async base class (with async/await)**
+
+```python
+class NoPoolAsyncConfig(DatabaseConfigProtocol):
+    is_async: ClassVar[bool] = True
+
+    async def migrate_up(
+        self, revision: str = "head", allow_missing: bool = False, auto_sync: bool = True, dry_run: bool = False
+    ) -> None:
+        """Apply database migrations up to specified revision."""
+        commands = cast("AsyncMigrationCommands", self._ensure_migration_commands())
+        await commands.upgrade(revision, allow_missing, auto_sync, dry_run)
+```
+
+**Key principles:**
+
+- Protocol defines the interface with union return type (`Awaitable[T] | T`)
+- Sync base classes implement without `async def` or `await`
+- Async base classes implement with `async def` and `await`
+- Each base class has concrete implementation - no need for child classes to override
+- Use `cast()` to narrow types when delegating to command objects
+- All 4 base classes (NoPoolSyncConfig, NoPoolAsyncConfig, SyncDatabaseConfig, AsyncDatabaseConfig) implement the same way
+
+**Benefits:**
+
+- Single source of truth (protocol) for API contract
+- Each base class provides complete implementation
+- Child adapter classes (AsyncpgConfig, SqliteConfig, etc.) inherit working methods automatically
+- Type checkers understand sync vs async based on `is_async` class variable
+- No code duplication across adapters
+
+**When to use:**
+
+- Adding convenience methods that delegate to external command objects
+- Methods that need identical behavior across all adapters
+- Operations that differ only in sync vs async execution
+- Any protocol method where behavior is determined by sync/async mode
+
+**Anti-patterns to avoid:**
+
+- Don't use runtime `if self.is_async:` checks in a single implementation
+- Don't make protocol methods concrete (always use `@abstractmethod`)
+- Don't duplicate logic across the 4 base classes
+- Don't forget to update all 4 base classes when adding new methods
 
 ### Database Connection Flow
 

docs/changelog.rst

@@ -10,6 +10,72 @@ SQLSpec Changelog
 Recent Updates
 ==============
 
+Migration Convenience Methods on Config Classes
+------------------------------------------------
+
+Added migration methods directly to database configuration classes, eliminating the need to instantiate separate command objects.
+
+**What's New:**
+
+All database configs (both sync and async) now provide migration methods:
+
+- ``migrate_up()`` / ``upgrade()`` - Apply migrations up to a revision
+- ``migrate_down()`` / ``downgrade()`` - Rollback migrations
+- ``get_current_migration()`` - Check current version
+- ``create_migration()`` - Create new migration file
+- ``init_migrations()`` - Initialize migrations directory
+- ``stamp_migration()`` - Stamp database to specific revision
+- ``fix_migrations()`` - Convert timestamp to sequential migrations
+
+**Before (verbose):**
+
+.. code-block:: python
+
+   from sqlspec.adapters.asyncpg import AsyncpgConfig
+   from sqlspec.migrations.commands import AsyncMigrationCommands
+
+   config = AsyncpgConfig(
+       pool_config={"dsn": "postgresql://..."},
+       migration_config={"script_location": "migrations"}
+   )
+
+   commands = AsyncMigrationCommands(config)
+   await commands.upgrade("head")
+
+**After (recommended):**
+
+.. code-block:: python
+
+   from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+   config = AsyncpgConfig(
+       pool_config={"dsn": "postgresql://..."},
+       migration_config={"script_location": "migrations"}
+   )
+
+   await config.upgrade("head")
+
+**Key Benefits:**
+
+- Simpler API - no need to import and instantiate command classes
+- Works with both sync and async adapters
+- Full backward compatibility - command classes still available
+- Cleaner test fixtures and deployment scripts
+
+**Async Adapters** (AsyncPG, Asyncmy, Aiosqlite, Psqlpy):
+
+.. code-block:: python
+
+   await config.migrate_up("head")
+   await config.create_migration("add users")
+
+**Sync Adapters** (SQLite, DuckDB):
+
+.. code-block:: python
+
+   config.migrate_up("head")  # No await needed
+   config.create_migration("add users")
+
 SQL Loader Graceful Error Handling
 -----------------------------------
 

docs/guides/migrations/hybrid-versioning.md

@@ -272,6 +272,92 @@ Always preview before applying:
 sqlspec --config myapp.config fix --dry-run
 ```
 
+## Programmatic API
+
+For Python-based migration automation, use the config method directly instead of CLI commands:
+
+### Async Configuration
+
+```python
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+config = AsyncpgConfig(
+    pool_config={"dsn": "postgresql://user:pass@localhost/mydb"},
+    migration_config={
+        "enabled": True,
+        "script_location": "migrations",
+    }
+)
+
+# Preview conversions
+await config.fix_migrations(dry_run=True)
+
+# Apply conversions (auto-approve)
+await config.fix_migrations(dry_run=False, update_database=True, yes=True)
+
+# Files only (skip database update)
+await config.fix_migrations(dry_run=False, update_database=False, yes=True)
+```
+
+### Sync Configuration
+
+```python
+from sqlspec.adapters.sqlite import SqliteConfig
+
+config = SqliteConfig(
+    pool_config={"database": "myapp.db"},
+    migration_config={
+        "enabled": True,
+        "script_location": "migrations",
+    }
+)
+
+# Preview conversions (no await needed)
+config.fix_migrations(dry_run=True)
+
+# Apply conversions (auto-approve)
+config.fix_migrations(dry_run=False, update_database=True, yes=True)
+
+# Files only (skip database update)
+config.fix_migrations(dry_run=False, update_database=False, yes=True)
+```
+
+### Use Cases
+
+The programmatic API is useful for:
+
+- **Custom deployment scripts** - Integrate migration fixing into deployment automation
+- **Testing workflows** - Automate migration testing in CI/CD pipelines
+- **Framework integrations** - Build migration support into web framework startup hooks
+- **Monitoring tools** - Track migration conversions programmatically
+
+### Example: Custom Deployment Script
+
+```python
+import asyncio
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+async def deploy():
+    config = AsyncpgConfig(
+        pool_config={"dsn": "postgresql://..."},
+        migration_config={"script_location": "migrations"}
+    )
+
+    # Step 1: Convert migrations to sequential
+    print("Converting migrations to sequential format...")
+    await config.fix_migrations(dry_run=False, update_database=True, yes=True)
+
+    # Step 2: Apply all pending migrations
+    print("Applying migrations...")
+    await config.upgrade("head")
+
+    # Step 3: Verify current version
+    current = await config.get_current_migration(verbose=True)
+    print(f"Deployed to version: {current}")
+
+asyncio.run(deploy())
+```
+
 ## Best Practices
 
 ### 1. Always Use Version Control