Function _resolve_from_config

trinity/cli/view.py:20–62

Resolve (db_url, table_name, tokenizer_path) from a Trinity config file. Reads the experience pipeline's saved input. The viewer is only usable when the pipeline actually writes a SQL database, so requests are rejected when ``save_input`` is disabled or ``input_save_path`` is not a data

(config_path: str)

def _resolve_from_config(config_path: str) -> tuple[str, str, str]:
    """Resolve (db_url, table_name, tokenizer_path) from a Trinity config file.

    Reads the experience pipeline's saved input. The viewer is only usable when
    the pipeline actually writes a SQL database, so requests are rejected when
    ``save_input`` is disabled or ``input_save_path`` is not a database URL.
    """
    from trinity.buffer.storage.queue import is_database_url
    from trinity.common.config import load_config

    config = load_config(config_path)
    pipeline_cfg = config.data_processor.experience_pipeline

    if not pipeline_cfg.save_input:
        raise typer.BadParameter(
            f"Cannot view from {config_path}: "
            "data_processor.experience_pipeline.save_input is False, "
            "so no input database is produced."
        )
    save_path = pipeline_cfg.input_save_path
    if save_path is None:
        # save_input is on but no path given: the pipeline writes a SQLite DB
        # under the buffer cache dir. Resolve the same default so `view` works
        # without a full config validation pass.
        from trinity.buffer.pipelines.experience_pipeline import default_input_save_path

        save_path = default_input_save_path(config)
    if not is_database_url(save_path):
        raise typer.BadParameter(
            f"Cannot view from {config_path}: "
            "data_processor.experience_pipeline.input_save_path "
            f"({save_path!r}) is not a database URL, so no SQL database is produced."
        )

    tokenizer_path = config.model.model_path
    if not tokenizer_path:
        raise typer.BadParameter(f"Cannot view from {config_path}: model.model_path is not set.")

    _warn_if_renamed(config)

    # Mirrors ExperiencePipeline: the SQL input writer is created with table
    # name "pipeline_input" (see trinity/buffer/pipelines/experience_pipeline.py).
    return save_path, "pipeline_input", tokenizer_path


def _warn_if_renamed(config) -> None: