fix(db): 适配 SQLAlchemy 2.0 API 变更
更新了与数据库交互的代码,以适配 SQLAlchemy 2.0 引入的 API 变更,解决相关的废弃警告和异步调用问题。 主要变更包括: - 使用 `scalars().first()` 替代已废弃的 `scalar()` 方法。 - 在获取所有标量结果时显式调用 `.all()`。 - 对异步引擎 `get_engine()` 的调用进行 `await`。 - 移除了向 Inspector 方法(如 `get_table_names`)传递多余的 `conn` 参数。
This commit is contained in:
@@ -249,7 +249,7 @@ class ChatManager:
|
|||||||
# 检查数据库中是否存在
|
# 检查数据库中是否存在
|
||||||
async def _db_find_stream_async(s_id: str):
|
async def _db_find_stream_async(s_id: str):
|
||||||
async with get_db_session() as session:
|
async with get_db_session() as session:
|
||||||
return (await session.execute(select(ChatStreams).where(ChatStreams.stream_id == s_id))).scalar()
|
return (await session.execute(select(ChatStreams).where(ChatStreams.stream_id == s_id))).scalars().first()
|
||||||
|
|
||||||
model_instance = await _db_find_stream_async(stream_id)
|
model_instance = await _db_find_stream_async(stream_id)
|
||||||
|
|
||||||
@@ -396,7 +396,7 @@ class ChatManager:
|
|||||||
async def _db_load_all_streams_async():
|
async def _db_load_all_streams_async():
|
||||||
loaded_streams_data = []
|
loaded_streams_data = []
|
||||||
async with get_db_session() as session:
|
async with get_db_session() as session:
|
||||||
for model_instance in (await session.execute(select(ChatStreams))).scalars():
|
for model_instance in (await session.execute(select(ChatStreams))).scalars().all():
|
||||||
user_info_data = {
|
user_info_data = {
|
||||||
"platform": model_instance.user_platform,
|
"platform": model_instance.user_platform,
|
||||||
"user_id": model_instance.user_id,
|
"user_id": model_instance.user_id,
|
||||||
|
|||||||
@@ -18,7 +18,7 @@ async def check_and_migrate_database():
|
|||||||
- 自动为现有表创建缺失的索引。
|
- 自动为现有表创建缺失的索引。
|
||||||
"""
|
"""
|
||||||
logger.info("正在检查数据库结构并执行自动迁移...")
|
logger.info("正在检查数据库结构并执行自动迁移...")
|
||||||
engine = get_engine()
|
engine = await get_engine()
|
||||||
|
|
||||||
async with engine.connect() as connection:
|
async with engine.connect() as connection:
|
||||||
# 在同步上下文中运行inspector操作
|
# 在同步上下文中运行inspector操作
|
||||||
@@ -28,7 +28,7 @@ async def check_and_migrate_database():
|
|||||||
inspector = await connection.run_sync(get_inspector)
|
inspector = await connection.run_sync(get_inspector)
|
||||||
|
|
||||||
# 在同步lambda中传递inspector
|
# 在同步lambda中传递inspector
|
||||||
db_table_names = await connection.run_sync(lambda conn: set(inspector.get_table_names(conn)))
|
db_table_names = await connection.run_sync(lambda conn: set(inspector.get_table_names()))
|
||||||
|
|
||||||
# 1. 首先处理表的创建
|
# 1. 首先处理表的创建
|
||||||
tables_to_create = []
|
tables_to_create = []
|
||||||
@@ -60,7 +60,7 @@ async def check_and_migrate_database():
|
|||||||
try:
|
try:
|
||||||
# 检查并添加缺失的列
|
# 检查并添加缺失的列
|
||||||
db_columns = await connection.run_sync(
|
db_columns = await connection.run_sync(
|
||||||
lambda conn: {col["name"] for col in inspector.get_columns(table_name, conn)}
|
lambda conn: {col["name"] for col in inspector.get_columns(table_name)}
|
||||||
)
|
)
|
||||||
model_columns = {col.name for col in table.c}
|
model_columns = {col.name for col in table.c}
|
||||||
missing_columns = model_columns - db_columns
|
missing_columns = model_columns - db_columns
|
||||||
@@ -100,7 +100,7 @@ async def check_and_migrate_database():
|
|||||||
|
|
||||||
# 检查并创建缺失的索引
|
# 检查并创建缺失的索引
|
||||||
db_indexes = await connection.run_sync(
|
db_indexes = await connection.run_sync(
|
||||||
lambda conn: {idx["name"] for idx in inspector.get_indexes(table_name, conn)}
|
lambda conn: {idx["name"] for idx in inspector.get_indexes(table_name)}
|
||||||
)
|
)
|
||||||
model_indexes = {idx.name for idx in table.indexes}
|
model_indexes = {idx.name for idx in table.indexes}
|
||||||
missing_indexes = model_indexes - db_indexes
|
missing_indexes = model_indexes - db_indexes
|
||||||
|
|||||||
Reference in New Issue
Block a user