153 lines
5.7 KiB
Python
153 lines
5.7 KiB
Python
"""
|
||
Main entry point for the Structured Output Union events API task.
|
||
|
||
This script demonstrates how to parse a raw log containing multiple HTTP events
|
||
using Pydantic v2 models and LangChain structured output. It can be run as a
|
||
stand‑alone CLI or imported as a module.
|
||
|
||
Requirements (see requirements.txt):
|
||
- langchain-core>=1.0.0
|
||
- langchain-openai
|
||
- pydantic>=2.0
|
||
- python-dotenv
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import List, Union
|
||
|
||
from dotenv import load_dotenv
|
||
from langchain_core.output_parsers import PydanticOutputParser
|
||
from langchain_openai import ChatOpenAI
|
||
from pydantic import BaseModel, Field, ValidationError
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Models
|
||
# ---------------------------------------------------------------------------
|
||
class HttpOkEvent(BaseModel):
|
||
"""Represents a successful HTTP request."""
|
||
|
||
kind: str = Field("ok", description="Discriminator for the union.")
|
||
status: int = Field(200, description="HTTP status code (must be 200).")
|
||
path: str = Field(..., description="Requested URL path.")
|
||
duration_ms: int = Field(..., description="Duration of the request in milliseconds.")
|
||
|
||
class HttpErrorEvent(BaseModel):
|
||
"""Represents a failed HTTP request."""
|
||
|
||
kind: str = Field("error", description="Discriminator for the union.")
|
||
status: int = Field(..., ge=400, le=599, description="HTTP error status code.")
|
||
path: str = Field(..., description="Requested URL path.")
|
||
error_message: str = Field(..., description="Human readable error message.")
|
||
|
||
# Union with discriminator ``kind``. Pydantic v2 automatically uses the field
|
||
# named ``kind`` to decide which model to instantiate.
|
||
ApiEvent = Union[HttpOkEvent, HttpErrorEvent]
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Parser helper
|
||
# ---------------------------------------------------------------------------
|
||
parser = PydanticOutputParser(pydantic_object=ApiEvent)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LLM wrapper
|
||
# ---------------------------------------------------------------------------
|
||
load_dotenv() # Load JARVIS API key from .env if present.
|
||
llm = ChatOpenAI(
|
||
model="openai/gpt-oss-20b:free",
|
||
base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1",
|
||
api_key=os.getenv("JOURNAL_MCP_PAT"),
|
||
temperature=0.0,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Core logic
|
||
# ---------------------------------------------------------------------------
|
||
def parse_event(raw: str) -> ApiEvent:
|
||
"""Parse a single raw log line into an :class:`ApiEvent`.
|
||
|
||
The function sends the raw text to the LLM with a structured output prompt
|
||
and then validates the result using Pydantic. If validation fails, the
|
||
exception is propagated so that callers can decide how to handle it.
|
||
"""
|
||
# Build a simple prompt that instructs the model to return JSON matching
|
||
# one of the two event schemas.
|
||
prompt = f"Parse the following log line into JSON:\n{raw}\nJSON:" # noqa: E501
|
||
response = llm.invoke([prompt])
|
||
json_text = parser.parse(response.content)
|
||
try:
|
||
return ApiEvent.model_validate(json_text) # type: ignore[arg-type]
|
||
except ValidationError as exc: # pragma: no cover - defensive
|
||
raise ValueError(f"Failed to validate event: {exc}") from exc
|
||
|
||
|
||
def parse_log(raw_log: str) -> List[ApiEvent]:
|
||
"""Parse a multiline log into a list of :class:`ApiEvent` objects.
|
||
|
||
The function splits the input on newlines and ignores empty lines.
|
||
Each non‑empty line is parsed individually.
|
||
"""
|
||
events: List[ApiEvent] = []
|
||
for line in raw_log.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
event = parse_event(line)
|
||
events.append(event)
|
||
except Exception as exc: # pragma: no cover - log and skip
|
||
print(f"Warning: could not parse line '{line}': {exc}", file=sys.stderr)
|
||
return events
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI entry point
|
||
# ---------------------------------------------------------------------------
|
||
def main() -> None:
|
||
parser_cli = argparse.ArgumentParser(description="Parse raw HTTP logs into structured events.")
|
||
parser_cli.add_argument(
|
||
"--log-file",
|
||
type=Path,
|
||
help="Path to a file containing the raw log. If omitted, a sample log is used.",
|
||
)
|
||
args = parser_cli.parse_args()
|
||
|
||
if args.log_file and args.log_file.exists():
|
||
raw_log = args.log_file.read_text(encoding="utf-8")
|
||
else:
|
||
# Sample log with mixed success and error events.
|
||
raw_log = """
|
||
GET /api/users 200 OK duration=123ms
|
||
POST /api/login 404 Not Found: user not found
|
||
PUT /api/items/42 500 Internal Server Error: database timeout
|
||
GET /health 200 OK duration=45ms
|
||
"""
|
||
print("Using sample log:\n", raw_log)
|
||
|
||
events = parse_log(raw_log)
|
||
if not events:
|
||
print("No valid events parsed.")
|
||
sys.exit(1)
|
||
|
||
# Pretty‑print the results as a table.
|
||
header = f"{'Kind':<6} | {'Path':<20} | {'Status':<6} | Details"
|
||
print(header)
|
||
print("-" * len(header))
|
||
for ev in events:
|
||
if isinstance(ev, HttpOkEvent):
|
||
details = f"duration={ev.duration_ms}ms"
|
||
else: # HttpErrorEvent
|
||
details = f"error='{ev.error_message}'"
|
||
print(f"{ev.kind:<6} | {ev.path:<20} | {ev.status:<6} | {details}")
|
||
|
||
# Also output the raw Pydantic model dumps for debugging.
|
||
print("\nModel dumps:\n")
|
||
for ev in events:
|
||
print(ev.model_dump_json(indent=2))
|
||
|
||
if __name__ == "__main__": # pragma: no cover - entry point
|
||
main()
|