ontocast.api.process_request

Shared /process and /process_unit request body parsing.

ParsedProcessRequest dataclass

Fields shared by /process and /process_unit after reading the body.

Source code in ontocast/api/process_request.py

@dataclass(frozen=True)
class ParsedProcessRequest:
    """Fields shared by ``/process`` and ``/process_unit`` after reading the body."""

    files_dict: dict[str, bytes]
    max_visits: int
    strip_provenance: bool
    ontology_user_instruction: str
    ontology_selection_user_instruction: str
    facts_user_instruction: str
    ontology_context_fixed_ontology_id: str
    render_mode: str | None
    llm_graph_format: str | None
    ontology_context_mode_value: OntologyContextMode
    target_sections: list[str] | None
    exclude_sections: list[str] | None
    summarize_sections: list[str] | None
    summary_max_sentences: int
    document_type_hint: str | None
    section_schema_id: str | None
    document_metadata: dict[str, object]

build_agent_state_from_parsed(parsed, *, server_config, resolved_tenant, resolved_project, max_chunks)

Construct AgentState after tenancy resolution and enum parsing.

Source code in ontocast/api/process_request.py

def build_agent_state_from_parsed(
    parsed: ParsedProcessRequest,
    *,
    server_config: ServerConfig,
    resolved_tenant: str,
    resolved_project: str,
    max_chunks: int | None,
) -> AgentState:
    """Construct ``AgentState`` after tenancy resolution and enum parsing."""
    render_mode_value = parse_render_mode_param(
        parsed.render_mode,
        server_config.render_mode,
    )
    llm_graph_format_value = parse_llm_graph_format_param(
        parsed.llm_graph_format,
        server_config.llm_graph_format,
    )
    return AgentState(
        raw_input=parsed.files_dict,
        max_visits=parsed.max_visits,
        max_chunks=max_chunks,
        render_mode=render_mode_value,
        llm_graph_format=llm_graph_format_value,
        ontology_context_mode=parsed.ontology_context_mode_value,
        tenant=resolved_tenant,
        project=resolved_project,
        ontology_user_instruction=parsed.ontology_user_instruction,
        ontology_selection_user_instruction=parsed.ontology_selection_user_instruction,
        facts_user_instruction=parsed.facts_user_instruction,
        ontology_context_fixed_ontology_id=parsed.ontology_context_fixed_ontology_id,
        target_sections=parsed.target_sections,
        exclude_sections=parsed.exclude_sections,
        summarize_sections=parsed.summarize_sections,
        summary_max_sentences=parsed.summary_max_sentences,
        document_type_hint=parsed.document_type_hint,
        section_schema_id=parsed.section_schema_id,
        document_metadata=dict(parsed.document_metadata),
    )

load_parsed_process_request(request, server_config, *, log_label='process') async

Read request parameters from query string, JSON body or multipart form.

All three transports are read through :data:_PARAM_SPECS, in precedence order (query, then body/form), so every parameter is honoured on every transport. The body branches differ only in how raw values are obtained -- a decoded JSON object, or the form's multi-items -- never in which parameters they support.

Source code in ontocast/api/process_request.py

async def load_parsed_process_request(
    request: Request,
    server_config: ServerConfig,
    *,
    log_label: str = "process",
) -> ParsedProcessRequest | JSONResponse:
    """Read request parameters from query string, JSON body or multipart form.

    All three transports are read through :data:`_PARAM_SPECS`, in precedence
    order (query, then body/form), so every parameter is honoured on every
    transport. The body branches differ only in how raw values are obtained --
    a decoded JSON object, or the form's multi-items -- never in which
    parameters they support.
    """
    content_type = request.headers.get("content-type") or ""
    logger.debug("%s Content-Type: %s", log_label, content_type)

    values: dict[str, Any] = {
        "render_mode": None,
        "llm_graph_format": None,
        "ontology_context_mode": None,
        "ontology_user_instruction": "",
        "ontology_selection_user_instruction": "",
        "facts_user_instruction": "",
        "ontology_context_fixed_ontology_id": "",
        "strip_provenance": False,
        "max_visits": server_config.max_visits_per_node,
        "summary_max_sentences": 5,
        "target_sections": None,
        "exclude_sections": None,
        "summarize_sections": None,
        "document_type_hint": None,
        "section_schema_id": None,
        "document_metadata": {},
    }

    _apply_source(values, dict(request.query_params))

    if content_type.startswith("application/json"):
        bytes_data = await request.body()
        logger.debug("%s JSON body length: %s", log_label, len(bytes_data))
        files_dict = {"input.json": bytes_data}
        try:
            parsed_obj = json.loads(bytes_data.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError):
            logger.debug(
                "%s JSON body could not be decoded for parameter extraction",
                log_label,
            )
        else:
            if isinstance(parsed_obj, dict):
                _apply_source(values, parsed_obj)
    elif content_type.startswith("multipart/form-data"):
        form = await request.form()
        files_dict = {}
        form_values: dict[str, Any] = {}
        for key, value in form.multi_items():
            if isinstance(value, StarletteUploadFile):
                files_dict[key] = await value.read()
            else:
                form_values[key] = str(value)
        _apply_source(values, form_values)
        if not files_dict:
            return JSONResponse(
                status_code=400,
                content=StatusErrorBody(
                    error="No file provided",
                    error_type="ValidationError",
                ).model_dump(),
            )
    else:
        return JSONResponse(
            status_code=400,
            content=StatusErrorBody(
                error=f"Unsupported content type: {content_type}",
                error_type="ValidationError",
            ).model_dump(),
        )

    ontology_context_mode_value: OntologyContextMode = (
        parse_ontology_context_mode_param(
            values["ontology_context_mode"],
            server_config.ontology_context_mode,
        )
    )

    fixed_ontology_id = values["ontology_context_fixed_ontology_id"]
    ontology_context_mode_value = resolve_ontology_context_mode(
        ontology_context_mode_value,
        fixed_ontology_id,
    )
    if (
        ontology_context_mode_value == OntologyContextMode.FIXED_SINGLE_ONTOLOGY
        and not fixed_ontology_id
    ):
        return missing_fixed_catalog_ontology_id_response()

    return ParsedProcessRequest(
        files_dict=files_dict,
        max_visits=values["max_visits"],
        strip_provenance=values["strip_provenance"],
        ontology_user_instruction=values["ontology_user_instruction"],
        ontology_selection_user_instruction=values[
            "ontology_selection_user_instruction"
        ],
        facts_user_instruction=values["facts_user_instruction"],
        ontology_context_fixed_ontology_id=fixed_ontology_id,
        render_mode=values["render_mode"],
        llm_graph_format=values["llm_graph_format"],
        ontology_context_mode_value=ontology_context_mode_value,
        target_sections=values["target_sections"],
        exclude_sections=values["exclude_sections"],
        summarize_sections=values["summarize_sections"],
        summary_max_sentences=values["summary_max_sentences"],
        document_type_hint=values["document_type_hint"],
        section_schema_id=values["section_schema_id"],
        document_metadata=values["document_metadata"],
    )