(self, database_url: str = None)
| 95 | """数据库会话管理器""" |
| 96 | |
| 97 | def __init__(self, database_url: str = None): |
| 98 | if database_url is None: |
| 99 | env_url = os.environ.get("APP_DATABASE_URL") or os.environ.get("DATABASE_URL") |
| 100 | if env_url: |
| 101 | database_url = env_url |
| 102 | else: |
| 103 | # 优先使用 APP_DATA_DIR 环境变量（PyInstaller 打包后由 webui.py 设置） |
| 104 | data_dir = os.environ.get('APP_DATA_DIR') or os.path.join( |
| 105 | os.path.dirname(os.path.dirname(os.path.dirname(__file__))), |
| 106 | 'data' |
| 107 | ) |
| 108 | db_path = os.path.join(data_dir, 'database.db') |
| 109 | # 确保目录存在 |
| 110 | os.makedirs(data_dir, exist_ok=True) |
| 111 | database_url = f"sqlite:///{db_path}" |
| 112 | |
| 113 | self.database_url = _build_sqlalchemy_url(database_url) |
| 114 | self.engine = create_engine( |
| 115 | self.database_url, |
| 116 | **_build_engine_options(self.database_url), |
| 117 | ) |
| 118 | self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine) |
| 119 | |
| 120 | def get_db(self) -> Generator[Session, None, None]: |
| 121 | """ |
nothing calls this directly
no test coverage detected