fix(safety): backfill text() 콜론 bind 오인 — exec_driver_sql 로 교체
정규식 '(?:' 의 콜론을 text() 가 bind param 으로 해석 (migration 러너 동일 함정). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -36,7 +36,7 @@ import sys
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
|
||||
|
||||
from sqlalchemy import text
|
||||
# text() 미사용 — exec_driver_sql (정규식 콜론 함정)
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
# ─── 술어별 (라벨, 카운트 SQL, 적용 SQL) ───────────────────────────────────────
|
||||
@@ -191,12 +191,14 @@ async def main() -> None:
|
||||
trans = await conn.begin()
|
||||
try:
|
||||
for label, sql in STEPS:
|
||||
result = await conn.execute(text(sql))
|
||||
# text() 는 정규식의 '(?:' 콜론을 bind param 으로 오인 (migration 러너와
|
||||
# 동일 함정) → driver 직결 실행
|
||||
result = await conn.exec_driver_sql(sql)
|
||||
print(f"[{tag}] {label}: {result.rowcount}행")
|
||||
|
||||
print("\n─── 검증 (트랜잭션 내 미리보기) ───")
|
||||
for label, sql in VERIFY_SQL:
|
||||
result = await conn.execute(text(sql))
|
||||
result = await conn.exec_driver_sql(sql)
|
||||
rows = result.fetchall()
|
||||
print(f"\n{label}:")
|
||||
for row in rows:
|
||||
|
||||
Reference in New Issue
Block a user