99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""
|
|
Migration script to convert is_admin boolean field to role string field
|
|
"""
|
|
import sqlite3
|
|
import os
|
|
from pathlib import Path
|
|
from sqlalchemy.engine.url import make_url
|
|
|
|
def migrate_users_table():
|
|
"""Add role column to users table and migrate data from is_admin"""
|
|
|
|
# Get database path from environment or use default
|
|
db_url = os.getenv("DATABASE_URL", "sqlite:///./data/drugs.db")
|
|
|
|
# Parse SQLite URL to get the file path
|
|
if db_url.startswith("sqlite:///"):
|
|
db_path = db_url.replace("sqlite:///", "")
|
|
# Handle relative paths
|
|
if not db_path.startswith("/"):
|
|
db_path = Path("/app/data") / "drugs.db"
|
|
else:
|
|
db_path = Path(db_path)
|
|
else:
|
|
print(f"Unsupported database URL: {db_url}")
|
|
return
|
|
|
|
if not db_path.exists():
|
|
print(f"Database does not exist at {db_path}, skipping migration")
|
|
return
|
|
|
|
print(f"Connecting to database at {db_path}")
|
|
conn = sqlite3.connect(str(db_path))
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# Check if role column already exists
|
|
cursor.execute("PRAGMA table_info(users)")
|
|
columns = [col[1] for col in cursor.fetchall()]
|
|
|
|
if "role" in columns:
|
|
print("Role column already exists, skipping migration")
|
|
conn.close()
|
|
return
|
|
|
|
if not columns:
|
|
print("Users table does not exist yet, skipping migration")
|
|
conn.close()
|
|
return
|
|
|
|
print("Migrating users table: adding role column...")
|
|
|
|
# Add role column with default value
|
|
cursor.execute("ALTER TABLE users ADD COLUMN role VARCHAR DEFAULT 'user'")
|
|
|
|
# Migrate data from is_admin to role
|
|
if "is_admin" in columns:
|
|
print("Migrating data from is_admin to role...")
|
|
cursor.execute("""
|
|
UPDATE users
|
|
SET role = CASE
|
|
WHEN is_admin = 1 THEN 'admin'
|
|
ELSE 'user'
|
|
END
|
|
""")
|
|
|
|
# Drop the old is_admin column
|
|
# SQLite doesn't support DROP COLUMN directly in older versions,
|
|
# so we use a workaround
|
|
cursor.execute("ALTER TABLE users RENAME TO users_old")
|
|
cursor.execute("""
|
|
CREATE TABLE users (
|
|
id INTEGER PRIMARY KEY,
|
|
username VARCHAR UNIQUE NOT NULL,
|
|
hashed_password VARCHAR NOT NULL,
|
|
role VARCHAR NOT NULL DEFAULT 'user',
|
|
created_at DATETIME
|
|
)
|
|
""")
|
|
cursor.execute("""
|
|
INSERT INTO users (id, username, hashed_password, role, created_at)
|
|
SELECT id, username, hashed_password, role, created_at FROM users_old
|
|
""")
|
|
cursor.execute("DROP TABLE users_old")
|
|
print("Successfully migrated is_admin to role and cleaned up old column")
|
|
|
|
conn.commit()
|
|
print("Migration completed successfully!")
|
|
|
|
except sqlite3.OperationalError as e:
|
|
print(f"Migration error: {e}")
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
migrate_users_table()
|
|
|