# Python OpenAI Structured JSON Tutorial

> Source: <https://dev.to/gateofai/python-openai-structured-json-tutorial-2k4a>
> Published: 2026-08-13 18:08:05+00:00

🚀 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.