tracer-cloud--opensre/integrations/rds/__init__.py

"""Shared AWS RDS integration helpers.

Provides configuration normalization, source detection, and parameter
extraction for the RDS investigation tools. All AWS API calls are
read-only and routed through the shared aws_sdk_client allowlist.
"""

from __future__ import annotations

import logging
from typing import Any

from config.strict_config import StrictConfigModel
from integrations._relational import env_str
from integrations._validation_helpers import report_classify_failure

logger = logging.getLogger(__name__)

DEFAULT_RDS_REGION = "us-east-1"


class RDSConfig(StrictConfigModel):
    """Normalized RDS connection settings."""

    db_instance_identifier: str = ""
    region: str = DEFAULT_RDS_REGION
    integration_id: str = ""

    @property
    def is_configured(self) -> bool:
        return bool(self.db_instance_identifier and self.region)


def build_rds_config(raw: dict[str, Any] | None) -> RDSConfig:
    """Build a normalized RDS config object from env/store data."""
    return RDSConfig.model_validate(raw or {})


def rds_config_from_env() -> RDSConfig | None:
    """Load an RDS config from env vars."""
    db_id = env_str("RDS_DB_INSTANCE_IDENTIFIER")
    if not db_id:
        return None
    return build_rds_config(
        {
            "db_instance_identifier": db_id,
            "region": env_str("AWS_REGION") or env_str("RDS_REGION") or DEFAULT_RDS_REGION,
        }
    )


def rds_is_available(sources: dict[str, dict]) -> bool:
    """Check if RDS integration identifying params are present.

    A scenario-injected ``_backend`` (FixtureAWSBackend in synthetic tests)
    counts on its own — synthetic scenarios always carry the DB identifier
    in scenario metadata, and we want the RDS tools to be selectable in
    synthetic mode regardless of whether the alert annotations also surfaced
    the identifier.
    """
    rds = sources.get("rds", {})
    return bool(rds.get("db_instance_identifier") or rds.get("_backend"))


def rds_extract_params(sources: dict[str, dict]) -> dict[str, Any]:
    """Extract RDS identifying params (db_instance_identifier, region).

    Forwards the optional synthetic ``_backend`` handle as ``aws_backend`` so
    the RDS tools can short-circuit to fixture data instead of hitting real
    boto3. Without this hop, ``execute_aws_sdk_call`` would silently leak to
    whatever AWS account the developer happens to be authenticated against
    during a synthetic test run.
    """
    rds = sources.get("rds", {})
    region = (
        str(rds.get("region") or "").strip()
        or env_str("AWS_REGION")
        or env_str("RDS_REGION")
        or DEFAULT_RDS_REGION
    )
    return {
        "db_instance_identifier": str(rds.get("db_instance_identifier", "")).strip(),
        "region": region,
        "aws_backend": rds.get("_backend"),
    }


def classify(credentials: dict[str, Any], record_id: str) -> tuple[RDSConfig | None, str | None]:
    try:
        cfg = build_rds_config(
            {
                "db_instance_identifier": credentials.get("db_instance_identifier", ""),
                "region": credentials.get("region", DEFAULT_RDS_REGION),
                "integration_id": record_id,
            }
        )
    except Exception as exc:
        report_classify_failure(exc, logger=logger, integration="rds", record_id=record_id)
        return None, None
    if cfg.is_configured:
        return cfg, "rds"
    return None, None