FlaskProject/db_utils/data.py

# -*- coding: utf-8 -*-
import mysql.connector
from mysql.connector import Error

# 配置数据库连接
DB_CONFIG = {
    'host': '60.204.139.57',
    'user': 'root',
    'password': '1',
    'port': 6003,
    'database': 'files'  # 替换为数据库名称
}

# 初始化数据库
def initialize_database():
    try:
        # 连接 MySQL 服务器（不指定数据库）
        connection = mysql.connector.connect(
            host=DB_CONFIG['host'],
            user=DB_CONFIG['user'],
            password=DB_CONFIG['password'],
            port=DB_CONFIG['port']
        )

        if connection.is_connected():
            print("成功连接到 MySQL 服务器")
            cursor = connection.cursor()

            # 检查数据库是否存在
            cursor.execute(f"SHOW DATABASES LIKE '{DB_CONFIG['database']}';")
            result = cursor.fetchone()

            if result:
                print(f"数据库 {DB_CONFIG['database']} 已存在。")
            else:
                print(f"数据库 {DB_CONFIG['database']} 不存在，正在创建...")
                cursor.execute(f"CREATE DATABASE {DB_CONFIG['database']};")
                print(f"数据库 {DB_CONFIG['database']} 创建成功。")

    except Error as err:
        print(f"连接 MySQL 时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 服务器连接已关闭")

# 初始化表
def initialize_table():
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 创建表
            cursor.execute('''CREATE TABLE IF NOT EXISTS fruit (
                                id INT AUTO_INCREMENT PRIMARY KEY,
                                name VARCHAR(100),
                                origin VARCHAR(100)
                              )''')
            print("--- 创建 fruit 表 ---")
    except Error as err:
        print(f"初始化表时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")

# 插入数据
def insert_data(name, origin):
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 插入数据
            cursor.execute("INSERT INTO fruit (name, origin) VALUES (%s, %s)", (name, origin))
            connection.commit()
            print(f"插入数据成功：({name}, {origin})")
    except Error as err:
        print(f"插入数据时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")

def delete_data(name):
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 删除数据
            cursor.execute("DELETE FROM fruit WHERE name = %s", (name,))
            connection.commit()
            print(f"成功删除名称为 {name} 的记录")
    except Error as err:
        print(f"删除数据时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")


# 查询数据
def fetch_data():
    try:
        connection = mysql.connector.connect(**DB_CONFIG)
        if connection.is_connected():
            cursor = connection.cursor()

            # 查询数据
            cursor.execute("SELECT * FROM fruit")
            rows = cursor.fetchall()
            print("--- 查询结果 ---")
            for row in rows:
                print(row)
            return rows
    except Error as err:
        print(f"查询数据时发生错误：{err}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
            print("MySQL 连接已关闭")

# 主函数
if __name__ == "__main__":
    # 初始化数据库和表
    initialize_database()
    initialize_table()

    # 插入示例数据
    insert_data('Apple', 'China')
    insert_data('Banana', 'Ecuador')

    # 查询数据
    fetch_data()