API Reference

class Runner:
    """
    High-level orchestration class that:
    1. Loads configuration (databases, queries, polygons, checks, output)
    2. Resolves DB credentials
    3. Loads SQL
    4. Loads cached polygons
    5. Fetches data from the selected database
    6. Optionally checks for duplicates
    7. Performs configured checks
    8. Applies [output] column selection and returns the final table
    """

    def __init__(self, config_path: Path):
        """
        Initialize the Runner with a path to the TOML configuration file.

        Parameters
        ----------
        config_path : Path
            Path to the TOML configuration file.
        """
        self._cm = ConfigManager(Path(config_path))
        ConfigValidator(self._cm).validate()
        self._dbm = DatabaseManager(self._cm)

    # Helpers
    def list_queries(self) -> list[str]:
        """
        Return the list of query names defined in [[queries]].
        """
        return [q.get("name", "") for q in self._cm.get_queries()]

    def list_checks(self) -> list[str]:
        """
        Return the list of check names defined in [[checks]].
        """
        return [c.get("name", "") for c in self._cm.get_checks()]

    # Main runner
    def run(
        self,
        *,
        query_name: str | None = None,
        multi_query: bool = True,
        perform_duplicates: bool = True,
        perform_checks: bool = True,
        sql_text_override: str | None = None,
        sql_params: Mapping[str, Any] | None = None,
    ) -> RunResult:
        """
        Run the full pipeline:

        0. Load settings (via ConfigManager)
        1. Load credentials (via DatabaseManager)
        2. Load SQL (via io.sql.load_sql) or override with a wrapped SQL string
        3. Load cached polygons
        4. Fetch data from one or more databases
        5. Check for duplicates, if requested
        6. Perform checks, if configured
        7. Apply [output] column selection and return the final table

         CLI and frontend modules can extend or override behavior by:
        - Supplying a wrapped SQL string and params instead of the base SQL.
        - Toggling perform_duplicates / perform_checks flags.
        - Adjusting output selection or adding extra post-processing.

        Parameters
        ----------
        query_name : str or None, default=None
            Name of the query to run. If None and multi_query is set to False, ConfigManager.select_query() will choose
            a single active query if unambiguous. If multi_query is True, it is possible use this parameter to select a
            specific query.
        multi_query : bool, default=True
            Bool variable to set whether to run all non-skipped queries or a single query. Defaults to True
            If True, run every non-skipped query and concatenate the results. When more than one [[database]] profile
            is configured, every active [[queries]] entry must include a nonempty `database` field matching a configured
            profile name. With exactly one database profile, `database` is optional.
        perform_duplicates : bool, default=True
            If True, run duplicate detection and annotate the events table. If False, skip duplicate detection.
        perform_checks : bool, default=True
            If True, run configured checks and annotate the events table.
        sql_text_override : str or None, default=None
            If provided, this SQL string is executed instead of the base SQL from the config.
            Intended for CLI/frontend wrapping or custom runs.
        sql_params : Mapping[str, Any] or None, default=None
            Optional parameters dict passed to pandas.read_sql via DatabaseManager.fetch_events.
            CLI/frontend can use this to apply time/author filters safely.

        Returns
        -------
        RunResult
            A dataclass containing:
            - total_events: pd.DataFrame of all fetched events (before duplicates/checks)
            - duplicates: pd.DataFrame of detected duplicates (if perform_duplicates is True)
            - checks: pd.DataFrame of events after checks (if perform_checks is True)
            - output: pd.DataFrame of final output after [output] column selection
        """
        # 0. Load settings from the TOML configuration file
        event_type_col = self._cm.get_event_type_column()
        # time_col = self._cm.get_time_column()
        # author_col = self._cm.get_author_column()

        # 1. Load polygons (cached polygons for spatial checks)
        polygons_cfg = self._cm.get_polygons()
        polygon_cache = load_polygons(polygons_cfg, base_dir=self._cm.base_dir)

        # 2. Fetch events (single or multiple queries)
        # If multi_query is set and query_name is None, fetch all non-skipped queries and concatenate results
        if multi_query and not query_name:
            events_df = self._fetch_all_queries(
                sql_text_override=sql_text_override,
                sql_params=sql_params,
            )
        else:
            events_df = self._fetch_single_query(
                query_name=query_name,
                sql_text_override=sql_text_override,
                sql_params=sql_params,
            )

        # 3. Check for duplicates, if requested
        if perform_duplicates:  # Check this for multiple [[duplicates]] settings
            duplicates_cfg = self._cm.config_data.get("duplicates", [])
            if isinstance(duplicates_cfg, list) and duplicates_cfg:
                dup_df = run_duplicates(
                    events=events_df,
                    duplicates_cfg=duplicates_cfg,
                    event_type_col=event_type_col
                )
            else:
                dup_df = events_df.iloc[0:0].copy()
        else:
            dup_df = events_df.iloc[0:0].copy()

        # 4. Perform checks, if configured and enabled
        if perform_checks:
            checks_cfg = self._cm.get_checks()
            if checks_cfg:
                flagged_df = run_checks(
                    events=events_df,
                    checks=checks_cfg,
                    polygon_cache=polygon_cache,
                    event_type_col=event_type_col
                )
            else:
                flagged_df = events_df.iloc[0:0].copy()
        else:
            flagged_df = events_df.iloc[0:0].copy()

        # 5. Apply [output] column selection and return final table
        output_df = self._apply_output_selection(events=events_df, checks=flagged_df, dups=dup_df)

        # 6. Save results to disk if [output] section is defined
        output_cfg = self._cm.config_data.get("output", {})

        if output_cfg.get("save", False):
            self._save_output(output_df)

        return RunResult(
            total_events=events_df,
            duplicates=dup_df,
            checks=flagged_df,
            output=output_df,
        )

    def update_from_cache(
            self,
            events_df: pd.DataFrame,
            reload_config: bool = True
    ) -> RunResult:
        """
        Re-run duplicates, polygons, and checks on an existing DataFrame without re-querying the database.
        Useful for applying TOML configuration updates quickly.

        Parameters
        ----------
        events_df : pd.DataFrame
            The cached events DataFrame (usually from a previous RunResult.total_events).
        reload_config : bool, default=True
            If True, reloads the TOML file to pick up any user changes to checks, duplicates, or output.
        """
        # 0. Reload configuration to capture TOML updates
        if reload_config:
            self._cm = ConfigManager(self._cm.config_path)
            ConfigValidator(self._cm).validate()
            # We don't necessarily need to reload DatabaseManager since we are skipping fetching

        event_type_col = self._cm.get_event_type_column()

        # 1. Reload polygons (in case spatial paths or parameters changed)
        polygons_cfg = self._cm.get_polygons()
        polygon_cache = load_polygons(polygons_cfg, base_dir=self._cm.base_dir)

        # 2. Re-evaluate duplicates
        duplicates_cfg = self._cm.config_data.get("duplicates", [])
        if isinstance(duplicates_cfg, list) and duplicates_cfg:
            dup_df = run_duplicates(
                events=events_df,
                duplicates_cfg=duplicates_cfg,
                event_type_col=event_type_col
            )
        else:
            dup_df = events_df.iloc[0:0].copy()

        # 3. Re-evaluate checks
        checks_cfg = self._cm.get_checks()
        if checks_cfg:
            flagged_df = run_checks(
                events=events_df,
                checks=checks_cfg,
                polygon_cache=polygon_cache,
                event_type_col=event_type_col
            )
        else:
            flagged_df = events_df.iloc[0:0].copy()

        # 4. Apply output selection based on new configurations
        output_df = self._apply_output_selection(events=events_df, checks=flagged_df, dups=dup_df)

        # 5. Save output if configured in the updated TOML
        output_cfg = self._cm.config_data.get("output", {})
        if output_cfg.get("save", False):
            self._save_output(output_df)

        # Return a fresh RunResult
        return RunResult(
            total_events=events_df,
            duplicates=dup_df,
            checks=flagged_df,
            output=output_df,
        )

    # Internal helpers
    def _fetch_single_query(
            self,
            *,
            query_name: str | None,
            sql_text_override: str | None,
            sql_params: Mapping[str, Any] | None,
    ) -> pd.DataFrame:
        """
        Fetch events for a single query (one [[queries]] entry).
        """
        query_cfg = self._cm.select_query(name=query_name)

        if sql_text_override is not None:
            sql_text = sql_text_override
        else:
            sql_text = load_sql(
                query_cfg=query_cfg,
                base_dir=self._cm.base_dir,
                config_name=self._cm.config_path.name,
            )

        events_df = self._dbm.fetch_events(
            query_cfg=query_cfg,
            sql_text=sql_text,
            params=sql_params,
            read_sql_kwargs={},
        )
        return events_df

    def _fetch_all_queries(
            self,
            *,
            sql_text_override: str | None,
            sql_params: Mapping[str, Any] | None,
    ) -> pd.DataFrame:
        """
        Fetch events for all non-skipped [[queries]] entries and concatenate them.

        Requirements:
        - Every active query must return the same canonical output columns.
        - With multiple database profiles, every active query must explicitly declare `database`.
        - With one database profile, `database` is optional; an unnamed sole profile resolves to DEFAULT_DATABASE_NAME.
        """
        query_cfgs = self._cm.get_queries()
        active_queries: list[Mapping[str, Any]] = [
            q for q in query_cfgs if not q.get("skip", False)
        ]

        # If no active queries: return empty table
        if not active_queries:
            return pd.DataFrame()

        frames: list[pd.DataFrame] = []

        for query_cfg in active_queries:
            # Use the override if provided, otherwise load from file
            if sql_text_override is not None:
                sql_text = sql_text_override
            else:
                sql_text = load_sql(
                    query_cfg=query_cfg,
                    base_dir=self._cm.base_dir,
                    config_name=self._cm.config_path.name,
                )

            # Pass the sql_params instead of None
            df = self._dbm.fetch_events(
                query_cfg=query_cfg,
                sql_text=sql_text,
                params=sql_params,
                read_sql_kwargs={},
            )

            frames.append(df)

        # Concatenate all frames; ignore indices (reindex after concat)
        combined = pd.concat(frames, axis=0, ignore_index=True)

        # Try to sort by time-column
        try:
            time_col = self._cm.get_time_column()

            if time_col not in combined.columns:
                raise ValueError(
                    f"Configured [settings].time_column {time_col!r} is missing from the combined query result. "
                    f"Ensure every active SQL query returns this column with the same alias."
                )

            combined[time_col] = pd.to_datetime(
                combined[time_col],
                errors="raise",
                utc=True,
            )

            return combined.sort_values(
                by=time_col,
                kind="stable",
                ignore_index=True,
            )
        except Exception as e:
            warnings.warn(f"Error occurred while sorting by time column: {e}. Returning unsorted DataFrame.")
            return combined

    def _apply_output_selection(self, events: pd.DataFrame, checks: pd.DataFrame, dups:pd.DataFrame) -> pd.DataFrame:
        """
        Apply [output] column selection from the TOML configuration.

        If no [output] section is defined, return the full DataFrame.
        """
        cfg_data = self._cm.config_data
        output_cfg = cfg_data.get("output")

        joined_table = self._build_output_table(events=events, duplicates=dups, checks=checks)
        if output_cfg is None:
            return joined_table

        if not isinstance(output_cfg, dict):
            raise TypeError(
                "Expected [output] to be a single table in the TOML config."
            )

        columns = output_cfg.get("columns")
        if not columns:
            return joined_table

        missing = [c for c in columns if c not in joined_table.columns]
        if missing:
            raise ValueError(
                f"Configured [output].columns {missing!r} are missing from the events DataFrame. "
                f"Check your SQL SELECT list and [output] configuration."
            )
        selected_columns = list(columns)

        if "Observations" not in selected_columns:
            selected_columns.append("Observations")

        return joined_table[selected_columns]


    def _build_output_table(
        self,
        events: pd.DataFrame,
        duplicates: pd.DataFrame,
        checks: pd.DataFrame,
    ) -> pd.DataFrame:
        """Return one review row per event with merged observations."""
        result_frames = [
            frame[["Observations"]]
            for frame in (duplicates, checks)
            if not frame.empty
        ]

        if not result_frames:
            review_df = events.iloc[0:0].copy()
            review_df["Observations"] = pd.Series(dtype="object")
            return review_df

        all_observations = pd.concat(result_frames, axis=0)

        merged_observations = (
            all_observations.groupby(level=0, sort=True)["Observations"]
            .agg(lambda messages: "; ".join(dict.fromkeys(messages)))
        )

        review_df = events.loc[merged_observations.index].copy()
        review_df["Observations"] = merged_observations

        return review_df.reset_index(drop=True)

    def _save_output(self, output: pd.DataFrame) -> Path:
        """Save a completed review table using the [output] settings."""
        output_cfg = self._cm.config_data.get("output", {})

        file_format = output_cfg.get("file_format", "csv").lower()
        default_name = f"output.{file_format}"

        raw_path = output_cfg.get("file_path", default_name)
        path = Path(raw_path)

        if not path.is_absolute():
            path = Path.cwd() / path

        path.parent.mkdir(parents=True, exist_ok=True)

        writers = {
            "csv": lambda: output.to_csv(path, index=False),
            "parquet": lambda: output.to_parquet(path, index=False),
            "json": lambda: output.to_json(
                path,
                orient="records",
                date_format="iso",
                indent=2,
            ),
            "excel": lambda: output.to_excel(path, index=False),
            "feather": lambda: output.to_feather(path),
        }

        try:
            writers[file_format]()
        except KeyError as exc:
            raise ValueError(
                f"Unsupported output format: {file_format!r}."
            ) from exc

        return path

class ConfigManager:
    """
    Load and expose structured access to the TOML configuration file.

    This class is responsible for:
    - Reading the TOML file once.
    - Tracking the base directory (for resolving relative paths).
    - Providing accessors for [[database]], [[queries]], [[polygons]] and [[checks]].
    - Selecting a single query entry based on name and skip flags.
    """
    def __init__(self, config_path: Path):
        """
        Initialize the ConfigManager with a TOML configuration file.

        Parameters
        ----------
        config_path : Path
            Path to the TOML configuration file.
        """
        config_path = Path(config_path)
        if not config_path.exists():
            raise FileNotFoundError(f"{config_path} does not exist")
        if not config_path.is_file():
            raise ValueError(f"{config_path} is not a file")
        if config_path.suffix.lower() != ".toml":
            warnings.warn(
                f"Expected a .toml file but got {config_path.suffix!r}. "
                "Attempting to parse anyway.",
                stacklevel=2,
            )
        self._config_path: Path = config_path.resolve()
        self._base_dir: Path = config_path.parent
        with self._config_path.open("rb") as f:
            self._config_data: dict[str, Any] = tomllib.load(f)


    # Properties
    @property
    def config_path(self) -> Path:
        """Return the path to the TOML configuration file."""
        return self._config_path

    @property
    def base_dir(self) -> Path:
        """Return the base directory of the TOML configuration file."""
        return self._base_dir

    @property
    def config_data(self) -> dict[str, Any]:
        """Return the configuration data of the TOML configuration file."""
        return self._config_data


    # Getter methods
    def get_database(self) -> list[dict]:
        """
        Return the list of database profiles defined in [[database]].

        Returns
        -------
        list[dict]
            Each dict is one [[database]] entry. Returns an empty list
            if no databases are defined.
        """
        databases = self._config_data.get("database", [])

        if databases is None:
            return []

        if not isinstance(databases, list):
            raise TypeError(
                "Expected 'database' to be an array of tables "
                f"([[database]]), but got {type(databases).__name__!r}."
            )

        if not all(isinstance(database, dict) for database in databases):
            raise TypeError(
                "Each [[database]] entry must be a TOML table."
            )

        return databases

    def get_queries(self) -> list[dict]:
        """
        Return the list of queries defined in [[queries]].

        Returns
        -------
        list[dict]
            Each dict is one [[queries]] entry. Returns an empty list
            if no queries are defined.
        """
        queries = self._config_data.get("queries", [])

        if queries is None:
            return []

        if not isinstance(queries, list):
            raise TypeError(
                "Expected 'queries' to be an array of tables "
                f"([[queries]]), but got {type(queries).__name__!r}."
            )

        if not all(isinstance(query, dict) for query in queries):
            raise TypeError(
                "Each [[queries]] entry must be a TOML table."
            )

        return queries

    def get_checks(self) -> list[dict]:
        """
        Return the list of checks defined in [[checks]].

        Returns
        -------
        list[dict]
            Each dict is one [[checks]] entry. Returns an empty list
            if no checks are defined.
        """
        checks = self._config_data.get("checks", [])
        if checks is None:
            return []
        if not isinstance(checks, list):
            raise TypeError(
                "Expected 'checks' to be an array of tables ([[checks]]), "
                f"got {type(checks).__name__!r}."
            )
        if not all(isinstance(check, dict) for check in checks):
            raise TypeError(
                "Each [[checks]] entry must be a TOML table."
            )

        return checks

    def get_polygons(self) -> list[dict]:
        """
        Return the list of polygons defined in [[polygons]].

        Returns
        -------
        list[dict]
            Each dict is one [[polygons]] entry. Returns an empty list
            if no polygons are defined.
        """
        polys = self._config_data.get("polygons", [])
        if polys is None:
            return []
        if not isinstance(polys, list):
            raise TypeError(
                "Expected 'polygons' to be an array of tables ([[polygons]]), "
                f"got {type(polys).__name__!r}."
            )
        return polys


    # Query selection
    def select_query(self, name: str | None = None) -> dict:
        """
        Select a single query configuration from [[queries]]. Used for handle multiple cases in [[queries]] semantics.

        Selection rules:
        - Queries with skip=true are ignored.
        - If `name` is provided, the query with that name must exist
          and must not be skipped; otherwise a ValueError is raised.
        - If `name` is None and there is exactly one active (non-skipped)
          query, that query is returned.
        - If `name` is None and there are zero active queries, a ValueError
          is raised.
        - If `name` is None and there are multiple active queries, a ValueError
          is raised asking the caller to specify a name.

        Parameters
        ----------
        name : str or None, default=None
            Name of the query to select. If None, selection falls back
            to the single active query if exactly one exists.

        Returns
        -------
        dict
            The selected [[queries]] entry.

        Raises
        ------
        ValueError
            If no suitable query can be selected.
        """
        queries = self.get_queries()
        active = [q for q in queries if not q.get("skip", False)]

        if name is not None:
            if not isinstance(name, str) or not name.strip():
                raise ValueError("Query name must be a nonempty string when provided.")

            requested_name = name.strip()

            for q in active:
                if q.get("name") == requested_name:
                    return q

            raise ValueError(
                f"No active query named {requested_name!r} found in [[queries]] "
                f"of {self._config_path.name!r}."
            )

        # name is None: try to infer
        if not active:
            raise ValueError(
                f"No active queries defined in [[queries]] section of {self._config_path.name!r}."
            )

        if len(active) == 1:
            return active[0]

        names = [q.get("name", "<unnamed>") for q in active]

        raise ValueError(
            "Multiple active queries defined in [[queries]]; "
            "please specify which one to use by name. "
            f"Available names: {', '.join(repr(n) for n in names)}."
        )


    # Helpers
    def default_database_name(self) -> str | None:
        """
        Return the default database profile name if unambiguous.

        Rules:
        - Zero or multiple [[database]] profiles return None.
        - A sole unnamed profile returns DEFAULT_DATABASE_NAME.
        - A sole named profile returns its stripped explicit name.
        - An empty or invalid explicit name returns None.

        Returns
        -------
        str or None
        """
        databases = self.get_database()

        if len(databases) != 1:
            return None

        raw_name = databases[0].get("name")

        if raw_name is None:
            return DEFAULT_DATABASE_NAME

        if not isinstance(raw_name, str):
            return None

        profile_name = raw_name.strip()

        if not profile_name:
            return None

        return profile_name


    # Settings accessors
    def _get_settings(self) -> dict[str, Any]:
        """Return [settings], or an empty mapping when it is absent."""
        settings = self._config_data.get("settings", {})

        if settings is None:
            return {}

        if not isinstance(settings, dict):
            raise TypeError(
                "Expected [settings] to be a TOML table, "
                f"but got {type(settings).__name__!r}."
            )

        return settings


    def _get_setting_string(
            self,
            key: str,
            default: str,
    ) -> str:
        """Read one string-valued setting, falling back to its default."""
        value = self._get_settings().get(key, default)

        if not isinstance(value, str):
            raise TypeError(
                f"Expected [settings].{key} to be a string, "
                f"but got {type(value).__name__!r}."
            )

        value = value.strip()

        if not value:
            raise ValueError(
                f"[settings].{key} must be a nonempty string."
            )

        return value


    def get_event_type_column(self) -> str:
        """
        Return the event-type column name.

        Defaults to 'event_type'.
        """
        return self._get_setting_string(
            key="event_type_column",
            default="event_type",
        )


    def get_time_column(self) -> str:
        """
        Return the event-time column name.

        Defaults to 'time_value'.
        """
        return self._get_setting_string(
            key="time_column",
            default="time_value",
        )

    def get_author_column(self) -> str:
        """
        Return the author column name.

        Defaults to 'creationInfo_author'.
        """
        return self._get_setting_string(
            key="author_column",
            default="creationInfo_author",
        )

class DatabaseManager:
    """
    Manage database credentials and DB selection for queries.

    This class is responsible for:
    - Resolve credentials for each [[database]] profile
    - Decide which database profile to use for a given [[queries]] entry
    - Fetch data via SQL queries, applying time filters if required
    """
    def __init__(self, config_manager: ConfigManager):
        """
        Initialize the DatabaseManager from a ConfigManager. A single [[database]] entry may omit `name`
        It is stored internally under DEFAULT_DATABASE_NAME. Multiple profiles must each have a unique,
        nonempty name.

        Parameters
        ----------
        config_manager : ConfigManager
            Already-initialized ConfigManager pointing to the TOML file.
        """
        self._config_manager = config_manager
        self._base_dir: Path = config_manager.base_dir
        self._config_name: str = config_manager.config_path.name

        # Build credentials_by_name from [[database]] entries
        self._credentials_named: dict[str, dict[str, Any]] = {}

        db_cfgs = config_manager.get_database()

        if not db_cfgs:
            raise ValueError(f"No [[database]] entries found in {self._config_name!r}")

        multiple_databases = len(db_cfgs) > 1

        for index, db_cfg in enumerate(db_cfgs, start=1):
            profile_name = self._resolve_profile_name(
                db_cfg=db_cfg,
                index=index,
                multiple_databases=multiple_databases,
            )

            if profile_name in self._credentials_named:
                raise ValueError(
                    f"[[database]] entry #{index} in {self._config_name!r} "
                    f"has duplicate name {profile_name!r}. Database names must be unique."
                )

            credentials = load_credentials(
                db_cfg=db_cfg,
                base_dir=self._base_dir,
                config_name=self._config_name,
            )

            self._credentials_named[profile_name] = credentials


    def _resolve_profile_name(
            self,
            *,
            db_cfg: Mapping[str, Any],
            index: int,
            multiple_databases: bool,
    ) -> str:
        """
        Return the internal routing name for one [[database]] entry.

        With one database, an omitted name is represented internally by DEFAULT_DATABASE_NAME. With multiple databases,
        every entry needs an explicit name so queries can select a database unambiguously.
        """
        raw_name = db_cfg.get("name")

        if raw_name is None:
            if multiple_databases:
                raise ValueError(
                    f"[[database]] entry #{index} in {self._config_name!r} is missing a 'name' field. A name is "
                    f"required when multiple database profiles are configured."
                )

            return DEFAULT_DATABASE_NAME

        if not isinstance(raw_name, str) or not raw_name.strip():
            raise ValueError(
                f"[[database]] entry #{index} in {self._config_name!r} must have a nonempty string 'name' field."
            )

        profile_name = raw_name.strip()

        if profile_name == DEFAULT_DATABASE_NAME:
            raise ValueError(
                f"[[database]] entry #{index} in {self._config_name!r} "
                f"cannot use reserved name {DEFAULT_DATABASE_NAME!r}."
            )

        return profile_name

    # Getter methods
    def get_credentials(self, db_name: str) -> dict[str, Any]:
        """
        Return the credentials associated with a given [[database]] profile.

        Parameters
        ----------
        db_name : str
            The name of the profile to query.

        Returns
        ---------
        dict[str, Any]
            Credentials with keys: host, port, user, password, database.

        Raises
        ---------
        KeyError
            If no credentials are found for the requested db_name.
        """
        try:
            return self._credentials_named[db_name]
        except KeyError:
            raise KeyError(f"No credentials found for database profile '{db_name!r}' in {self._config_name}")

    def get_db_name_for_query(self, query_cfg: Mapping[str, Any]) -> str:
        """
        Determine which database profile to use for a given [[queries]] entry.

        Routing:
        - If the query configuration specifies a "database" field, use that.
        - If the query configuration does not specify a "database" field, use ConfigManager.default_database_name()
        - If neither yields a valid name, raise a ValueError.

        Parameters
        ----------
        query_cfg : Mapping[str, Any]
            The configuration dictionary for a specific query. Parsed from [[queries]] entry.

        Returns
        -------
        str
            The name of the database profile to use.

        Raises
        ------
        ValueError
            If the query configuration does not specify a valid database.
        """
        configured_name = query_cfg.get("database")
        if configured_name is not None:
            if not isinstance(configured_name, str) or not configured_name.strip():
                raise ValueError(
                    f"Query configuration {query_cfg.get('name', '<unnamed>')!r} has an invalid 'database' field. "
                    "It must be a nonempty string."
                )
            db_name = configured_name.strip()
            if db_name not in self._credentials_named:
                raise ValueError(
                    f"Query configuration {query_cfg.get('name', '<unnamed>')!r} specifies database profile "
                    f"{db_name!r}, but no such profile exists in {self._config_name!r}."
                )
            return db_name

        if len(self._credentials_named) == 1:
            return next(iter(self._credentials_named))

        raise ValueError(
            f"Query configuration {query_cfg.get('name', '<unnamed>')!r} does not specify a database, "
            f"and there are multiple database profiles available in {self._config_name!r}. "
            f"Please specify a database explicitly."
        )

    def fetch_events(
            self,
            query_cfg: Mapping[str, Any],
            sql_text: str,
            params: Mapping[str, Any] | None = None,
            read_sql_kwargs: Mapping[str, Any] | None = None,
    ) -> pd.DataFrame:
        """
        Execute the given SQL query against the database selected for this query and return the resulting event table.

        This method assumes that any filtering (time range, author, etc.) has already been applied to `sql_text` by
        the caller (core pipeline, CLI wrapper, or frontend). It does not modify the SQL.

        Parameters
        ----------
        query_cfg : Mapping[str, Any]
            Parsed contents of one [[queries]] entry from the TOML file.
            Used only to determine which database profile to use and for error messages / logging.
        sql_text : str
            Final SQL string to execute. May be a base query or a wrapped query with WHERE clauses and ORDER BY.
        params : Mapping[str, Any] or None, default=None
            Optional parameter dict to pass to pandas.read_sql for safe value substitution
            (e.g., {"start": "...", "end": "...", "author": "..."}).
        read_sql_kwargs : Mapping[str, Any] or None, default=None
            Additional keyword arguments forwarded to pandas.read_sql, such as chunksize or dtype hints.

        Returns
        -------
        pd.DataFrame
            Table containing the fetched events.

        Raises
        ------
        ConnectionError
            If the database connection cannot be established.
        pymysql.MySQLError
            If the underlying MySQL driver raises an error during execution.
        """
        if read_sql_kwargs is None:
            read_sql_kwargs = {}

        # Determine which DB profile to use
        db_name = self.get_db_name_for_query(query_cfg)
        credentials = self.get_credentials(db_name)

        # Connect and execute query
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", UserWarning)
            try:
                conn = pymysql.connect(
                    host=credentials["host"],
                    user=credentials["user"],
                    password=credentials["password"],
                    database=credentials["database"],
                    port=int(credentials.get("port", 3306)),
                )
            except pymysql.MySQLError as ce:
                raise ConnectionError(
                    f"Failed to connect to database profile {db_name!r} with the provided credentials."
                )
            try:
                # Simple progress bar: one step representing the whole query
                with tqdm(
                        total=1,
                        desc=f"Querying database '{db_name}' for {query_cfg.get('name', '<unnamed>')}",
                        unit="query",
                        leave=False,
                        bar_format="{desc}",
                ) as pbar:
                    df = pd.read_sql(
                        sql_text,
                        conn,
                        params=params,
                        **read_sql_kwargs,
                    )
                    pbar.update(1)
            finally:
                conn.close()

        return df