初始化知识库数据库 Args: db_path: 数据库文件路径, 默认位于 AstrBot 数据目录下的 knowledge_base/kb.db
(self, db_path: str | None = None)
| 21 | |
| 22 | class KBSQLiteDatabase: |
| 23 | def __init__(self, db_path: str | None = None) -> None: |
| 24 | """初始化知识库数据库 |
| 25 | |
| 26 | Args: |
| 27 | db_path: 数据库文件路径, 默认位于 AstrBot 数据目录下的 knowledge_base/kb.db |
| 28 | |
| 29 | """ |
| 30 | if db_path is None: |
| 31 | db_path = str(Path(get_astrbot_knowledge_base_path()) / "kb.db") |
| 32 | self.db_path = db_path |
| 33 | self.DATABASE_URL = f"sqlite+aiosqlite:///{db_path}" |
| 34 | self.inited = False |
| 35 | |
| 36 | # 确保目录存在 |
| 37 | Path(db_path).parent.mkdir(parents=True, exist_ok=True) |
| 38 | |
| 39 | # 创建异步引擎 |
| 40 | self.engine = create_async_engine( |
| 41 | self.DATABASE_URL, |
| 42 | echo=False, |
| 43 | pool_pre_ping=True, |
| 44 | pool_recycle=3600, |
| 45 | ) |
| 46 | |
| 47 | # 创建会话工厂 |
| 48 | self.async_session = async_sessionmaker( |
| 49 | self.engine, |
| 50 | class_=AsyncSession, |
| 51 | expire_on_commit=False, |
| 52 | ) |
| 53 | |
| 54 | @asynccontextmanager |
| 55 | async def get_db(self): |
nothing calls this directly
no test coverage detected