""" 数据库连接和会话管理 """ from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os from pathlib import Path from backend.config import DATA_DIR # 确保数据目录存在 os.makedirs(DATA_DIR, exist_ok=True) # 数据库URL SQLALCHEMY_DATABASE_URL = f"sqlite:///{DATA_DIR}/lightstar.db" # 创建引擎 engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) # 创建会话 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 基础模型类 Base = declarative_base() # 依赖函数,获取数据库会话 def get_db(): db = SessionLocal() try: yield db finally: db.close()