5109d6f824
之前 (await ...).scalars() 链式在 SQLAlchemy 2.0 async 下报 'coroutine' has no attribute 'scalars' 错误。改为先 await 拿 result 再 .scalars(),这是 SQLAlchemy 2.0 推荐的 async 写法。
64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
"""批量修 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}")
|