from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, sessionmaker # 1. 配置数据库连接地址(注释掉 SQLite,直接使用 MySQL) # SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@host.docker.internal:3306/0914tw" # 2. 创建 engine(数据库引擎) # 注意:MySQL 不需要 connect_args={"check_same_thread": False},这行要去掉 engine = create_engine(SQLALCHEMY_DATABASE_URL) # 3. 创建 Base(ORM 基类) Base = declarative_base() # 4. 创建 SessionLocal(会话工厂,必须在 engine 最终确定后再绑定) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 提供数据库会话,接口通过 Depends(get_db) 自动获取 def get_db(): db = SessionLocal() # 创建一个新的数据库会话 try: yield db # yield = 把会话"借出去"给接口函数用(执行到这里暂停,等接口用完) finally: db.close() # finally = 无论接口成功还是报错,最后一定关闭会话,释放数据库连接