wish/FastApi/scripts/test_db.py

"""Test DB connection: create tables, insert a user, and query it.

This script reads DATABASE_URL from environment (or uses app.core.db defaults).
Run locally with e.g. (PowerShell):
$env:DATABASE_URL = "mysql+pymysql://root:%40%40qinys12346..@192.168.1.177:3306/wishing_tree?charset=utf8mb4&serverTimezone=Asia/Shanghai"
python scripts/test_db.py
"""
import os
import sys

try:
    from app.core.db import engine, SessionLocal, Base
except Exception as e:
    print("Failed to import database module:", e)
    raise

# Import models so they are registered with Base
try:
    import app.models.user as user_model
except Exception as e:
    print("Failed to import models:", e)
    raise

from app.models.user import User


def main():
    print("Engine URL:", str(engine.url))
    try:
        print("Creating tables...")
        Base.metadata.create_all(bind=engine)
    except Exception as e:
        print("Create tables failed:", e)
        sys.exit(2)

    db = SessionLocal()
    try:
        # Insert a test user (avoid duplicate unique constraint)
        existing = db.query(User).filter(User.username == "test_user").first()
        if existing:
            print("Test user already exists:", existing)
        else:
            u = User(username="test_user", mobile="18888888888", email="test@example.com")
            db.add(u)
            db.commit()
            db.refresh(u)
            print("Inserted user:", u)

        # Query back
        one = db.query(User).filter(User.username == "test_user").first()
        print("Queried user:", one)
    except Exception as e:
        print("DB operation failed:", e)
        raise
    finally:
        db.close()


if __name__ == "__main__":
    main()