yoyo-migrations

import asyncio import os from pathlib import Path from yoyo import get_migrations from yoyo.connections import connect # Create a temporary directory for migrations TEMP_MIGRATIONS_DIR = Path('./temp_yoyo_migrations_quickstart') TEMP_MIGRATIONS_DIR.mkdir(exist_ok=True) # Create dummy migration files (TEMP_MIGRATIONS_DIR / '0001.create_table.sql').write_text( 'CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT);' ) (TEMP_MIGRATIONS_DIR / '0002.add_email_field.sql').write_text( 'ALTER TABLE users ADD COLUMN email TEXT;' ) # Use an in-memory SQLite database for the example DB_URI = "sqlite:///:memory:" async def run_yoyo_programmatically(): print(f"Connecting to database: {DB_URI}") backend = await connect(DB_URI) try: migrations = get_migrations(TEMP_MIGRATIONS_DIR) print(f"Found {len(migrations)} migrations in {TEMP_MIGRATIONS_DIR}") with backend.transaction(): to_apply = backend.to_apply(migrations) if to_apply: applied_migrations = await backend.apply_migrations(to_apply) print(f"Successfully applied {len(applied_migrations)} migrations.") else: print("No new migrations to apply.") # Example of checking current schema (simplified) cursor = await backend.cursor() await cursor.execute("PRAGMA table_info(users);") schema = await cursor.fetchall() print("Current 'users' table schema:") for col in schema: print(f" - {col[1]} ({col[2]})") # col[1]=name, col[2]=type await cursor.close() finally: await backend.close() # Clean up temporary migration files and directory for f in TEMP_MIGRATIONS_DIR.iterdir(): f.unlink() TEMP_MIGRATIONS_DIR.rmdir() if __name__ == '__main__': asyncio.run(run_yoyo_programmatically())