Migrations

from framefox.core.orm.migration.alembic_manager import AlembicManager

manager = AlembicManager()

# Create migrations directory structure
manager.setup_directories()

# Copy template files
manager.setup_templates()

manager = AlembicManager()

# Create migration with auto-detection
migration_file = manager.create_migration(
    message="Add user table",
    autogenerate=True
)

if migration_file:
    print(f"Created migration: {migration_file}")

migration_file = manager.create_migration(
    message="Custom migration",
    autogenerate=False
)

from framefox.core.orm.abstract_entity import AbstractEntity
from sqlmodel import Field

class User(AbstractEntity, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)  # Added unique constraint
    age: int  # New field

manager.create_migration("Add age field and email unique constraint")

"""Add age field and email unique constraint

Revision ID: abc123
Revises: def456
Create Date: 2024-03-05 10:30:00
"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    # Add age column
    op.add_column('user', sa.Column('age', sa.Integer(), nullable=False))
    
    # Add unique constraint to email
    op.create_unique_constraint('uq_user_email', 'user', ['email'])

def downgrade():
    # Remove unique constraint
    op.drop_constraint('uq_user_email', 'user', type_='unique')
    
    # Remove age column
    op.drop_column('user', 'age')

manager = AlembicManager()

success, migrations_applied = manager.upgrade()

if success:
    if migrations_applied:
        print("Migrations applied successfully")
    else:
        print("Database is up to date")
else:
    print("Migration failed")

# Upgrade to specific revision
manager.upgrade(revision="abc123")

# Upgrade one version forward
manager.upgrade(revision="+1")

# Rollback one version
success = manager.downgrade("-1")

# Rollback to specific revision
success = manager.downgrade("abc123")

# Rollback all migrations
success = manager.downgrade("base")

if success:
    print("Rollback successful")
else:
    print("Rollback failed")

from sqlalchemy import create_engine, text

engine = create_engine(manager.get_database_url_string())
with engine.connect() as conn:
    current = conn.execute(text("SELECT version_num FROM alembic_version")).scalar()
    print(f"Current version: {current}")

migrations = manager.get_existing_migrations()
for migration in migrations:
    print(migration)

content = manager.get_migration_content("abc123_add_user_table.py")
print(content)

content = manager.get_migration_content("abc123_add_user_table.py")
has_changes = manager.has_changes(content)

if has_changes:
    print("Migration contains changes")
else:
    print("Empty migration (no changes detected)")

success = manager.delete_migration("abc123_add_user_table.py")

if success:
    print("Migration deleted")
else:
    print("Migration not found")

def get_database_url_string(self) -> str:
    settings = Settings()
    db_config = settings.database_url
    
    if isinstance(db_config, str):
        return db_config
    
    # Build URL from config object
    if db_config.driver == "sqlite":
        return f"sqlite:///{db_config.database}"
    
    dialect = "mysql+pymysql" if db_config.driver == "mysql" else db_config.driver
    return f"{dialect}://{username}:{password}@{host}:{port}/{database}"

# SQLite
"sqlite:///./database.db"

# MySQL
"mysql+pymysql://user:password@localhost:3306/mydb"

# PostgreSQL
"postgresql://user:password@localhost:5432/mydb"

success = manager.create_alembic_version_table()

if success:
    print("Alembic version table created")

success = manager.clear_alembic_version_table()

if success:
    print("Version history cleared")

manager.cleanup_migrations()
print("All migrations cleaned up")

success = manager.clear_all_migrations()

if success:
    print("All migrations and history cleared")

# In migration file
def upgrade():
    op.add_column('user', sa.Column('phone', sa.String(20), nullable=True))

def downgrade():
    op.drop_column('user', 'phone')

def upgrade():
    op.drop_column('user', 'old_field')

def downgrade():
    op.add_column('user', sa.Column('old_field', sa.String(100)))

def upgrade():
    op.alter_column('user', 'old_name', new_column_name='new_name')

def downgrade():
    op.alter_column('user', 'new_name', new_column_name='old_name')

def upgrade():
    op.create_index('idx_user_email', 'user', ['email'])

def downgrade():
    op.drop_index('idx_user_email', 'user')

def upgrade():
    op.create_foreign_key(
        'fk_post_user_id',
        'post', 'user',
        ['user_id'], ['id']
    )

def downgrade():
    op.drop_constraint('fk_post_user_id', 'post', type_='foreignkey')

def upgrade():
    op.create_table(
        'post',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('content', sa.Text(), nullable=False),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(['user_id'], ['user.id'])
    )

def downgrade():
    op.drop_table('post')

from framefox.core.orm.migration.alembic_manager import AlembicManager
import click

@click.group()
def migrate():
    """Database migration commands"""
    pass

@migrate.command()
@click.argument('message')
def create(message):
    """Create a new migration"""
    manager = AlembicManager()
    file = manager.create_migration(message)
    click.echo(f"Created: {file}")

@migrate.command()
def upgrade():
    """Apply pending migrations"""
    manager = AlembicManager()
    success, applied = manager.upgrade()
    if success:
        click.echo("Migrations applied" if applied else "Already up to date")
    else:
        click.echo("Migration failed", err=True)

@migrate.command()
@click.argument('revision')
def downgrade(revision):
    """Rollback to a previous migration"""
    manager = AlembicManager()
    if manager.downgrade(revision):
        click.echo("Rollback successful")
    else:
        click.echo("Rollback failed", err=True)

if __name__ == '__main__':
    migrate()