223 lines
7.1 KiB
Python
223 lines
7.1 KiB
Python
"""
|
||
Темп миграций Alembic / schema_bootstrap для слабой флешки (SD на Raspberry Pi).
|
||
|
||
Паузы и PRAGMA journal_mode=DELETE + synchronous=FULL снижают пиковую нагрузку на носитель
|
||
во время DDL; после миграции рабочие подключения снова получают WAL из create_app.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import platform
|
||
import time
|
||
from typing import Any, Optional
|
||
|
||
_log = logging.getLogger(__name__)
|
||
|
||
|
||
def is_arm_platform() -> bool:
|
||
machine = platform.machine().lower()
|
||
return any(part in machine for part in ("arm", "aarch64"))
|
||
|
||
|
||
def _env_float(name: str, default: float) -> float:
|
||
raw = os.getenv(name, "").strip()
|
||
if not raw:
|
||
return default
|
||
try:
|
||
return max(0.0, float(raw))
|
||
except ValueError:
|
||
return default
|
||
|
||
|
||
def migration_pause_sec() -> float:
|
||
"""Пауза между крупными фазами (до/после upgrade, между bind)."""
|
||
default = 1.5 if is_arm_platform() else 0.0
|
||
return _env_float("WESP_MIGRATION_PAUSE_SEC", default)
|
||
|
||
|
||
def migration_pause_bind_sec() -> float:
|
||
"""Пауза между recipes.db и reports.db в env.py."""
|
||
default = 3.0 if is_arm_platform() else 0.5
|
||
return _env_float("WESP_MIGRATION_PAUSE_BIND_SEC", default)
|
||
|
||
|
||
def migration_pause_step_sec() -> float:
|
||
"""Короткая пауза после каждой таблицы / колонки в schema_bootstrap."""
|
||
default = 0.2 if is_arm_platform() else 0.0
|
||
return _env_float("WESP_MIGRATION_PAUSE_STEP_SEC", default)
|
||
|
||
|
||
def migration_sqlite_journal_mode() -> str:
|
||
"""Пусто по умолчанию: смена journal_mode (DELETE) даёт database is locked с WAL."""
|
||
raw = os.getenv("WESP_MIGRATION_SQLITE_JOURNAL_MODE", "").strip()
|
||
if raw:
|
||
return raw.upper()
|
||
# Только на Pi явно можно задать DELETE в wesp.env; без env — не трогаем режим.
|
||
return ""
|
||
|
||
|
||
def migration_sqlite_synchronous() -> str:
|
||
return (os.getenv("WESP_MIGRATION_SQLITE_SYNCHRONOUS", "FULL") or "FULL").strip().upper()
|
||
|
||
|
||
def migration_sqlite_busy_timeout_ms() -> int:
|
||
raw = os.getenv("WESP_MIGRATION_SQLITE_BUSY_TIMEOUT_MS", "").strip()
|
||
if raw:
|
||
try:
|
||
return max(1000, int(raw))
|
||
except ValueError:
|
||
pass
|
||
return 60000
|
||
|
||
|
||
def apply_sqlite_migration_pragmas(dbapi_connection: Any) -> None:
|
||
"""На время DDL: меньше соседних -wal/-shm и более предсказуемый fsync."""
|
||
cursor = dbapi_connection.cursor()
|
||
try:
|
||
cursor.execute(f"PRAGMA busy_timeout={migration_sqlite_busy_timeout_ms()}")
|
||
sync = migration_sqlite_synchronous()
|
||
if sync:
|
||
cursor.execute(f"PRAGMA synchronous={sync}")
|
||
cursor.execute("PRAGMA cache_size=-2000")
|
||
|
||
jm = migration_sqlite_journal_mode()
|
||
if not jm:
|
||
return
|
||
# journal_mode требует эксклюзивной блокировки; при неудаче DDL всё равно идёт в текущем режиме.
|
||
for attempt in range(12):
|
||
try:
|
||
cursor.execute(f"PRAGMA journal_mode={jm}")
|
||
break
|
||
except Exception as exc:
|
||
msg = str(exc).lower()
|
||
if "locked" not in msg and "busy" not in msg:
|
||
raise
|
||
if attempt >= 11:
|
||
_log.warning(
|
||
"PRAGMA journal_mode=%s пропущен (файл занят): %s",
|
||
jm,
|
||
exc,
|
||
)
|
||
return
|
||
time.sleep(0.35 * (attempt + 1))
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def sqlite_path_from_uri(uri: str) -> Optional[str]:
|
||
try:
|
||
from sqlalchemy.engine.url import make_url
|
||
|
||
u = make_url(str(uri))
|
||
if u.drivername != "sqlite" or not u.database or u.database == ":memory:":
|
||
return None
|
||
path = str(u.database)
|
||
if not os.path.isabs(path):
|
||
path = os.path.abspath(path)
|
||
return path
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _checkpoint_sqlite_file(path: str) -> None:
|
||
import sqlite3
|
||
|
||
conn = sqlite3.connect(path, timeout=60)
|
||
try:
|
||
conn.execute(f"PRAGMA busy_timeout={migration_sqlite_busy_timeout_ms()}")
|
||
try:
|
||
conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||
except sqlite3.OperationalError as exc:
|
||
if "locked" not in str(exc).lower() and "busy" not in str(exc).lower():
|
||
raise
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def release_sqlalchemy_sqlite_locks(app: Any) -> None:
|
||
"""Закрывает пулы Flask-SQLAlchemy и сбрасывает WAL перед Alembic."""
|
||
from app import db # noqa: WPS433
|
||
|
||
uris: list[str] = [str(app.config.get("SQLALCHEMY_DATABASE_URI") or "")]
|
||
binds = app.config.get("SQLALCHEMY_BINDS") or {}
|
||
if binds.get("reports"):
|
||
uris.append(str(binds["reports"]))
|
||
|
||
for uri in uris:
|
||
path = sqlite_path_from_uri(uri)
|
||
if not path or not os.path.isfile(path):
|
||
continue
|
||
try:
|
||
_checkpoint_sqlite_file(path)
|
||
except Exception:
|
||
_log.warning("WAL checkpoint перед миграцией: %s", path, exc_info=True)
|
||
|
||
try:
|
||
with app.app_context():
|
||
db.session.remove()
|
||
except Exception:
|
||
_log.debug("release_sqlalchemy_sqlite_locks: session.remove", exc_info=True)
|
||
|
||
engines: list[Any] = []
|
||
try:
|
||
engines.extend(db.engines.values())
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if db.engine not in engines:
|
||
engines.append(db.engine)
|
||
except Exception:
|
||
pass
|
||
|
||
for eng in engines:
|
||
try:
|
||
if eng is not None:
|
||
eng.dispose()
|
||
except Exception:
|
||
_log.debug("dispose engine failed", exc_info=True)
|
||
|
||
if uris:
|
||
time.sleep(0.4)
|
||
|
||
|
||
def sqlite_flush_after_migration(dbapi_connection: Any) -> None:
|
||
"""Сброс буферов перед паузой (если до этого был WAL)."""
|
||
cursor = dbapi_connection.cursor()
|
||
try:
|
||
try:
|
||
cursor.execute("PRAGMA wal_checkpoint(FULL)")
|
||
except Exception:
|
||
pass
|
||
try:
|
||
cursor.execute("PRAGMA optimize")
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def migration_pause(phase: str, *, seconds: Optional[float] = None) -> None:
|
||
sec = migration_pause_sec() if seconds is None else max(0.0, float(seconds))
|
||
if sec <= 0:
|
||
return
|
||
_log.info("Миграция: пауза %.1f с (%s)", sec, phase)
|
||
time.sleep(sec)
|
||
|
||
|
||
def migration_pause_bind(phase: str) -> None:
|
||
sec = migration_pause_bind_sec()
|
||
if sec <= 0:
|
||
return
|
||
_log.info("Миграция: пауза между БД %.1f с (%s)", sec, phase)
|
||
time.sleep(sec)
|
||
|
||
|
||
def migration_pause_step(phase: str) -> None:
|
||
sec = migration_pause_step_sec()
|
||
if sec <= 0:
|
||
return
|
||
_log.debug("Миграция: шаг %.2f с (%s)", sec, phase)
|
||
time.sleep(sec)
|