Migrations API

The Manticore CockroachDB client provides both synchronous and asynchronous migration support to help manage database schema changes.

Synchronous Migration

Migration

Database migration.

Source code in manticore_cockroachdb/migration.py

class Migration:
    """Database migration."""

    def __init__(
        self,
        version: int,
        description: str,
        up_sql: str,
        down_sql: Optional[str] = None
    ):
        """Initialize migration.

        Args:
            version: Migration version number
            description: Migration description
            up_sql: SQL to apply migration
            down_sql: SQL to revert migration
        """
        self.version = version
        self.description = description
        self.up_sql = up_sql
        self.down_sql = down_sql

Functions

__init__(version, description, up_sql, down_sql=None)

Initialize migration.

Parameters:

Name	Type	Description	Default
version	int	Migration version number	required
description	str	Migration description	required
up_sql	str	SQL to apply migration	required
down_sql	Optional[str]	SQL to revert migration	None

Source code in manticore_cockroachdb/migration.py

def __init__(
    self,
    version: int,
    description: str,
    up_sql: str,
    down_sql: Optional[str] = None
):
    """Initialize migration.

    Args:
        version: Migration version number
        description: Migration description
        up_sql: SQL to apply migration
        down_sql: SQL to revert migration
    """
    self.version = version
    self.description = description
    self.up_sql = up_sql
    self.down_sql = down_sql

Asynchronous Migration

AsyncMigration

Asynchronous database migration.

Source code in manticore_cockroachdb/async_migration.py

class AsyncMigration:
    """Asynchronous database migration."""

    def __init__(
        self,
        version: int,
        description: str,
        up_sql: str,
        down_sql: Optional[str] = None
    ):
        """Initialize migration.

        Args:
            version: Migration version number
            description: Migration description
            up_sql: SQL to apply migration
            down_sql: SQL to revert migration
        """
        self.version = version
        self.description = description
        self.up_sql = up_sql
        self.down_sql = down_sql

Functions

__init__(version, description, up_sql, down_sql=None)

Initialize migration.

Parameters:

Name	Type	Description	Default
version	int	Migration version number	required
description	str	Migration description	required
up_sql	str	SQL to apply migration	required
down_sql	Optional[str]	SQL to revert migration	None

Source code in manticore_cockroachdb/async_migration.py

def __init__(
    self,
    version: int,
    description: str,
    up_sql: str,
    down_sql: Optional[str] = None
):
    """Initialize migration.

    Args:
        version: Migration version number
        description: Migration description
        up_sql: SQL to apply migration
        down_sql: SQL to revert migration
    """
    self.version = version
    self.description = description
    self.up_sql = up_sql
    self.down_sql = down_sql

Usage Examples

Synchronous Migrations

from manticore_cockroachdb import Database, Migration

# Connect to database
db = Database(database="example_db")

# Create migration instance
migration = Migration(db, migrations_dir="./migrations")

# Create a new migration
migration.create_migration(
    "create users table",
    """
    CREATE TABLE users (
        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        created_at TIMESTAMPTZ DEFAULT now()
    );
    """,
    """
    DROP TABLE users;
    """
)

# Load migrations
migrations = migration.load_migrations()
print(f"Loaded {len(migrations)} migrations")

# Apply migrations
applied = migration.migrate()
print(f"Applied {applied} migrations")

# Rollback last migration
rollback_count = migration.rollback(count=1)
print(f"Rolled back {rollback_count} migrations")

Asynchronous Migrations

import asyncio
from manticore_cockroachdb import AsyncDatabase, AsyncMigration

async def main():
    # Connect to database
    db = AsyncDatabase(database="example_db")
    await db.connect()

    try:
        # Create migration instance
        migration = AsyncMigration(db, migrations_dir="./async_migrations")

        # Create a new migration
        await migration.create_migration(
            "create async users table",
            """
            CREATE TABLE async_users (
                id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                name TEXT NOT NULL,
                email TEXT UNIQUE NOT NULL,
                created_at TIMESTAMPTZ DEFAULT now()
            );
            """,
            """
            DROP TABLE async_users;
            """
        )

        # Load migrations
        migrations = await migration.load_migrations()
        print(f"Loaded {len(migrations)} migrations")

        # Apply migrations
        applied = await migration.migrate()
        print(f"Applied {applied} migrations")

        # Rollback last migration
        rollback_count = await migration.rollback(count=1)
        print(f"Rolled back {rollback_count} migrations")

    finally:
        await db.close()

# Run the async function
asyncio.run(main())