解决SQLAlchemy创建数据库时“未知数据库”错误教程

DB_URI = f"mysql+pymysql://{self.root_username}:{self.root_password}@localhost/{self.new_database}"

import os
from dotenv import load_dotenv
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, declarative_base, relationship, Column, Integer, String

# 声明基类，用于定义ORM模型
Base = declarative_base()

# 定义一个简单的ORM模型作为示例
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

class DatabaseInitializer:
    def __init__(self):
        load_dotenv() # 加载 .env 文件中的环境变量

        self.root_username = os.getenv("ROOT_USERNAME")
        self.root_password = os.getenv("ROOT_PASSWORD")
        self.target_database_name = os.getenv("NEW_DATABASE")

        if None in [self.root_username, self.root_password, self.target_database_name]:
            raise ValueError("缺少必要的数据库环境变量（ROOT_USERNAME, ROOT_PASSWORD, NEW_DATABASE）。")

        # 用于连接到MySQL服务器（不指定具体数据库）的URI
        self.server_uri = f"mysql+pymysql://{self.root_username}:{self.root_password}@localhost/"
        # 用于连接到目标数据库的URI
        self.database_uri = f"mysql+pymysql://{self.root_username}:{self.root_password}@localhost/{self.target_database_name}"

    def create_database_if_not_exists(self):
        """
        连接到MySQL服务器并创建指定的数据库（如果不存在）。
        """
        print(f"尝试连接到MySQL服务器以创建数据库: {self.server_uri.split('@')[0]}@localhost/")
        # 创建一个连接到服务器而非特定数据库的引擎
        server_engine = create_engine(self.server_uri, echo=False) # echo=False 避免输出过多日志
        try:
            # 使用上下文管理器确保连接被正确关闭
            with server_engine.connect() as connection:
                # 使用 text() 包装原始SQL语句，并执行
                connection.execute(text(f"CREATE DATABASE IF NOT EXISTS {self.target_database_name}"))
                connection.commit() # 对于DDL操作，通常需要commit
            print(f"数据库 '{self.target_database_name}' 已确保存在。")
        except Exception as e:
            print(f"创建数据库时发生错误: {e}")
            raise
        finally:
            server_engine.dispose() # 释放所有连接池中的连接

    def create_tables(self):
        """
        连接到目标数据库并创建所有通过ORM模型定义的表。
        此方法应在数据库已存在之后调用。
        """
        print(f"尝试连接到目标数据库 '{self.target_database_name}' 以创建表。")
        # 创建一个连接到目标数据库的引擎
        target_engine = create_engine(self.database_uri, echo=False)
        try:
            # 使用Base.metadata.create_all()创建所有定义的表
            Base.metadata.create_all(target_engine)
            print(f"表已在数据库 '{self.target_database_name}' 中成功创建。")
        except Exception as e:
            print(f"创建表时发生错误: {e}")
            raise
        finally:
            target_engine.dispose() # 释放所有连接池中的连接

    def get_session(self):
        """
        返回一个绑定到目标数据库的SQLAlchemy会话。
        此方法应在数据库和表都已存在之后调用。
        """
        engine = create_engine(self.database_uri)
        Session = sessionmaker(bind=engine)
        return Session()

# 主执行逻辑
if __name__ == "__main__":
    # 请确保您的 .env 文件中包含以下变量：
    # ROOT_USERNAME=your_mysql_root_user
    # ROOT_PASSWORD=your_mysql_root_password
    # NEW_DATABASE=your_new_database_name

    db_initializer = DatabaseInitializer()

    # 步骤1: 创建数据库（如果不存在）
    db_initializer.create_database_if_not_exists()

    # 步骤2: 连接到新创建的数据库并创建表
    db_initializer.create_tables()

    # 可选：演示如何使用会话进行数据操作
    session = db_initializer.get_session()
    try:
        # 添加新用户
        new_user = User(name="Alice", email="alice@example.com")
        session.add(new_user)
        session.commit()
        print(f"添加用户: {new_user}")

        # 查询所有用户
        users = session.query(User).all()
        print("现有用户:")
        for user in users:
            print(user)
    except Exception as e:
        session.rollback() # 发生错误时回滚事务
        print(f"会话操作期间发生错误: {e}")
    finally:
        session.close() # 关闭会话
        print("会话已关闭。")