| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758 |
- # -*- coding: utf-8 -*-
- """MySQL 数据库：用户、习题、答题记录、错题、熟练度、答疑会话、学情报表"""
- import pymysql
- pymysql.install_as_MySQLdb()
- from sqlalchemy import create_engine, inspect, text
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.orm import sessionmaker, Session
- from backend.config import get_settings
- settings = get_settings()
- engine = create_engine(settings.MYSQL_URI, pool_pre_ping=True)
- SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
- Base = declarative_base()
- def get_db() -> Session:
- db = SessionLocal()
- try:
- yield db
- finally:
- db.close()
- def ensure_user_profile_columns() -> None:
- """
- 确保 users 表中存在个人信息相关字段。
- - 对于全新数据库：由 SQLAlchemy 根据模型一次性创建完整表结构；
- - 对于已存在的表：在缺少字段时自动执行 ALTER TABLE 补列，避免手动迁移。
- """
- inspector = inspect(engine)
- if "users" not in inspector.get_table_names():
- return
- existing_columns = {col["name"] for col in inspector.get_columns("users")}
- alters = []
- if "full_name" not in existing_columns:
- alters.append("ADD COLUMN full_name VARCHAR(50) NULL")
- if "student_id" not in existing_columns:
- alters.append("ADD COLUMN student_id VARCHAR(50) NULL")
- if "class_name" not in existing_columns:
- alters.append("ADD COLUMN class_name VARCHAR(100) NULL")
- if "grade" not in existing_columns:
- alters.append("ADD COLUMN grade VARCHAR(20) NULL")
- if "email" not in existing_columns:
- alters.append("ADD COLUMN email VARCHAR(120) NULL")
- if "phone" not in existing_columns:
- alters.append("ADD COLUMN phone VARCHAR(20) NULL")
- if not alters:
- return
- alter_sql = "ALTER TABLE users " + ", ".join(alters)
- with engine.connect() as conn:
- conn.execute(text(alter_sql))
|
