File size: 4,468 Bytes
cd9bca9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
"""
資料庫初始化腳本
使用 SQLAlchemy 建立資料表和插入範例資料
"""
from sqlalchemy import create_engine
from backend.database.connection import Base, engine
from backend.database.models import User, Product, Order, OrderItem, LineMessage, UserSession
from backend.config import settings
import logging
logger = logging.getLogger(__name__)
def create_tables():
"""建立所有資料表"""
try:
logger.info("開始建立資料表...")
Base.metadata.create_all(bind=engine)
logger.info("資料表建立完成！")
return True
except Exception as e:
logger.error(f"建立資料表失敗: {str(e)}")
return False
def insert_sample_data():
"""插入範例資料"""
from backend.database.connection import SessionLocal
db = SessionLocal()
try:
logger.info("開始插入範例資料...")
# 檢查是否已有資料
existing_products = db.query(Product).count()
if existing_products > 0:
logger.info("資料表已有資料，跳過範例資料插入")
return True
# 插入範例商品
sample_products = [
Product(
name="iPhone 15 Pro",
description="最新款 iPhone，配備 A17 Pro 晶片",
price=35900,
stock=50,
category="手機"
),
Product(
name="MacBook Air M2",
description="輕薄筆記型電腦，搭載 M2 晶片",
price=37900,
stock=30,
category="筆電"
),
Product(
name="AirPods Pro",
description="主動降噪無線耳機",
price=7490,
stock=100,
category="耳機"
),
Product(
name="iPad Air",
description="輕薄平板電腦，適合工作和娛樂",
price=18900,
stock=25,
category="平板"
),
Product(
name="Apple Watch Series 9",
description="智慧手錶，健康監測功能",
price=12900,
stock=40,
category="穿戴裝置"
)
]
for product in sample_products:
db.add(product)
# 插入範例用戶
sample_users = [
User(
user_id="U001",
name="張小明",
email="ming@example.com",
display_name="小明"
),
User(
user_id="U002",
name="李小華",
email="hua@example.com",
display_name="小華"
),
User(
user_id="U003",
name="王小美",
email="mei@example.com",
display_name="小美"
)
]
for user in sample_users:
db.add(user)
# 提交變更
db.commit()
logger.info("範例資料插入完成！")
return True
except Exception as e:
db.rollback()
logger.error(f"插入範例資料失敗: {str(e)}")
return False
finally:
db.close()
def init_database():
"""初始化資料庫"""
logger.info("=== 開始初始化資料庫 ===")
# 建立資料表
if not create_tables():
logger.error("資料表建立失敗，停止初始化")
return False
# 插入範例資料
if not insert_sample_data():
logger.error("範例資料插入失敗")
return False
logger.info("=== 資料庫初始化完成 ===")
return True
def drop_all_tables():
"""刪除所有資料表（謹慎使用）"""
try:
logger.warning("正在刪除所有資料表...")
Base.metadata.drop_all(bind=engine)
logger.warning("所有資料表已刪除！")
return True
except Exception as e:
logger.error(f"刪除資料表失敗: {str(e)}")
return False
if __name__ == "__main__":
import sys
if len(sys.argv) > 1 and sys.argv[1] == "--drop":
# 刪除並重建資料表
drop_all_tables()
init_database()
else:
# 正常初始化
init_database() |