Compare commits
No commits in common. "63e900466cc1370287060c23d616d26af84a758c" and "890f10cbabc4bd7876b967eb391a9f391101e281" have entirely different histories.
63e900466c
...
890f10cbab
@ -14,11 +14,9 @@ argv or in the database.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import Callable, Iterator
|
from collections.abc import Callable
|
||||||
import contextlib
|
|
||||||
import csv
|
import csv
|
||||||
import io
|
import io
|
||||||
import json
|
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import shutil
|
import shutil
|
||||||
@ -81,36 +79,6 @@ def jrunner_path() -> Path:
|
|||||||
return get_config().jrunner_path
|
return get_config().jrunner_path
|
||||||
|
|
||||||
|
|
||||||
def _write_passfile(entries: dict[str, dict]) -> str:
|
|
||||||
"""Write a jrunner alias file for `entries` and return its path.
|
|
||||||
|
|
||||||
``entries`` maps alias -> connection dict. Passwords resolve here and reach
|
|
||||||
jrunner through this file rather than through ``-scp``/``-dcp``, because
|
|
||||||
anything on argv is readable by any local user via ``ps`` for the lifetime
|
|
||||||
of the call. ``mkstemp`` creates the file 0600.
|
|
||||||
|
|
||||||
The caller is responsible for deleting it — every call site already has a
|
|
||||||
``finally`` that removes the temporary SQL file, so cleanup rides along
|
|
||||||
there rather than forcing the whole body into another ``with``.
|
|
||||||
"""
|
|
||||||
fd, path = tempfile.mkstemp(suffix=".jrunnerpass")
|
|
||||||
with os.fdopen(fd, "w") as f:
|
|
||||||
for alias, conn in entries.items():
|
|
||||||
f.write(f"[{alias}]\n")
|
|
||||||
f.write(f"url={conn['jdbc_url']}\n")
|
|
||||||
f.write(f"user={conn.get('username') or ''}\n")
|
|
||||||
f.write(f"pass={resolve_password(conn.get('password'))}\n")
|
|
||||||
return path
|
|
||||||
|
|
||||||
|
|
||||||
def _unlink_quietly(*paths: str) -> None:
|
|
||||||
for p in paths:
|
|
||||||
try:
|
|
||||||
os.unlink(p)
|
|
||||||
except (FileNotFoundError, TypeError):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def version() -> tuple[bool, str]:
|
def version() -> tuple[bool, str]:
|
||||||
"""Return (ok, message) for use by pipekit doctor."""
|
"""Return (ok, message) for use by pipekit doctor."""
|
||||||
path = jrunner_path()
|
path = jrunner_path()
|
||||||
@ -138,16 +106,15 @@ def query(
|
|||||||
) -> QueryResult:
|
) -> QueryResult:
|
||||||
"""Run `sql` in jrunner query mode and parse CSV output."""
|
"""Run `sql` in jrunner query mode and parse CSV output."""
|
||||||
path = jrunner_path()
|
path = jrunner_path()
|
||||||
conn = {"jdbc_url": jdbc_url, "username": username, "password": password}
|
pw = resolve_password(password)
|
||||||
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
|
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
|
||||||
f.write(sql)
|
f.write(sql)
|
||||||
sql_path = f.name
|
sql_path = f.name
|
||||||
pass_path = _write_passfile({"src": conn})
|
|
||||||
try:
|
try:
|
||||||
argv = [str(path),
|
argv = [str(path),
|
||||||
"-sc", "src",
|
"-scu", jdbc_url,
|
||||||
"--passfile", pass_path,
|
"-scn", username or "",
|
||||||
"--strict",
|
"-scp", pw,
|
||||||
"-sq", sql_path,
|
"-sq", sql_path,
|
||||||
"-f", "csv"]
|
"-f", "csv"]
|
||||||
if trim:
|
if trim:
|
||||||
@ -155,10 +122,10 @@ def query(
|
|||||||
r = subprocess.run(argv, capture_output=True, text=True,
|
r = subprocess.run(argv, capture_output=True, text=True,
|
||||||
timeout=timeout, env=_subprocess_env())
|
timeout=timeout, env=_subprocess_env())
|
||||||
finally:
|
finally:
|
||||||
_unlink_quietly(sql_path, pass_path)
|
os.unlink(sql_path)
|
||||||
|
|
||||||
if r.returncode != 0:
|
if r.returncode != 0:
|
||||||
raise JrunnerError(_error_message(r.stdout, r.stderr),
|
raise JrunnerError(r.stderr.strip() or r.stdout.strip(),
|
||||||
stdout=r.stdout, stderr=r.stderr)
|
stdout=r.stdout, stderr=r.stderr)
|
||||||
silent = _detect_silent_failure(r.stdout, r.stderr)
|
silent = _detect_silent_failure(r.stdout, r.stderr)
|
||||||
if silent:
|
if silent:
|
||||||
@ -194,13 +161,14 @@ def migrate(
|
|||||||
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
|
with tempfile.NamedTemporaryFile("w", suffix=".sql", delete=False) as f:
|
||||||
f.write(sql)
|
f.write(sql)
|
||||||
sql_path = f.name
|
sql_path = f.name
|
||||||
pass_path = _write_passfile({"src": source_conn, "dst": dest_conn})
|
|
||||||
try:
|
try:
|
||||||
argv = [str(path),
|
argv = [str(path),
|
||||||
"-sc", "src",
|
"-scu", source_conn["jdbc_url"],
|
||||||
"-dc", "dst",
|
"-scn", source_conn.get("username") or "",
|
||||||
"--passfile", pass_path,
|
"-scp", resolve_password(source_conn.get("password")),
|
||||||
"--strict",
|
"-dcu", dest_conn["jdbc_url"],
|
||||||
|
"-dcn", dest_conn.get("username") or "",
|
||||||
|
"-dcp", resolve_password(dest_conn.get("password")),
|
||||||
"-sq", sql_path,
|
"-sq", sql_path,
|
||||||
"-dt", dest_table]
|
"-dt", dest_table]
|
||||||
if trim:
|
if trim:
|
||||||
@ -247,23 +215,19 @@ def migrate(
|
|||||||
stdout = "".join(stdout_lines)
|
stdout = "".join(stdout_lines)
|
||||||
stderr = "".join(stderr_buf)
|
stderr = "".join(stderr_buf)
|
||||||
finally:
|
finally:
|
||||||
_unlink_quietly(sql_path, pass_path)
|
os.unlink(sql_path)
|
||||||
|
|
||||||
if proc.returncode != 0:
|
if proc.returncode != 0:
|
||||||
raise JrunnerError(_error_message(stdout, stderr),
|
raise JrunnerError(stderr.strip() or stdout.strip(),
|
||||||
stdout=stdout, stderr=stderr)
|
stdout=stdout, stderr=stderr)
|
||||||
silent = _detect_silent_failure(stdout, stderr)
|
silent = _detect_silent_failure(stdout, stderr)
|
||||||
if silent:
|
if silent:
|
||||||
raise JrunnerError(silent, stdout=stdout, stderr=stderr)
|
raise JrunnerError(silent, stdout=stdout, stderr=stderr)
|
||||||
|
|
||||||
summary = _parse_summary(stderr)
|
return MigrateResult(
|
||||||
row_count = (summary or {}).get("rows")
|
row_count=_parse_row_count(stdout + "\n" + stderr),
|
||||||
if row_count is None:
|
stdout=stdout, stderr=stderr,
|
||||||
# Older jrunner without @summary — fall back to reading the count out of
|
)
|
||||||
# its prose. Removable once the pinned jar is the only one in play.
|
|
||||||
row_count = _parse_row_count(stdout + "\n" + stderr)
|
|
||||||
|
|
||||||
return MigrateResult(row_count=row_count, stdout=stdout, stderr=stderr)
|
|
||||||
|
|
||||||
|
|
||||||
def run_dest_sql(conn: dict, sql: str, *, timeout: int = 600) -> QueryResult:
|
def run_dest_sql(conn: dict, sql: str, *, timeout: int = 600) -> QueryResult:
|
||||||
@ -274,27 +238,6 @@ def run_dest_sql(conn: dict, sql: str, *, timeout: int = 600) -> QueryResult:
|
|||||||
sql, timeout=timeout, trim=False)
|
sql, timeout=timeout, trim=False)
|
||||||
|
|
||||||
|
|
||||||
_SUMMARY_PREFIX = "@summary "
|
|
||||||
|
|
||||||
|
|
||||||
def _parse_summary(stderr: str) -> dict | None:
|
|
||||||
"""Parse jrunner's ``@summary {...}`` line out of stderr.
|
|
||||||
|
|
||||||
jrunner emits exactly one, last, carrying ``status`` and (where known)
|
|
||||||
``rows`` and ``ms``. It lives on stderr because migration-mode stdout is
|
|
||||||
human-readable progress and query-mode stdout is CSV data. Returns None if
|
|
||||||
absent, which is how an older jrunner presents.
|
|
||||||
"""
|
|
||||||
for line in reversed((stderr or "").splitlines()):
|
|
||||||
line = line.strip()
|
|
||||||
if line.startswith(_SUMMARY_PREFIX):
|
|
||||||
try:
|
|
||||||
return json.loads(line[len(_SUMMARY_PREFIX):])
|
|
||||||
except ValueError:
|
|
||||||
return None
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
_ROW_COUNT_PATTERNS = (
|
_ROW_COUNT_PATTERNS = (
|
||||||
re.compile(r"(\d+)\s+rows?\s+(?:inserted|transferred|migrated|written)", re.I),
|
re.compile(r"(\d+)\s+rows?\s+(?:inserted|transferred|migrated|written)", re.I),
|
||||||
re.compile(r"inserted\s+(\d+)\s+rows?", re.I),
|
re.compile(r"inserted\s+(\d+)\s+rows?", re.I),
|
||||||
@ -317,26 +260,6 @@ def _parse_row_count(text: str) -> int | None:
|
|||||||
# nearly every failure site (see jrunner.java). Detect those by scanning
|
# nearly every failure site (see jrunner.java). Detect those by scanning
|
||||||
# for a Java stack-trace signature so callers don't treat silent failures
|
# for a Java stack-trace signature so callers don't treat silent failures
|
||||||
# as success.
|
# as success.
|
||||||
_JVM_NOISE_RE = re.compile(r"^Picked up JAVA_TOOL_OPTIONS:.*$", re.M)
|
|
||||||
|
|
||||||
|
|
||||||
def _error_message(stdout: str, stderr: str) -> str:
|
|
||||||
"""Best single-line cause for a failed jrunner call.
|
|
||||||
|
|
||||||
The JVM prints "Picked up JAVA_TOOL_OPTIONS: ..." as the first stderr line
|
|
||||||
whenever JAVA_TOOL_OPTIONS is set (which _subprocess_env always does, to
|
|
||||||
keep jt400 headless), so raising raw stderr surfaces that banner instead of
|
|
||||||
the actual error. Prefer the Java exception header, which names the real
|
|
||||||
failure, and fall back to whatever output remains once the banner is gone.
|
|
||||||
"""
|
|
||||||
cleaned_err = _JVM_NOISE_RE.sub("", stderr or "").strip()
|
|
||||||
cleaned_out = _JVM_NOISE_RE.sub("", stdout or "").strip()
|
|
||||||
m = _EXCEPTION_HEADER_RE.search(cleaned_err) or _EXCEPTION_HEADER_RE.search(cleaned_out)
|
|
||||||
if m:
|
|
||||||
return m.group(0).strip()
|
|
||||||
return cleaned_err or cleaned_out or "jrunner failed with no output"
|
|
||||||
|
|
||||||
|
|
||||||
_STACK_FRAME_RE = re.compile(r"^\s*at [\w.$<>]+\([^)\n]*\.java:\d+\)", re.M)
|
_STACK_FRAME_RE = re.compile(r"^\s*at [\w.$<>]+\([^)\n]*\.java:\d+\)", re.M)
|
||||||
_EXCEPTION_HEADER_RE = re.compile(
|
_EXCEPTION_HEADER_RE = re.compile(
|
||||||
r"^(?:[\w.$]+\.)*[\w$]+(?:Exception|Error)(?::[^\n]*)?$", re.M)
|
r"^(?:[\w.$]+\.)*[\w$]+(?:Exception|Error)(?::[^\n]*)?$", re.M)
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user