Veritabanı Migrasyon Betikleri: Versionlama ve Rollback

1# migrations/001_create_users.sql
2CREATE TABLE users (
3id SERIAL PRIMARY KEY,
4email VARCHAR(255) UNIQUE NOT NULL,
5created_at TIMESTAMP DEFAULT NOW()
6);
7
8# migrations/002_add_username.sql
9ALTER TABLE users ADD COLUMN username VARCHAR(100);
10
11# migrations/003_create_posts.sql
12CREATE TABLE posts (
13id SERIAL PRIMARY KEY,
14user_id INTEGER REFERENCES users(id),
15title VARCHAR(255),
16created_at TIMESTAMP DEFAULT NOW()
17);

1import sqlite3
2import os
3from datetime import datetime
4
5class MigrationManager:
6def __init__(self, db_path, migrations_dir):
7self.db_path = db_path
8self.migrations_dir = migrations_dir
9self.init_db()
10
11def init_db(self):
12conn = sqlite3.connect(self.db_path)
13cursor = conn.cursor()
14cursor.execute('''
15CREATE TABLE IF NOT EXISTS migrations (
16version TEXT PRIMARY KEY,
17executed_at TIMESTAMP,
18status TEXT
19)
20''')
21conn.commit()
22conn.close()
23
24def get_applied_migrations(self):
25conn = sqlite3.connect(self.db_path)
26cursor = conn.cursor()
27cursor.execute('SELECT version FROM migrations WHERE status = "success"')
28result = [row[0] for row in cursor.fetchall()]
29conn.close()
30return result
31
32def migrate_up(self):
33applied = self.get_applied_migrations()
34migration_files = sorted(os.listdir(self.migrations_dir))
35
36for file in migration_files:
37version = file.split('_')[0]
38if version in applied:
39continue
40
41filepath = os.path.join(self.migrations_dir, file)
42try:
43with open(filepath, 'r') as f:
44sql = f.read()
45
46conn = sqlite3.connect(self.db_path)
47cursor = conn.cursor()
48cursor.executescript(sql)
49
50cursor.execute(
51'INSERT INTO migrations VALUES (?,?,?)',
52(version, datetime.now(), 'success')
53)
54conn.commit()
55conn.close()
56print(f" {version} migrated successfully")
57except Exception as e:
58print(f" {version} failed: {e}")
59conn.execute(
60'INSERT INTO migrations VALUES (?,?,?)',
61(version, datetime.now(), 'failed')
62)
63conn.commit()
64conn.close()
65break
66
67def rollback(self, steps=1):
68conn = sqlite3.connect(self.db_path)
69cursor = conn.cursor()
70cursor.execute(
71'SELECT version FROM migrations WHERE status = "success" ORDER BY version DESC LIMIT?',
72(steps,)
73)
74versions = [row[0] for row in cursor.fetchall()]
75
76for version in versions:
77rollback_file = os.path.join(self.migrations_dir, f"{version}_rollback.sql")
78if os.path.exists(rollback_file):
79with open(rollback_file, 'r') as f:
80sql = f.read()
81cursor.executescript(sql)
82cursor.execute('DELETE FROM migrations WHERE version =?', (version,))
83print(f"↻ {version} rolled back")
84
85conn.commit()
86conn.close()
87
88# Kullanım
89if __name__ == '__main__':
90manager = MigrationManager('app.db', 'migrations')
91manager.migrate_up()
92# manager.rollback(1) # Son 1 migrationu geri al

1001 migrated successfully
2002 migrated successfully
3003 migrated successfully