Python OpenAI Structured JSON Tutorial

Gate of AI published a tutorial on building a Python command-line tool that uses OpenAI's chat model for structured JSON extraction from travel journals, with Pydantic validation. The tool, named travel_journal_analyzer, loads text, Markdown, or CSV files, extracts cities, restaurants, ratings, and sentiment, and writes validated JSON reports. The tutorial emphasizes combining JSON schema with local validation to ensure reliable, typed output for downstream applications.

🚀 Technical Briefing:This tutorial is part of our deep-dive series on Agentic Workflows at Gate of AI . For the full technical breakdown, interactive code sandbox, and the native Arabic translation, visit the original article here . Tutorial Build a Python command-line tool that loads travel-journal files, requests JSON-shaped extraction from an OpenAI chat model, validates every response with Pydantic, and writes a reusable report. This tutorial builds travel journal analyzer , a small but production-minded Python application. It accepts a text, Markdown, CSV file, or a directory containing those formats. It converts each source into a consistent JournalEntry , sends the entry to an OpenAI chat model, validates the returned JSON locally, and writes one JSON report for downstream software. The report has a deliberately narrow purpose: identify cities mentioned in an entry, explicitly named restaurants, dishes, ratings where stated, sentiment, practical tips, and a concise summary. The application does not treat a valid JSON shape as proof that a claim is true. Its prompt instructs the model to extract facts only from the supplied entry, while local validation checks the data contract before results are exported. Reliable structured extraction combines two boundaries. The first is the requested JSON schema: it defines the fields an application expects. The second is local validation with Pydantic: it rejects malformed values, such as a rating outside a five-point range. Research on structured generation describes the importance of reliable, typed output for applications that need predictable data rather than unconstrained prose. In practical Python work, these boundaries make testing and maintenance substantially easier. Create a project and isolated virtual environment: mkdir travel-journal-analyzer cd travel-journal-analyzer python -m venv .venv macOS or Linux source .venv/bin/activate Windows PowerShell .venv\Scripts\Activate.ps1 python -m pip install --upgrade pip python -m pip install "openai =1.0.0" "pydantic =2.7.0" "python-dotenv =1.0.1" "pytest =8.0.0" mkdir data output tests Create .env . Keep this file out of source control. API keys belong in environment variables locally and in a deployment secret manager in production. OPENAI API KEY=your-api-key OPENAI MODEL=your-chat-model MAX ENTRY CHARACTERS=12000 REQUEST TIMEOUT SECONDS=45 Create .gitignore : .env .venv/ pycache / .pytest cache/ output/ .pyc This tutorial uses the modern client-based OpenAI Python pattern: from openai import OpenAI , then client.chat.completions.create ... . Do not use legacy module-level completion calls. Create models.py . Nullable fields represent facts that the source did not provide. That is preferable to inventing a city, cuisine, or rating. python from future import annotations from typing import Literal from pydantic import BaseModel, Field, field validator class JournalEntry BaseModel : source name: str = Field min length=1, max length=255 entry id: str = Field min length=1, max length=100 text: str = Field min length=1 @field validator "text" @classmethod def validate text cls, value: str - str: cleaned = value.strip if not cleaned: raise ValueError "Journal entry text cannot be blank." return cleaned class RestaurantFinding BaseModel : name: str = Field min length=1, max length=200 city: str | None = Field default=None, max length=120 country: str | None = Field default=None, max length=120 cuisine: str | None = Field default=None, max length=120 dishes: list str = Field default factory=list rating out of five: float | None = Field default=None, ge=0, le=5 sentiment: Literal "positive", "neutral", "negative" recommendation reason: str = Field min length=1, max length=600 class JournalAnalysis BaseModel : entry id: str = Field min length=1, max length=100 cities mentioned: list str = Field default factory=list restaurants: list RestaurantFinding = Field default factory=list travel tips: list str = Field default factory=list concise summary: str = Field min length=1, max length=1000 class AnalysisReport BaseModel : generated at utc: str model: str total entries: int = Field ge=0 successful analyses: int = Field ge=0 failed entries: list str = Field default factory=list analyses: list JournalAnalysis = Field default factory=list The models are not merely documentation. JournalAnalysis.model validate json turns the model response into a validated object. A response with an invalid sentiment label or a six-point rating fails before it reaches a database, spreadsheet, or customer-facing interface. Create journal loader.py . Plain-text and Markdown files create one entry each. A CSV file requires a text column and creates one entry for every non-empty row. The character limit prevents unexpectedly large requests. python from future import annotations import csv from pathlib import Path from models import JournalEntry SUPPORTED SUFFIXES = {".txt", ".md", ".csv"} def load journal entries path value: str, max characters: int - list JournalEntry : path = Path path value .expanduser .resolve if not path.exists : raise FileNotFoundError f"Input path does not exist: {path}" if path.is dir : entries: list JournalEntry = for child in sorted path.iterdir : if child.is file and child.suffix.lower in SUPPORTED SUFFIXES: entries.extend load journal entries str child , max characters if not entries: raise ValueError "Directory contains no supported input files." return entries if path.suffix.lower in {".txt", ".md"}: text = path.read text encoding="utf-8" .strip check length text, path.name, max characters return JournalEntry source name=path.name, entry id=path.stem, text=text if path.suffix.lower = ".csv": raise ValueError "Use a .txt, .md, .csv file, or directory." entries = with path.open "r", encoding="utf-8-sig", newline="" as handle: reader = csv.DictReader handle if not reader.fieldnames or "text" not in reader.fieldnames: raise ValueError "CSV must contain a column named 'text'." for row number, row in enumerate reader, start=2 : text = row.get "text" or "" .strip if not text: continue check length text, f"{path.name} row {row number}", max characters entries.append JournalEntry source name= row.get "source name" or path.name .strip , entry id= row.get "entry id" or f"{path.stem}-{row number}" .strip , text=text, if not entries: raise ValueError "CSV contains no non-empty text rows." return entries def check length text: str, label: str, maximum: int - None: if not text: raise ValueError f"{label} is empty." if len text maximum: raise ValueError f"{label} exceeds the {maximum}-character limit." Save this sample as data/journal.txt : We arrived in Lisbon on a rainy Thursday. For dinner, we booked Taberna da Rua das Flores. The grilled octopus was tender, and I would rate the experience 4.5 out of 5. Reserve ahead because it is small. Two days later in Porto, Cafe Santiago served a memorable francesinha. The sandwich was rich but the service was slow, so I felt neutral overall. Create analyzer.py . The schema below is passed as a requested response format. Your configured model must support the selected response-format capability; verify that support in the current OpenAI documentation for the model available to your account before deploying. Regardless of the upstream response constraint, the code performs local Pydantic validation. python from future import annotations from openai import OpenAI from pydantic import ValidationError from models import JournalAnalysis, JournalEntry SCHEMA = { "name": "travel journal analysis", "strict": True, "schema": { "type": "object", "additionalProperties": False, "properties": { "entry id": {"type": "string"}, "cities mentioned": {"type": "array", "items": {"type": "string"}}, "restaurants": {"type": "array", "items": {"type": "object", "additionalProperties": False, "properties": { "name": {"type": "string"}, "city": {"type": "string", "null" }, "country": {"type": "string", "null" }, "cuisine": {"type": "string", "null" }, "dishes": {"type": "array", "items": {"type": "string"}}, "rating out of five": {"type": "number", "null" }, "sentiment": {"type": "string", "enum": "positive", "neutral", "negative" }, "recommendation reason": {"type": "string"} }, "required": "name", "city", "country", "cuisine", "dishes", "rating out of five", "sentiment", "recommendation reason" }}, "travel tips": {"type": "array", "items": {"type": "string"}}, "concise summary": {"type": "string"} }, "required": "entry id", "cities mentioned", "restaurants", "travel tips", "concise summary" } } SYSTEM PROMPT = """Extract facts only from the journal entry. Include restaurants only when explicitly named. Never invent facts. Use null for unknown scalar values. Return only schema-conforming JSON.""" class JournalAnalyzer: def init self, api key: str, model: str, timeout: float - None: self.model = model self.client = OpenAI api key=api key, timeout=timeout, max retries=0 def analyze self, entry: JournalEntry - JournalAnalysis: completion = self.client.chat.completions.create model=self.model, temperature=0, response format={"type": "json schema", "json schema": SCHEMA}, messages= {"role": "system", "content": SYSTEM PROMPT}, {"role": "user", "content": f"Required entry id: {entry.entry id}\n\nJournal:\n{entry.text}"}, , content = completion.choices 0 .message.content if not content: raise RuntimeError "The model returned an empty response." try: result = JournalAnalysis.model validate json content except ValidationError as error: raise RuntimeError f"Local schema validation failed: {error}" from error if result.entry id = entry.entry id: raise RuntimeError "Response entry id does not match the input entry." return result The extraction policy is intentionally conservative. An unnamed kiosk may be discussed negatively in a journal, but it must not become an invented restaurant record. If unnamed venues matter to the product, add a distinct field and clearly communicate that it represents an unnamed mention, not an official venue identity. python from future import annotations import argparse import os from datetime import datetime, timezone from pathlib import Path from dotenv import load dotenv from analyzer import JournalAnalyzer from journal loader import load journal entries from models import AnalysisReport def main - int: parser = argparse.ArgumentParser parser.add argument "input path" parser.add argument "--output", default="output/travel-analysis.json" args = parser.parse args load dotenv api key = os.getenv "OPENAI API KEY", "" .strip model = os.getenv "OPENAI MODEL", "" .strip if not api key or not model: raise RuntimeError "Set OPENAI API KEY and OPENAI MODEL in .env or your shell." maximum = int os.getenv "MAX ENTRY CHARACTERS", "12000" timeout = float os.getenv "REQUEST TIMEOUT SECONDS", "45" entries = load journal entries args.input path, maximum analyzer = JournalAnalyzer api key, model, timeout analyses = failed = for entry in entries: try: analyses.append analyzer.analyze entry except RuntimeError as error: failed.append entry.entry id print f"Failed {entry.entry id}: {error}" report = AnalysisReport generated at utc=datetime.now timezone.utc .isoformat , model=model, total entries=len entries , successful analyses=len analyses , failed entries=failed, analyses=analyses, destination = Path args.output destination.parent.mkdir parents=True, exist ok=True temporary = destination.with suffix destination.suffix + ".tmp" temporary.write text report.model dump json indent=2 , encoding="utf-8" temporary.replace destination print f"Wrote {report.successful analyses}/{report.total entries} analyses to {destination}" return 1 if failed else 0 if name == " main ": raise SystemExit main Run the tool and inspect the output: python main.py data/journal.txt --output output/journal-report.json python -m json.tool output/journal-report.json The temporary-file replacement avoids leaving a partially written final report if the process stops during serialization. The process returns a non-zero status when one or more entries fail, which is useful in scheduled jobs and CI. Test deterministic components without calling an API. For example, verify that the loader rejects a CSV without a text column, that blank rows are skipped, and that ratings over five fail Pydantic validation. Mock the OpenAI client for analyzer unit tests, then run a small intentional smoke test with a real credential. python from models import RestaurantFinding from pydantic import ValidationError import pytest def test invalid rating is rejected : with pytest.raises ValidationError : RestaurantFinding name="Example", sentiment="positive", rating out of five=6, recommendation reason="Unsupported rating range.", Schema validity is not factual validity. Review representative outputs for named-entity grounding, rating fidelity, correct sentiment, and unsupported details. Maintain a permissioned evaluation set when changing prompts, model configuration, schemas, or SDK versions. Avoid logging raw journal content by default because travel notes can contain personal data. For GCC and Middle East deployments, apply the same data-handling discipline to Arabic and English input, local privacy obligations, data residency requirements, and organizational retention rules. This tutorial does not make claims about a particular regional provider, initiative, or infrastructure arrangement; teams should validate those requirements with their legal, security, and platform stakeholders before production use. The central lesson is simple: use typed contracts and controlled inputs to turn flexible model output into application data, then test both the software boundary and the factual usefulness of the result.