24 lines
1.1 KiB
Python
24 lines
1.1 KiB
Python
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 = 无论接口成功还是报错，最后一定关闭会话，释放数据库连接 |