解决SQLAlchemy创建表时“未知数据库”错误的专业指南

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

# 定义Base对象，所有ORM模型都将继承它
Base = declarative_base()

# 示例ORM模型（根据您的实际需求定义）
class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String(255), nullable=False)
    author = Column(String(255), nullable=False)
    price = Column(Float)

    def __repr__(self):
        return f"<Book(id={self.id}, title='{self.title}', author='{self.author}')>"

class DatabaseManager:
    def __init__(self):
        load_dotenv()

        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")

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

        self.engine = None
        self.Session = None
        self.session = None

    def _get_engine_for_db_creation(self):
        """获取用于创建数据库的引擎（不连接到特定数据库）。"""
        return create_engine(self.db_creation_uri, echo=False)

    def _get_engine_for_target_db(self):
        """获取用于连接到目标数据库的引擎。"""
        return create_engine(self.target_db_uri, echo=False)

    def create_database_if_not_exists(self):
        """连接到MySQL服务器并创建目标数据库（如果不存在）。"""
        print(f"尝试创建或验证数据库 '{self.target_database_name}'...")
        engine_for_creation = self._get_engine_for_db_creation()
        try:
            with engine_for_creation.connect() as connection:
                # 执行DDL语句创建数据库
                connection.execute(text(f"CREATE DATABASE IF NOT EXISTS {self.target_database_name}"))
                # 对于某些MySQL配置或驱动，可能需要显式commit
                connection.commit()
            print(f"数据库 '{self.target_database_name}' 已成功创建或已存在。")
        except Exception as e:
            print(f"创建数据库时发生错误: {e}")
            raise
        finally:
            engine_for_creation.dispose() # 确保关闭连接池中的所有连接

    def initialize_target_db_connection(self):
        """初始化连接到目标数据库的引擎和会话。"""
        print(f"初始化连接到数据库 '{self.target_database_name}'...")
        self.engine = self._get_engine_for_target_db()
        self.Session = sessionmaker(bind=self.engine)
        self.session = self.Session()
        print("目标数据库连接初始化完成。")

    def create_tables(self):
        """在已连接的数据库中创建所有定义的表。"""
        if not self.engine:
            raise RuntimeError("数据库连接未初始化。请先调用 initialize_target_db_connection。")
        print("开始创建表结构...")
        Base.metadata.create_all(self.engine)
        print("表结构创建成功。")

    def close_connection(self):
        """关闭当前会话并处理引擎资源。"""
        if self.session:
            self.session.close()
            print("会话已关闭。")
        if self.engine:
            self.engine.dispose() # 释放引擎持有的所有连接
            print("引擎资源已释放。")

# 示例使用
if __name__ == "__main__":
    # 为了演示，可以创建一个临时的.env文件
    # 在实际应用中，请确保您的.env文件包含正确的数据库凭据
    with open(".env", "w") as f:
        f.write("ROOT_USERNAME=root\n")
        f.write("ROOT_PASSWORD=your_mysql_root_password\n") # **请替换为您的MySQL root用户密码**
        f.write("NEW_DATABASE=my_new_books_db\n") # 目标数据库名称

    db_manager = DatabaseManager()
    try:
        # 步骤1: 创建数据库实例
        db_manager.create_database_if_not_exists()

        # 步骤2: 初始化连接到新创建的数据库，并创建表
        db_manager.initialize_target_db_connection()
        db_manager.create_tables()

        # 此时，您可以开始向数据库中添加数据
        # new_book = Book(title="The Great Gatsby", author="F. Scott Fitzgerald", price=12.99)
        # db_manager.session.add(new_book)
        # db_manager.session.commit()
        # print(f"添加书籍: {new_book}")

    except Exception as e:
        print(f"执行数据库操作时发生错误: {e}")
    finally:
        db_manager.close_connection()
        # 清理临时的.env文件
        if os.path.exists(".env"):
            os.remove(".env")