Explorar o código

test(db): add User model and DB test script (not containing secrets)

- add app/models/user.py
- add scripts/test_db.py

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com
tanlie hai 1 mes
pai
achega
abd926f038
Modificáronse 4 ficheiros con 75 adicións e 0 borrados
  1. 0 0
      FastApi/.env.example
  2. 0 0
      FastApi/README.md
  3. 15 0
      FastApi/app/models/user.py
  4. 60 0
      FastApi/scripts/test_db.py

+ 0 - 0
.env.example → FastApi/.env.example


+ 0 - 0
README.md → FastApi/README.md


+ 15 - 0
FastApi/app/models/user.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, String, DateTime, func
+from app.core.db import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String(50), unique=True, nullable=False, comment="用户名")
+    mobile = Column(String(32), unique=True, nullable=True, comment="手机号")
+    email = Column(String(120), unique=True, nullable=True, comment="邮箱")
+    created_at = Column(DateTime, server_default=func.now(), nullable=False)
+
+    def __repr__(self):
+        return f"<User id={self.id} username={self.username} mobile={self.mobile}>"}

+ 60 - 0
FastApi/scripts/test_db.py

@@ -0,0 +1,60 @@
+"""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()