mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 19:01:54 +00:00
kevinveenbirkenbachandClaude Opus 4.8
cf765ee00c
SPOT: declare the var-lib base once in default.env as INFINITO_DIR_VAR_LIB; the env build derives INFINITO_PLAYWRIGHT_REPORTS_BASE_DIR from it; reference it everywhere (utils/paths.py, scripts via ${...:?}, tests, the dev inventory's DIR_VAR_LIB injection) instead of hardcoding the literal. group_vars keeps the literal default; the env-lookup lives in the dev inventory per the no-ansible-env-lookups lint rule. pull_specific_host.py --folder is now required and the role passes {{ DIR_BACKUPS }}.
Swarm e2e: collect_playwright_reports.sh docker-cp's the reports off the manager plus a workflow upload step, so swarm Playwright traces reach CI artifacts (was compose-only).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
"""Strip databases.csv rows for purged application entities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from utils.paths import FILE_DATABASE_SECRETS
|
|
from utils.roles.entity_name import get_entity_name
|
|
|
|
DEFAULT_CSV = FILE_DATABASE_SECRETS
|
|
|
|
_CSV_DELIMITER = ";"
|
|
_HEADER_COLUMNS = ("instance", "database", "username", "password")
|
|
|
|
|
|
def _resolve_csv_file() -> Path:
|
|
env_path = os.environ.get("FILE_DATABASE_SECRETS")
|
|
return Path(env_path) if env_path else DEFAULT_CSV
|
|
|
|
|
|
def _row_matches(row: list[str], targets: set[str]) -> bool:
|
|
if len(row) < 3:
|
|
return False
|
|
database = row[1].strip()
|
|
username = row[2].strip()
|
|
return database in targets or username in targets
|
|
|
|
|
|
def _is_header(row: list[str]) -> bool:
|
|
if len(row) < len(_HEADER_COLUMNS):
|
|
return False
|
|
return all(
|
|
row[i].strip().lower() == _HEADER_COLUMNS[i]
|
|
for i in range(len(_HEADER_COLUMNS))
|
|
)
|
|
|
|
|
|
def wipe_database_entries(
|
|
app_ids: list[str], csv_file: Path | None = None
|
|
) -> list[str]:
|
|
"""Remove rows whose database or username column matches an entity
|
|
derived from *app_ids* via :func:`get_entity_name`."""
|
|
path = csv_file or _resolve_csv_file()
|
|
if not path.exists():
|
|
return []
|
|
|
|
targets: set[str] = set()
|
|
for app_id in app_ids:
|
|
entity = get_entity_name(app_id)
|
|
if entity:
|
|
targets.add(entity)
|
|
if not targets:
|
|
return []
|
|
|
|
with path.open(newline="", encoding="utf-8") as fh:
|
|
rows = list(csv.reader(fh, delimiter=_CSV_DELIMITER))
|
|
|
|
kept: list[list[str]] = []
|
|
removed: list[str] = []
|
|
for row in rows:
|
|
if not row:
|
|
kept.append(row)
|
|
continue
|
|
if _is_header(row):
|
|
kept.append(row)
|
|
continue
|
|
if _row_matches(row, targets):
|
|
removed.append(f"{row[1].strip()}:{row[2].strip()}")
|
|
continue
|
|
kept.append(row)
|
|
|
|
if removed:
|
|
with path.open("w", newline="", encoding="utf-8") as fh:
|
|
csv.writer(fh, delimiter=_CSV_DELIMITER).writerows(kept)
|
|
|
|
return removed
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if not argv:
|
|
print(
|
|
"usage: python -m utils.cleanup.databases_csv <APP_ID> [APP_ID ...]",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
removed = wipe_database_entries(argv)
|
|
if removed:
|
|
print(f">>> Wiped databases.csv entries: {', '.join(removed)}")
|
|
else:
|
|
print(">>> No databases.csv entries to wipe")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|