332 lines
12 KiB
Python
332 lines
12 KiB
Python
# encoding=utf-8
|
|
"""
|
|
Database configuration and models for multi-database support using SQLAlchemy.
|
|
Supports PostgreSQL, MariaDB/MySQL, and SQLite.
|
|
"""
|
|
|
|
import os
|
|
from sqlalchemy import create_engine, Column, Integer, String, Date, DateTime, Text, SmallInteger, ForeignKey, Boolean, Float
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, relationship
|
|
from sqlalchemy.dialects.mysql import LONGTEXT
|
|
from sqlalchemy.dialects.postgresql import JSON
|
|
from datetime import datetime
|
|
|
|
# Base class for all models
|
|
Base = declarative_base()
|
|
|
|
class DatabaseConfig:
|
|
"""Database configuration class for multiple database support."""
|
|
|
|
def __init__(self):
|
|
self.database_url = self._get_database_url()
|
|
self.engine = create_engine(self.database_url, echo=False)
|
|
self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
|
|
|
|
def _get_database_url(self):
|
|
"""Get database URL from environment variables or configuration."""
|
|
db_type = os.getenv('DB_TYPE', os.getenv('DATABASE_TYPE', 'sqlite')).lower()
|
|
|
|
if db_type == 'postgresql':
|
|
return self._get_postgresql_url()
|
|
elif db_type in ['mysql', 'mariadb']:
|
|
return self._get_mysql_url()
|
|
elif db_type == 'sqlite':
|
|
return self._get_sqlite_url()
|
|
else:
|
|
raise ValueError(f"Unsupported database type: {db_type}")
|
|
|
|
def _get_postgresql_url(self):
|
|
"""Get PostgreSQL connection URL."""
|
|
host = os.getenv('DB_HOST', os.getenv('POSTGRES_HOST', 'localhost'))
|
|
port = os.getenv('DB_PORT', os.getenv('POSTGRES_PORT', '5432'))
|
|
database = os.getenv('DB_NAME', os.getenv('POSTGRES_DATABASE', 'hockey_results'))
|
|
username = os.getenv('DB_USER', os.getenv('POSTGRES_USER', 'postgres'))
|
|
password = os.getenv('DB_PASSWORD', os.getenv('POSTGRES_PASSWORD', ''))
|
|
|
|
return f"postgresql://{username}:{password}@{host}:{port}/{database}"
|
|
|
|
def _get_mysql_url(self):
|
|
"""Get MySQL/MariaDB connection URL."""
|
|
host = os.getenv('DB_HOST', os.getenv('MYSQL_HOST', 'localhost'))
|
|
port = os.getenv('DB_PORT', os.getenv('MYSQL_PORT', '3306'))
|
|
database = os.getenv('DB_NAME', os.getenv('MYSQL_DATABASE', 'hockey_results'))
|
|
username = os.getenv('DB_USER', os.getenv('MYSQL_USER', 'root'))
|
|
password = os.getenv('DB_PASSWORD', os.getenv('MYSQL_PASSWORD', ''))
|
|
charset = os.getenv('MYSQL_CHARSET', 'utf8mb4')
|
|
|
|
return f"mysql+pymysql://{username}:{password}@{host}:{port}/{database}?charset={charset}"
|
|
|
|
def _get_sqlite_url(self):
|
|
"""Get SQLite connection URL."""
|
|
database_path = os.getenv('DB_NAME', os.getenv('SQLITE_DATABASE_PATH', 'hockey_results.db'))
|
|
return f"sqlite:///{database_path}"
|
|
|
|
def get_session(self):
|
|
"""Get database session."""
|
|
return self.SessionLocal()
|
|
|
|
def create_tables(self):
|
|
"""Create all database tables."""
|
|
Base.metadata.create_all(bind=self.engine)
|
|
|
|
# Global database configuration instance
|
|
db_config = DatabaseConfig()
|
|
|
|
# Database Models
|
|
class Player(Base):
|
|
"""Player model."""
|
|
__tablename__ = 'players'
|
|
|
|
player_number = Column(Integer, primary_key=True)
|
|
player_forenames = Column(String(50))
|
|
player_surname = Column(String(30))
|
|
player_nickname = Column(String(30))
|
|
player_chinese_name = Column(String(10))
|
|
player_email = Column(String(255))
|
|
player_dob = Column(Date)
|
|
player_hkid = Column(String(20))
|
|
player_tel_number = Column(String(30))
|
|
player_team = Column(String(6))
|
|
player_picture_url = Column(String(255))
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
class Club(Base):
|
|
"""Club model."""
|
|
__tablename__ = 'clubs'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
hockey_club = Column(String(100), unique=True, nullable=False)
|
|
logo_url = Column(String(255))
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class Team(Base):
|
|
"""Team model."""
|
|
__tablename__ = 'teams'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
club = Column(String(100), ForeignKey('clubs.hockey_club'))
|
|
team = Column(String(10))
|
|
display_name = Column(String(100))
|
|
league = Column(String(50))
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class MatchSquad(Base):
|
|
"""Match squad model."""
|
|
__tablename__ = 'match_squad'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
player_number = Column(Integer, ForeignKey('players.player_number'))
|
|
player_forenames = Column(String(50))
|
|
player_surname = Column(String(30))
|
|
player_nickname = Column(String(30))
|
|
match_date = Column(Date)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class SquadHistory(Base):
|
|
"""Historical match squad records."""
|
|
__tablename__ = 'squad_history'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
player_number = Column(Integer, ForeignKey('players.player_number'))
|
|
player_forenames = Column(String(50))
|
|
player_surname = Column(String(30))
|
|
player_nickname = Column(String(30))
|
|
match_date = Column(Date)
|
|
fixture_number = Column(String(20))
|
|
archived_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class HockeyFixture(Base):
|
|
"""Hockey fixture model."""
|
|
__tablename__ = 'hockey_fixtures'
|
|
|
|
fixture_number = Column(Integer, primary_key=True)
|
|
date = Column(Date)
|
|
time = Column(String(10))
|
|
home_team = Column(String(100))
|
|
away_team = Column(String(100))
|
|
venue = Column(String(255))
|
|
home_score = Column(Integer)
|
|
away_score = Column(Integer)
|
|
umpire1 = Column(String(100))
|
|
umpire2 = Column(String(100))
|
|
match_official = Column(String(100))
|
|
division = Column(String(50))
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class AdminSettings(Base):
|
|
"""Admin settings model."""
|
|
__tablename__ = 'admin_settings'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
userid = Column(String(50), default='admin')
|
|
next_fixture = Column(Integer)
|
|
next_club = Column(String(100))
|
|
next_team = Column(String(100))
|
|
next_date = Column(Date)
|
|
curr_motm = Column(Integer, ForeignKey('players.player_number'))
|
|
curr_dotd = Column(Integer, ForeignKey('players.player_number'))
|
|
oppo_logo = Column(String(255))
|
|
hkfc_logo = Column(String(255))
|
|
motm_url_suffix = Column(String(50))
|
|
prev_fixture = Column(Integer)
|
|
voting_deadline = Column(DateTime)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
class MotmVote(Base):
|
|
"""MOTM/DotD voting model."""
|
|
__tablename__ = 'motm_votes'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
player_number = Column(Integer, ForeignKey('players.player_number'))
|
|
player_name = Column(String(100))
|
|
motm_total = Column(Integer, default=0)
|
|
dotd_total = Column(Integer, default=0)
|
|
goals_total = Column(Integer, default=0)
|
|
assists_total = Column(Integer, default=0)
|
|
fixture_number = Column(Integer)
|
|
motm_votes = Column(Integer, default=0)
|
|
dotd_votes = Column(Integer, default=0)
|
|
goals = Column(Integer, default=0)
|
|
assists = Column(Integer, default=0)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
class MatchComment(Base):
|
|
"""Match comments model."""
|
|
__tablename__ = 'match_comments'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
match_date = Column(Date)
|
|
opposition = Column(String(100))
|
|
comment = Column(Text)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
class HockeyUser(Base):
|
|
"""User authentication model."""
|
|
__tablename__ = 'hockey_users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
username = Column(String(50), unique=True, nullable=False)
|
|
email = Column(String(255), unique=True, nullable=False)
|
|
password = Column(String(255), nullable=False)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
last_login = Column(DateTime)
|
|
|
|
class S3Settings(Base):
|
|
"""S3/MinIO storage configuration model.
|
|
|
|
Note: Credentials (access_key_id, secret_access_key) are NEVER stored in the database.
|
|
They must be provided via environment variables for security.
|
|
"""
|
|
__tablename__ = 's3_settings'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
userid = Column(String(50), default='admin')
|
|
enabled = Column(Boolean, default=False)
|
|
storage_provider = Column(String(20), default='aws') # 'aws' or 'minio'
|
|
endpoint = Column(String(255), default='') # MinIO endpoint or custom S3 endpoint
|
|
region = Column(String(50), default='us-east-1')
|
|
bucket_name = Column(String(255), default='')
|
|
bucket_prefix = Column(String(255), default='assets/')
|
|
use_signed_urls = Column(Boolean, default=True)
|
|
signed_url_expiry = Column(Integer, default=3600) # seconds
|
|
fallback_to_static = Column(Boolean, default=True)
|
|
use_ssl = Column(Boolean, default=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
|
|
# Database utility functions
|
|
def get_db_session():
|
|
"""Get database session."""
|
|
return db_config.get_session()
|
|
|
|
def execute_sql(sql_command, params=None):
|
|
"""Execute SQL command with parameters."""
|
|
session = get_db_session()
|
|
try:
|
|
if params:
|
|
result = session.execute(sql_command, params)
|
|
else:
|
|
result = session.execute(sql_command)
|
|
session.commit()
|
|
return result
|
|
except Exception as e:
|
|
session.rollback()
|
|
print(f"SQL Error: {e}")
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
def fetch_all(sql_command, params=None):
|
|
"""Fetch all results from SQL query."""
|
|
session = get_db_session()
|
|
try:
|
|
if params:
|
|
result = session.execute(sql_command, params)
|
|
else:
|
|
result = session.execute(sql_command)
|
|
rows = result.fetchall()
|
|
# Convert to list of dictionaries for compatibility
|
|
return [dict(row._mapping) for row in rows] if rows else []
|
|
except Exception as e:
|
|
print(f"SQL Error: {e}")
|
|
return []
|
|
finally:
|
|
session.close()
|
|
|
|
def fetch_one(sql_command, params=None):
|
|
"""Fetch one result from SQL query."""
|
|
session = get_db_session()
|
|
try:
|
|
if params:
|
|
result = session.execute(sql_command, params)
|
|
else:
|
|
result = session.execute(sql_command)
|
|
row = result.fetchone()
|
|
return dict(row._mapping) if row else None
|
|
except Exception as e:
|
|
print(f"SQL Error: {e}")
|
|
return None
|
|
finally:
|
|
session.close()
|
|
|
|
# Legacy compatibility functions
|
|
def sql_write(sql_cmd, params=None):
|
|
"""Legacy compatibility function for sql_write."""
|
|
try:
|
|
execute_sql(sql_cmd, params)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Write Error: {e}")
|
|
return False
|
|
|
|
def sql_write_static(sql_cmd, params=None):
|
|
"""Legacy compatibility function for sql_write_static."""
|
|
return sql_write(sql_cmd, params)
|
|
|
|
def sql_read(sql_cmd, params=None):
|
|
"""Legacy compatibility function for sql_read."""
|
|
try:
|
|
result = fetch_all(sql_cmd, params)
|
|
return result
|
|
except Exception as e:
|
|
print(f"Read Error: {e}")
|
|
return []
|
|
|
|
def sql_read_static(sql_cmd, params=None):
|
|
"""Legacy compatibility function for sql_read_static."""
|
|
return sql_read(sql_cmd, params)
|
|
|
|
# Initialize database tables
|
|
def init_database():
|
|
"""Initialize database tables."""
|
|
try:
|
|
db_config.create_tables()
|
|
print("Database tables created successfully")
|
|
except Exception as e:
|
|
print(f"Database initialization error: {e}")
|
|
raise
|