58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from pathlib import Path
|
|
import sys
|
|
|
|
from sqlalchemy import text
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.database import engine # noqa: E402
|
|
|
|
|
|
def _column_exists(conn, table_name: str, column_name: str) -> bool:
|
|
return bool(conn.execute(text("""
|
|
SELECT COUNT(1)
|
|
FROM information_schema.columns
|
|
WHERE table_schema = DATABASE()
|
|
AND table_name = :table_name
|
|
AND column_name = :column_name
|
|
"""), {"table_name": table_name, "column_name": column_name}).scalar())
|
|
|
|
|
|
def _primary_columns(conn, table_name: str) -> list[str]:
|
|
rows = conn.execute(text("""
|
|
SELECT column_name
|
|
FROM information_schema.key_column_usage
|
|
WHERE table_schema = DATABASE()
|
|
AND table_name = :table_name
|
|
AND constraint_name = 'PRIMARY'
|
|
ORDER BY ordinal_position
|
|
"""), {"table_name": table_name}).all()
|
|
return [row[0] for row in rows]
|
|
|
|
|
|
def main() -> None:
|
|
with engine.begin() as conn:
|
|
if not _column_exists(conn, "work_session_devices", "process_name"):
|
|
conn.execute(text("""
|
|
ALTER TABLE work_session_devices
|
|
ADD COLUMN process_name VARCHAR(128) NOT NULL DEFAULT '' AFTER device_no
|
|
"""))
|
|
|
|
if not _column_exists(conn, "device_qrcodes", "process_name"):
|
|
conn.execute(text("""
|
|
ALTER TABLE device_qrcodes
|
|
ADD COLUMN process_name VARCHAR(128) NOT NULL DEFAULT '' AFTER device_no
|
|
"""))
|
|
|
|
primary_columns = _primary_columns(conn, "device_qrcodes")
|
|
if primary_columns != ["device_no", "process_name"]:
|
|
conn.execute(text("ALTER TABLE device_qrcodes DROP PRIMARY KEY"))
|
|
conn.execute(text("ALTER TABLE device_qrcodes ADD PRIMARY KEY (device_no, process_name)"))
|
|
|
|
print("mold process qrcode migrated")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|