fix(deploy): parameterize alembic schema probe

2026-06-21 07:04:39 +08:00
parent 654c0f3ab0
commit a9751783b2
1 changed files with 11 additions and 2 deletions
@@ -46,8 +46,17 @@ from sqlalchemy import create_engine, text
 url = os.environ["DATABASE_URL"]
 engine = create_engine(url)
 with engine.begin() as conn:
-    has_version = conn.execute(text("SELECT to_regclass($$public.alembic_version$$)")).scalar() is not None
-    has_tables = conn.execute(text("SELECT COUNT(*) FROM pg_tables WHERE schemaname = 'public' AND tablename != 'alembic_version'")).scalar()
+    has_version = conn.execute(
+        text("SELECT to_regclass(:table_name)"),
+        {"table_name": "public.alembic_version"},
+    ).scalar() is not None
+    has_tables = conn.execute(
+        text(
+            "SELECT COUNT(*) FROM pg_tables "
+            "WHERE schemaname = :schema_name AND tablename != :version_table"
+        ),
+        {"schema_name": "public", "version_table": "alembic_version"},
+    ).scalar()
    if has_tables and not has_version:
        raise SystemExit(42)
 PY