fix: API 全部改用显式两步走 await session.execute + result.scalars()

之前 (await ...).scalars() 链式在 SQLAlchemy 2.0 async 下报
'coroutine' has no attribute 'scalars' 错误。改为先 await 拿 result
再 .scalars(),这是 SQLAlchemy 2.0 推荐的 async 写法。
This commit is contained in:
Mavis
2026-06-07 23:22:56 +08:00
parent 2e75985a3c
commit 5109d6f824
8 changed files with 147 additions and 36 deletions

63
scripts/_fix_scalars.py Normal file
View File

@@ -0,0 +1,63 @@
"""批量修 API 文件:把 `(await ...).scalars()` 改成显式两步走。"""
import re
from pathlib import Path
api_dir = Path("D:/selftools/diary-news/backend/app/api")
files = list(api_dir.glob("*.py"))
# 模式 1: user = (await ...).scalars().first() (多行括号形式)
# 模式 2: rows = (await ...).scalars() (单行)
# 都改成 result = await ...; user = result.scalars().first() 这种
for f in files:
src = f.read_text(encoding="utf-8")
orig = src
changed = False
# 模式 1:跨行的( await ... ).scalars().first()
# 匹配:任意前缀(空白)+ ( 多行 await session.execute(...) ) .scalars() .first()
pat1 = re.compile(
r'(\s+)([\w_]+)\s*=\s*\(\s*\n'
r'(\s+)await\s+session\.execute\((.*?)\)\s*\n'
r'\s+\)\s*'
r'\.scalars\(\)\s*'
r'\.first\(\)',
re.DOTALL,
)
def repl1(m):
indent = m.group(1)
var = m.group(2)
inner_indent = m.group(3)
exec_arg = m.group(4)
return (
f"{indent}result = await session.execute({exec_arg})\n"
f"{indent}{var} = result.scalars().first()"
)
new = pat1.sub(repl1, src)
if new != src:
changed = True
src = new
# 模式 2:单行 (await session.execute(...)).scalars()
pat2 = re.compile(
r'(\s+)([\w_]+)\s*=\s*\(await\s+session\.execute\((.*?)\)\)\s*\.scalars\(\)',
re.DOTALL,
)
def repl2(m):
indent = m.group(1)
var = m.group(2)
exec_arg = m.group(3)
return (
f"{indent}result = await session.execute({exec_arg})\n"
f"{indent}{var} = result.scalars()"
)
new = pat2.sub(repl2, src)
if new != src:
changed = True
src = new
if changed:
f.write_text(src, encoding="utf-8")
print(f"[ok] {f.name}")
else:
print(f" {f.name}")