67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import os
|
|
import subprocess
|
|
from typing import Any, Mapping
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CatalogExportResult:
|
|
status: str
|
|
command: str | None = None
|
|
workdir: str | None = None
|
|
returncode: int | None = None
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
|
|
|
|
def _read_config_value(config_snapshot: Mapping[str, Any] | None, key: str) -> str | None:
|
|
snapshot = config_snapshot or {}
|
|
if key in snapshot:
|
|
value = snapshot.get(key)
|
|
return None if value is None else str(value)
|
|
value = os.environ.get(key)
|
|
return None if value is None else str(value)
|
|
|
|
|
|
def run_catalog_export_command(config_snapshot: Mapping[str, Any] | None) -> CatalogExportResult:
|
|
command = _read_config_value(config_snapshot, "CATALOG_EXPORT_COMMAND")
|
|
workdir = _read_config_value(config_snapshot, "CATALOG_EXPORT_WORKDIR")
|
|
normalized_command = (command or "").strip()
|
|
|
|
if not normalized_command:
|
|
return CatalogExportResult(
|
|
status="skipped",
|
|
command=normalized_command or None,
|
|
workdir=workdir,
|
|
)
|
|
|
|
try:
|
|
completed = subprocess.run(
|
|
normalized_command,
|
|
shell=True,
|
|
cwd=workdir,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except Exception as exc:
|
|
return CatalogExportResult(
|
|
status="failed",
|
|
command=normalized_command,
|
|
workdir=workdir,
|
|
stderr=str(exc) or exc.__class__.__name__,
|
|
)
|
|
|
|
status = "succeeded" if completed.returncode == 0 else "failed"
|
|
|
|
return CatalogExportResult(
|
|
status=status,
|
|
command=normalized_command,
|
|
workdir=workdir,
|
|
returncode=completed.returncode,
|
|
stdout=completed.stdout or "",
|
|
stderr=completed.stderr or "",
|
|
)
|