form2-limit-migrations: миграции для листа лимиты и для второй формы в целом

api/alembic/versions/0013_form2_limit.py

@@ -0,0 +1,84 @@
+import os
+import re
+import time
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0013"
+down_revision = "0012"
+branch_labels = None
+depends_on = None
+
+
+_DOLLAR_TAG_RE = re.compile(r"\$\w+\$")
+
+
+def _find_dollar_tag(line: str) -> str | None:
+    m = _DOLLAR_TAG_RE.search(line.strip())
+    return m.group(0) if m else None
+
+
+def _split_statements(sql: str) -> list[str]:
+    statements: list[str] = []
+    current: list[str] = []
+    in_dollar = False
+    dollar_tag: str | None = None
+
+    for line in sql.split("\n"):
+        stripped = line.strip()
+        if stripped.startswith("--"):
+            continue
+
+        if not in_dollar:
+            tag = _find_dollar_tag(stripped)
+            if tag and tag.endswith("$") and tag.startswith("$"):
+                dollar_tag = tag
+                in_dollar = True
+                current.append(line)
+                continue
+
+        if in_dollar and dollar_tag and stripped.startswith(dollar_tag):
+            after = stripped[len(dollar_tag):].strip()
+            if after == ";" or after == "":
+                in_dollar = False
+                dollar_tag = None
+                if after == ";":
+                    current.append(line)
+                    statements.append("\n".join(current))
+                    current = []
+                    continue
+
+        if not in_dollar and stripped.rstrip().endswith(";"):
+            current.append(line)
+            statements.append("\n".join(current))
+            current = []
+            continue
+
+        current.append(line)
+
+    remaining = "\n".join(current).strip()
+    if remaining:
+        statements.append(remaining)
+
+    return statements
+
+
+def upgrade() -> None:
+    ddl_path = os.path.join(os.path.dirname(__file__), "sql", "0013_form2_limit.sql")
+    with open(ddl_path) as f:
+        content = f.read()
+
+    statements = _split_statements(content)
+    for stmt in statements:
+        stripped = stmt.strip().rstrip(";").strip()
+        if not stripped:
+            continue
+        if all(l.strip().startswith("--") or not l.strip() for l in stripped.split("\n")):
+            continue
+        op.execute(stripped)
+
+
+def downgrade() -> None:
+    pass