63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
import datetime
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from sqlite3 import Connection
|
|
|
|
CREATE_MIGRATIONS_TABLE = """
|
|
CREATE TABLE IF NOT EXISTS migrations (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
file TEXT,
|
|
created_at TEXT
|
|
);
|
|
"""
|
|
|
|
def init_db():
|
|
conn = sqlite3.connect("db.sqlite")
|
|
|
|
# Create migration table, if not exists yet
|
|
conn.execute(CREATE_MIGRATIONS_TABLE)
|
|
conn.commit()
|
|
|
|
def create_migration(conn: Connection, migration: str):
|
|
conn.execute(f"""
|
|
INSERT INTO migrations (file, created_at) VALUES ('{migration}', '{datetime.datetime.now()}');
|
|
""")
|
|
|
|
"""
|
|
Returns True, if the migration has already been migrates once.
|
|
"""
|
|
def check_migration(conn: Connection, migration: str):
|
|
res = conn.cursor().execute(f"""
|
|
SELECT EXISTS(SELECT file FROM migrations WHERE file = '{migration}');
|
|
""")
|
|
return res.fetchone()[0] == 1
|
|
|
|
if __name__ == "__main__":
|
|
# Init DB with migrations table
|
|
init_db()
|
|
|
|
migrations_path = os.path.join(os.path.dirname(__file__), 'migrations')
|
|
|
|
try:
|
|
if not os.path.isdir(migrations_path):
|
|
raise FileNotFoundError
|
|
|
|
conn = sqlite3.connect("db.sqlite")
|
|
for file in sorted(os.listdir(migrations_path)):
|
|
if not file.endswith('.sql'):
|
|
continue
|
|
with open(os.path.join(os.path.dirname(__file__), 'migrations', file)) as migration:
|
|
migration_name = Path(file).stem
|
|
|
|
if check_migration(conn, migration_name):
|
|
continue
|
|
|
|
print(f"Migrating {migration_name}...")
|
|
create_migration(conn, migration_name)
|
|
conn.execute(migration.read())
|
|
print(f"Migrated {migration_name}\n")
|
|
|
|
conn.commit()
|
|
except FileNotFoundError:
|
|
print("migrations/ Directive could not be found within the db directory, contained in the project main directory.") |