Validate Records in OpenSearch from a CSV (with Logging and Health Checks)

recordId,recordDate 00000000-0000-0000-0000-000000000001,2025-07-31 00000000-0000-0000-0000-000000000002,2025-07-31 

OPENSEARCH_URL=https://your-opensearch-host:9200 OPENSEARCH_USER=your-username OPENSEARCH_PASS=your-password INDEX_PATTERN=my_index_pattern_2025*/_search INPUT_CSV=./inputs/records.csv LOG_DIR=./logs OUTPUT_DIR=./outputs 

""" OpenSearch existence checks for (recordId, recordDate) pairs with a fast _cluster/health pre-check. What it does ------------ 1) Loads configuration strictly from a .env file (no hard-coded defaults). 2) Pre-checks OpenSearch availability via GET /_cluster/health (fails fast if unreachable or red). 3) Reads an input CSV with headers: recordId, recordDate (YYYY-MM-DD). 4) For each row, runs an existence query: - exact term match on recordId.keyword - range match on recordDate for the full day (00:00:00.000 -> 23:59:59.999) 5) Produces: - timestamped log file (PASS/FAIL per row) - timestamped results CSV (recordId, recordDate, status) Security note ------------- SSL verification is disabled here for convenience in internal/test setups. For production, enable certificate verification and use a trusted CA/cert chain. """ from __future__ import annotations import csv import json import logging import os import ssl from base64 import b64encode from dataclasses import dataclass from datetime import datetime from typing import Dict, Any from urllib.request import Request, urlopen, URLError, HTTPError from dotenv import load_dotenv # ---------------- Configuration ---------------- @dataclass(frozen=True) class Config: """Strongly-typed configuration loaded from environment.""" url: str user: str password: str index_pattern: str # e.g., "my_index_pattern_2025*/_search" input_csv: str # e.g., "./inputs/records.csv" log_dir: str # e.g., "./logs" output_dir: str # e.g., "./outputs" @staticmethod def from_env() -> "Config": load_dotenv() # only reads .env / process env; no defaults url = os.getenv("OPENSEARCH_URL") user = os.getenv("OPENSEARCH_USER") password = os.getenv("OPENSEARCH_PASS") index = os.getenv("INDEX_PATTERN") input_csv = os.getenv("INPUT_CSV") log_dir = os.getenv("LOG_DIR") output_dir = os.getenv("OUTPUT_DIR") missing = [k for k, v in { "OPENSEARCH_URL": url, "OPENSEARCH_USER": user, "OPENSEARCH_PASS": password, "INDEX_PATTERN": index, "INPUT_CSV": input_csv, "LOG_DIR": log_dir, "OUTPUT_DIR": output_dir, }.items() if not v] if missing: raise ValueError(f"Missing required variables in .env: {', '.join(missing)}") return Config(url, user, password, index, input_csv, log_dir, output_dir) # ---------------- Minimal OpenSearch client ---------------- class OpenSearchClient: """Tiny OpenSearch client using urllib (no external HTTP deps).""" def __init__(self, base_url: str, user: str, password: str, verify_ssl: bool = False) -> None: self.base_url = base_url.rstrip("/") self.auth_header = { "Authorization": "Basic " + b64encode(f"{user}:{password}".encode()).decode(), "Content-Type": "application/json", } self.ctx = ssl.create_default_context() if not verify_ssl: self.ctx.check_hostname = False self.ctx.verify_mode = ssl.CERT_NONE def get_json(self, endpoint: str) -> Dict[str, Any]: url = f"{self.base_url}/{endpoint.lstrip('/')}" req = Request(url, method="GET") # Only Authorization header on GET (no need for Content-Type) req.add_header("Authorization", self.auth_header["Authorization"]) with urlopen(req, context=self.ctx) as resp: payload = json.loads(resp.read().decode("utf-8")) logging.info("GET %s -> %s", endpoint, resp.status) return payload def post_json(self, endpoint: str, body: Dict[str, Any]) -> Dict[str, Any]: url = f"{self.base_url}/{endpoint.lstrip('/')}" req = Request(url, data=json.dumps(body).encode("utf-8"), method="POST") for k, v in self.auth_header.items(): req.add_header(k, v) with urlopen(req, context=self.ctx) as resp: payload = json.loads(resp.read().decode("utf-8")) logging.info("POST %s -> %s", endpoint, resp.status) return payload # ---------------- Health pre-check ---------------- def assert_cluster_healthy(client: OpenSearchClient) -> None: """ Fail fast if OpenSearch is not reachable or reports 'red' status. Accepts 'green' and 'yellow' as pass for read-only checks. """ try: health = client.get_json("_cluster/health") status = (health.get("status") or "").lower() if status not in {"green", "yellow"}: raise RuntimeError(f"Cluster health is '{status}', expected green/yellow.") logging.info("Cluster health OK: %s", status) except (HTTPError, URLError) as e: raise RuntimeError(f"Cluster health check failed: {e}") from e # ---------------- Existence query ---------------- def record_exists(client: OpenSearchClient, index_pattern: str, record_id: str, record_date: str) -> bool: """ Return True if any document exists for: - recordId.keyword == record_id - recordDate between 'record_date 00:00:00.000' and 'record_date 23:59:59.999' """ start = f"{record_date} 00:00:00.000" end = f"{record_date} 23:59:59.999" query = { "query": { "bool": { "must": [ {"term": {"recordId.keyword": record_id}}, {"range": {"recordDate": {"gte": start, "lte": end}}}, ] } }, "size": 0 } try: res = client.post_json(index_pattern, query) total = res.get("hits", {}).get("total", {}) value = total.get("value", total if isinstance(total, int) else 0) return value > 0 except Exception as e: logging.error("Query failed for recordId=%s date=%s: %s", record_id, record_date, e) return False # ---------------- Main ---------------- def main() -> None: cfg = Config.from_env() # Prepare paths + logging os.makedirs(cfg.log_dir, exist_ok=True) os.makedirs(cfg.output_dir, exist_ok=True) ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") log_file = os.path.join(cfg.log_dir, f"record_check_{ts}.log") out_file = os.path.join(cfg.output_dir, f"record_results_{ts}.csv") logging.basicConfig(filename=log_file, level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") # Initialize client client = OpenSearchClient(cfg.url, cfg.user, cfg.password, verify_ssl=False) # Pre-flight: cluster health assert_cluster_healthy(client) # Process CSV -> Results CSV with open(cfg.input_csv, newline="", encoding="utf-8") as fin, \ open(out_file, "w", newline="", encoding="utf-8") as fout: reader = csv.DictReader(fin) required_headers = {"recordId", "recordDate"} if not reader.fieldnames or required_headers - set(reader.fieldnames): raise ValueError(f"CSV must have headers: recordId,recordDate (got {reader.fieldnames})") writer = csv.writer(fout) writer.writerow(["recordId", "recordDate", "status"]) for row in reader: rec_id = (row.get("recordId") or "").strip() date = (row.get("recordDate") or "").strip() if not rec_id or not date: logging.warning("Skipping row with missing values: %s", row) continue ok = record_exists(client, cfg.index_pattern, rec_id, date) status = "PASS" if ok else "FAIL" writer.writerow([rec_id, date, status]) (logging.info if ok else logging.warning)("%s: recordId=%s date=%s", status, rec_id, date) print(f"Done.\nLog: {log_file}\nResults: {out_file}") if __name__ == "__main__": main()