Files
gh-christianlouis-leagueledger/app/db_migrations.py
T
Christian Krakau-Louis 6c260c5936 Add account deletion confirmation page, privacy settings template, and profile view template
- Created a new HTML template for account deletion confirmation with user-friendly messaging and a return link.
- Developed a privacy settings page allowing users to control visibility of their personal information with appropriate messaging for success and error states.
- Implemented a profile view template displaying user information, statistics, teams, and achievements based on user permissions.
- Added responsive design elements and improved user interface with Tailwind CSS classes for better aesthetics and usability.
2025-04-16 21:45:33 +02:00

242 lines
9.3 KiB
Python

from sqlalchemy import text, inspect, Column, String, JSON, MetaData, Table
from .db import engine
def table_exists(conn, table_name):
"""Check if a table exists in the database."""
result = conn.execute(text(f"""
SELECT COUNT(*) as count
FROM information_schema.tables
WHERE table_schema = DATABASE()
AND table_name = '{table_name}'
"""))
return result.scalar() > 0
def column_exists(conn, table_name, column_name):
"""Check if a column exists in a table."""
result = conn.execute(text(f"""
SELECT COUNT(*) as count
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = '{table_name}'
AND column_name = '{column_name}'
"""))
return result.scalar() > 0
def apply_migrations():
"""Apply all pending database migrations."""
# Check and add OAuth columns to users table
try:
with engine.connect() as conn:
# Make sure tables exist before trying to alter them
if not table_exists(conn, 'users'):
print("Users table doesn't exist yet. Skipping OAuth columns migration.")
return
# Check if the OAuth columns exist
if not column_exists(conn, 'users', 'is_oauth_user'):
print("Adding OAuth columns to users table...")
# Add the OAuth columns
conn.execute(text("""
ALTER TABLE users
ADD COLUMN is_oauth_user BOOLEAN DEFAULT FALSE,
ADD COLUMN oauth_id VARCHAR(255) NULL,
ADD COLUMN oauth_provider VARCHAR(50) NULL,
ADD COLUMN picture VARCHAR(255) NULL
"""))
conn.commit()
print("OAuth columns added successfully.")
else:
print("OAuth columns already exist in users table.")
# Check if the is_admin column exists in the users table
if not column_exists(conn, 'users', 'is_admin'):
print("Adding is_admin column to users table...")
# Add the is_admin column to users table
conn.execute(text("""
ALTER TABLE users
ADD COLUMN is_admin BOOLEAN DEFAULT FALSE
"""))
conn.commit()
print("is_admin column added successfully to users table.")
else:
print("is_admin column already exists in users table.")
# Only proceed with teams table if it exists
if table_exists(conn, 'teams'):
# Check if the owner_id column exists in the teams table
if not column_exists(conn, 'teams', 'owner_id'):
print("Adding owner_id column to teams table...")
# Add the owner_id column to teams table
conn.execute(text("""
ALTER TABLE teams
ADD COLUMN owner_id INT NULL,
ADD CONSTRAINT fk_teams_owner
FOREIGN KEY (owner_id) REFERENCES users(id)
ON DELETE SET NULL
"""))
conn.commit()
print("owner_id column added successfully to teams table.")
else:
print("owner_id column already exists in teams table.")
else:
print("Teams table doesn't exist yet. Skipping teams migrations.")
except Exception as e:
print(f"Error applying migrations: {str(e)}")
def run_migrations(engine):
"""
Run database migrations that can't be handled by SQLAlchemy's create_all()
"""
# Create a MetaData object
metadata = MetaData()
metadata.bind = engine
connection = engine.connect()
try:
print("Running migrations...")
# Check if the columns already exist before adding them
# Add additional_oauth_providers column if it doesn't exist
add_oauth_providers_column(connection)
# Add first_name and last_name columns if they don't exist
add_name_columns(connection)
# Add privacy_settings column if it doesn't exist
add_privacy_settings_column(connection)
# Add picture_manually_deleted column if it doesn't exist
add_picture_manually_deleted_column(connection)
print("Migrations completed successfully")
except Exception as e:
print(f"Error during migrations: {str(e)}")
finally:
connection.close()
def add_oauth_providers_column(connection):
"""Add additional_oauth_providers column to users table"""
try:
# Use database-agnostic way to check if column exists
inspector = inspect(engine)
columns = [col['name'] for col in inspector.get_columns('users')]
if 'additional_oauth_providers' not in columns:
print("Adding additional_oauth_providers column to users table")
# Add column with database-specific syntax
if engine.name == 'sqlite':
connection.execute(text("""
ALTER TABLE users
ADD COLUMN additional_oauth_providers JSON
"""))
else: # MySQL
connection.execute(text("""
ALTER TABLE users
ADD COLUMN additional_oauth_providers JSON NULL
"""))
connection.commit()
else:
print("Column additional_oauth_providers already exists")
except Exception as e:
print(f"Error adding additional_oauth_providers column: {str(e)}")
def add_name_columns(connection):
"""Add first_name and last_name columns to users table"""
try:
# Use database-agnostic way to check if columns exist
inspector = inspect(engine)
columns = [col['name'] for col in inspector.get_columns('users')]
# Add first_name if needed
if 'first_name' not in columns:
print("Adding first_name column to users table")
connection.execute(text("""
ALTER TABLE users
ADD COLUMN first_name VARCHAR(50) NULL
"""))
connection.commit()
else:
print("Column first_name already exists")
# Add last_name if needed
if 'last_name' not in columns:
print("Adding last_name column to users table")
connection.execute(text("""
ALTER TABLE users
ADD COLUMN last_name VARCHAR(50) NULL
"""))
connection.commit()
else:
print("Column last_name already exists")
except Exception as e:
print(f"Error adding name columns: {str(e)}")
def add_privacy_settings_column(connection):
"""Add privacy_settings column to users table"""
try:
# Use database-agnostic way to check if column exists
inspector = inspect(engine)
columns = [col['name'] for col in inspector.get_columns('users')]
if 'privacy_settings' not in columns:
print("Adding privacy_settings column to users table")
# Add column with database-specific syntax
if engine.name == 'sqlite':
connection.execute(text("""
ALTER TABLE users
ADD COLUMN privacy_settings JSON
"""))
else: # MySQL
connection.execute(text("""
ALTER TABLE users
ADD COLUMN privacy_settings JSON NULL
"""))
connection.commit()
print("Successfully added privacy_settings column to users table")
else:
print("Column privacy_settings already exists")
except Exception as e:
print(f"Error adding privacy_settings column: {str(e)}")
def add_picture_manually_deleted_column(connection):
"""Add picture_manually_deleted column to users table"""
try:
# Use database-agnostic way to check if column exists
inspector = inspect(engine)
columns = [col['name'] for col in inspector.get_columns('users')]
if 'picture_manually_deleted' not in columns:
print("Adding picture_manually_deleted column to users table")
# Add column with database-specific syntax
if engine.name == 'sqlite':
connection.execute(text("""
ALTER TABLE users
ADD COLUMN picture_manually_deleted BOOLEAN DEFAULT FALSE
"""))
else: # MySQL
connection.execute(text("""
ALTER TABLE users
ADD COLUMN picture_manually_deleted BOOLEAN DEFAULT FALSE
"""))
connection.commit()
print("Successfully added picture_manually_deleted column to users table")
else:
print("Column picture_manually_deleted already exists")
except Exception as e:
print(f"Error adding picture_manually_deleted column: {str(e)}")