from __future__ import annotations import os from contextlib import contextmanager from datetime import datetime from pathlib import Path from threading import RLock from typing import Any, Iterator from sqlalchemy import create_engine, inspect, text from sqlalchemy.engine import Engine, make_url from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session, sessionmaker from .models import ( Base, Coach, CoachTournament, Country, Game, GameControlState, GameTeamStatistic, GameOfficial, GameRoster, OperatorSession, PenaltyType, Player, PlayerTournament, PlayerSeasonStatistic, Referee, RefereeTournament, ShootoutAttempt, TournamentStanding, TournamentStandingRow, TournamentStatisticResource, TournamentStatisticRow, TeamSeasonStatistic, UserPreference, VmixAssignment, VmixDevice, VmixMappingProfile, VmixMappingField, MappingContextVariable, MappingContextValue, MappingSqlDataSource, ) class HockeyDatabase: """Remote PostgreSQL database layer with additive schema repair. The hockey application deliberately has no SQLite fallback. A valid remote PostgreSQL URL must be supplied through ``HOCKEY_DATABASE_URL``. """ def __init__(self, base_dir: Path) -> None: self.base_dir = Path(base_dir) self.url = self._required_postgresql_url() parsed_url = make_url(self.url) self._target = { "backend": "postgresql", "driver": parsed_url.drivername, "host": parsed_url.host or "", "port": parsed_url.port or 5432, "database": parsed_url.database or "", "remote_only": True, } self._schema_lock = RLock() self._last_schema_report: dict[str, Any] = { "ok": False, "repaired": [], "rebuilt": False, "error": "not_checked", **self._target, } self.engine: Engine = create_engine( self.url, future=True, hide_parameters=True, pool_pre_ping=True, pool_size=self._env_int("HOCKEY_DB_POOL_SIZE", 10, 1, 100), max_overflow=self._env_int("HOCKEY_DB_MAX_OVERFLOW", 20, 0, 200), pool_timeout=self._env_int("HOCKEY_DB_POOL_TIMEOUT", 30, 1, 300), pool_recycle=self._env_int("HOCKEY_DB_POOL_RECYCLE", 1800, 60, 86400), connect_args={ "connect_timeout": self._env_int( "HOCKEY_DB_CONNECT_TIMEOUT", 10, 1, 120 ), "application_name": "hockey-control-panel", }, ) self.SessionFactory = sessionmaker( bind=self.engine, autoflush=False, expire_on_commit=False, future=True, ) self._verify_connection() @staticmethod def _env_int(name: str, fallback: int, low: int, high: int) -> int: try: value = int(os.getenv(name, str(fallback))) except (TypeError, ValueError): value = fallback return max(low, min(high, value)) @staticmethod def _required_postgresql_url() -> str: raw = os.getenv("HOCKEY_DATABASE_URL", "").strip() if not raw: raise RuntimeError( "HOCKEY_DATABASE_URL не задан. Хоккейный проект работает только " "с удалённой PostgreSQL-базой. Создайте .env по примеру .env.example." ) # Accept common aliases, but always use the psycopg 3 driver installed # by requirements.txt. if raw.startswith("postgres://"): raw = "postgresql+psycopg://" + raw[len("postgres://") :] elif raw.startswith("postgresql://"): raw = "postgresql+psycopg://" + raw[len("postgresql://") :] try: parsed = make_url(raw) except Exception as error: raise RuntimeError("Некорректный HOCKEY_DATABASE_URL") from error if parsed.get_backend_name() != "postgresql": raise RuntimeError( "HOCKEY_DATABASE_URL должен указывать на PostgreSQL. " "SQLite и локальные файлы базы отключены." ) if not parsed.host or not parsed.database: raise RuntimeError( "В HOCKEY_DATABASE_URL должны быть указаны сервер и имя базы данных." ) if parsed.drivername != "postgresql+psycopg": parsed = parsed.set(drivername="postgresql+psycopg") return parsed.render_as_string(hide_password=False) def _verify_connection(self) -> None: try: with self.engine.connect() as connection: connection.execute(text("SELECT 1")) except SQLAlchemyError as error: target = f"{self._target['host']}:{self._target['port']}/{self._target['database']}" raise RuntimeError( "Не удалось подключиться к удалённой хоккейной PostgreSQL-базе " f"{target}. Проверьте HOCKEY_DATABASE_URL, сеть и права пользователя." ) from error def create_all(self) -> None: """Create missing tables and add newly introduced non-key columns.""" with self._schema_lock: self._verify_connection() try: migrated = self._migrate_operator_session_user_id() Base.metadata.create_all(self.engine) repaired = migrated + self._repair_game_table() for model in ( OperatorSession, Country, PenaltyType, Player, PlayerTournament, Coach, CoachTournament, GameRoster, GameControlState, ShootoutAttempt, Referee, RefereeTournament, GameOfficial, TournamentStanding, TournamentStandingRow, TournamentStatisticResource, TournamentStatisticRow, TeamSeasonStatistic, PlayerSeasonStatistic, GameTeamStatistic, UserPreference, VmixDevice, VmixAssignment, VmixMappingProfile, VmixMappingField, MappingContextVariable, MappingContextValue, MappingSqlDataSource, ): repaired.extend(self._repair_additive_table(model)) self._validate_game_table() Base.metadata.create_all(self.engine) self._last_schema_report = { "ok": True, "repaired": repaired, "rebuilt": False, "error": "", **self._target, } except Exception as error: self._last_schema_report = { "ok": False, "repaired": [], "rebuilt": False, "error": str(error), **self._target, } raise def _migrate_operator_session_user_id(self) -> list[str]: """Rename the legacy browser identity column without losing sessions.""" table_name = OperatorSession.__tablename__ inspector = inspect(self.engine) if table_name not in inspector.get_table_names(): return [] columns = {column["name"] for column in inspector.get_columns(table_name)} if "user_id" not in columns: return [] if "wfl_user_id" in columns: with self.engine.begin() as connection: connection.execute( text( f'UPDATE "{table_name}" SET "wfl_user_id" = "user_id" ' 'WHERE "wfl_user_id" = \'\' ' 'AND "user_id" NOT LIKE \'browser-%\'' ) ) connection.execute( text( f'UPDATE "{table_name}" SET "status" = \'closed\' ' 'WHERE "wfl_user_id" = \'\' OR "user_id" LIKE \'browser-%\'' ) ) connection.execute( text('DROP INDEX IF EXISTS "ix_hockey_sessions_user_status"') ) connection.execute( text(f'ALTER TABLE "{table_name}" DROP COLUMN "user_id"') ) return [f"{table_name}.dropped_legacy_user_id"] with self.engine.begin() as connection: connection.execute( text( f'ALTER TABLE "{table_name}" ' 'RENAME COLUMN "user_id" TO "wfl_user_id"' ) ) return [f"{table_name}.user_id->wfl_user_id"] def schema_status(self) -> dict[str, Any]: return dict(self._last_schema_report) def ensure_game_schema(self) -> dict[str, Any]: self.create_all() return self.schema_status() def _repair_game_table(self) -> list[str]: inspector = inspect(self.engine) if Game.__tablename__ not in inspector.get_table_names(): Game.__table__.create(self.engine, checkfirst=True) return ["created:hockey_games"] existing = { column["name"] for column in inspector.get_columns(Game.__tablename__) } repaired: list[str] = [] with self.engine.begin() as connection: for column in Game.__table__.columns: if column.name in existing: continue if column.primary_key: raise RuntimeError( f"В таблице отсутствует ключевой столбец {column.name}" ) connection.execute( text( f'ALTER TABLE "{Game.__tablename__}" ' f'ADD COLUMN "{column.name}" {self._column_add_ddl(column)}' ) ) repaired.append(column.name) return repaired def _repair_additive_table(self, model: Any) -> list[str]: table = model.__table__ inspector = inspect(self.engine) if table.name not in inspector.get_table_names(): table.create(self.engine, checkfirst=True) return [f"created:{table.name}"] existing = {column["name"] for column in inspector.get_columns(table.name)} repaired: list[str] = [] with self.engine.begin() as connection: for column in table.columns: if column.name in existing: continue if column.primary_key: raise RuntimeError( f"Cannot add missing primary key column {table.name}.{column.name}" ) connection.execute( text( f'ALTER TABLE "{table.name}" ' f'ADD COLUMN "{column.name}" {self._column_add_ddl(column)}' ) ) repaired.append(f"{table.name}.{column.name}") return repaired def _column_add_ddl(self, column: Any) -> str: sql_type = column.type.compile(dialect=self.engine.dialect) parts = [sql_type] if not column.nullable: parts.append("NOT NULL") default_sql = self._safe_default_sql(column) if default_sql is not None: parts.append(f"DEFAULT {default_sql}") return " ".join(parts) @staticmethod def _safe_default_sql(column: Any) -> str | None: try: python_type: type[Any] | None = column.type.python_type except (AttributeError, NotImplementedError): python_type = None if column.nullable: return None if python_type is bool: return "FALSE" if python_type in {int, float}: return "0" if python_type is datetime: return "CURRENT_TIMESTAMP" return "''" def _validate_game_table(self) -> None: inspector = inspect(self.engine) existing = { column["name"] for column in inspector.get_columns(Game.__tablename__) } expected = {column.name for column in Game.__table__.columns} missing = sorted(expected - existing) if missing: raise RuntimeError( "Не удалось обновить таблицу hockey_games. " f"Отсутствуют поля: {', '.join(missing)}" ) with self.engine.connect() as connection: connection.execute(text('SELECT * FROM "hockey_games" LIMIT 1')) @contextmanager def session(self) -> Iterator[Session]: session = self.SessionFactory() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close()