fix: rename secrets.py → vault.py to avoid shadowing stdlib secrets module

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/beanflows_pipeline/vault.py

@@ -0,0 +1,67 @@
+"""Secrets management via SOPS + age."""
+
+import subprocess
+from functools import lru_cache
+from pathlib import Path
+
+# Default secrets file path (relative to repo root)
+_DEFAULT_SECRETS_PATH = Path(__file__).parent.parent.parent / ".env.prod.sops"
+
+
+def _parse_dotenv(text: str) -> dict[str, str]:
+    """Parse dotenv-format text into a dict, skipping comments and blanks."""
+    result = {}
+    for line in text.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if "=" not in line:
+            continue
+        key, _, value = line.partition("=")
+        result[key.strip()] = value.strip()
+    return result
+
+
+@lru_cache(maxsize=1)
+def _load_environment(secrets_path: str = None) -> dict[str, str]:
+    """Decrypt and load secrets from a SOPS-encrypted dotenv file."""
+    path = Path(secrets_path) if secrets_path else _DEFAULT_SECRETS_PATH
+    assert path.exists(), f"Secrets file not found: {path}"
+    try:
+        result = subprocess.run(
+            ["sops", "--input-type", "dotenv", "--output-type", "dotenv", "--decrypt", str(path)],
+            capture_output=True,
+            text=True,
+            check=True,
+            timeout=30,
+        )
+        return _parse_dotenv(result.stdout)
+    except subprocess.CalledProcessError as e:
+        raise RuntimeError(f"Failed to decrypt secrets: {e.stderr.strip()}")
+    except FileNotFoundError:
+        raise RuntimeError(
+            "sops not found. Install with: brew install sops  "
+            "or see https://github.com/getsops/sops/releases"
+        )
+
+
+def get_secret(key: str, secrets_path: str = None) -> str | None:
+    """Get a secret value by key."""
+    env = _load_environment(secrets_path)
+    return env.get(key)
+
+
+def list_secrets(secrets_path: str = None) -> list[str]:
+    """List all available secret keys."""
+    env = _load_environment(secrets_path)
+    return list(env.keys())
+
+
+def test_connection(secrets_path: str = None) -> bool:
+    """Test that sops is available and can decrypt the secrets file."""
+    try:
+        _load_environment.cache_clear()
+        _load_environment(secrets_path)
+        return True
+    except Exception:
+        return False