sql_qa_pair/app/llm/parser_manager.py

"""Registry for converting LLM responses via pluggable parsers."""

from __future__ import annotations

from pathlib import Path
from threading import Lock
from typing import Dict, Optional, Type

import yaml

from app.const import CONFIG_DIR
from .parsers import BaseParser, PARSER_REGISTRY


class ParserManager:
    """Lazy singleton keeping parser instances referenced by name."""

    _instance: Optional["ParserManager"] = None
    _lock = Lock()

    def __new__(cls, config_path: Optional[Path] = None) -> "ParserManager":
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._build(config_path)
        return cls._instance

    def _build(self, config_path: Optional[Path]) -> None:
        path = Path(config_path or (CONFIG_DIR / "parser_settings.yaml"))
        self._parsers: Dict[str, BaseParser] = {}

        self._register_defaults()
        if path.exists():
            self._load_custom_parsers(path)

    def _register_defaults(self) -> None:
        for name, parser_cls in PARSER_REGISTRY.items():
            self._parsers[name] = parser_cls()

    def _load_custom_parsers(self, path: Path) -> None:
        with path.open("r", encoding="utf-8") as fh:
            parser_settings = yaml.safe_load(fh) or {}
        for name, config in parser_settings.items():
            parser_type = config.get("type")
            if not parser_type or parser_type not in PARSER_REGISTRY:
                continue
            parser_cls: Type[BaseParser] = PARSER_REGISTRY[parser_type]
            self._parsers[name] = parser_cls(**config.get("options", {}))

    def get_parser(self, name: str) -> BaseParser:
        if name not in self._parsers:
            raise KeyError(f"Parser {name} is not registered")
        return self._parsers[name]

    def register_parser(self, name: str, parser: BaseParser) -> None:
        self._parsers[name] = parser