129. File I/O And JSON Reports

📄 Aryan wants to save each RAM session as a JSON report and load the last 7 days of history. His save function corrupts files on crash, uses the wrong path join method, and doesn’t create the output directory if it doesn’t exist.

💡 Fun fact: The atomic write pattern (write to .tmp, then rename) is used by SQLite, Firefox, and almost every database on the planet. The POSIX rename() syscall is guaranteed atomic — the OS either sees the old file or the new one, never a half-written mess. This pattern has been saving data integrity since Unix was born.

⚠️ Watch out: Path.glob() returns files in an arbitrary, OS-defined order — it is NOT alphabetical, NOT chronological, and NOT consistent between runs. Always call sorted() on the result. For ISO date filenames like session_2024-01-15.json, alphabetical sort IS chronological sort, which is exactly why we name files that way.

🤔 Think about it: Why is Path.home() / '.ram_manager' / 'reports' better than a hardcoded string like '/home/aryan/.ram_manager/reports'? What happens when someone else runs your code?

Learning objectives

Create directories safely with Path.mkdir(parents=True, exist_ok=True)
Implement atomic file writes using a .tmp file and Path.replace()
Use sorted() on glob results before slicing for reliable ordering
Convert dataclass instances to dicts with dataclasses.asdict()
Load and slice file-based history by sorting filenames

Key concepts

Path.mkdir(parents=True, exist_ok=True) — safe directory creation
Path.replace() — atomic rename on POSIX
sorted(path.glob(…)) — deterministic file ordering
dataclasses.asdict() — dataclass to JSON-serializable dict
with_suffix(‘.tmp’) — temporary file path

Try it

import json
from pathlib import Path
from dataclasses import dataclass, asdict
from datetime import datetime
REPORT_DIR = Path.home() / ‘.ram_manager’ / ‘reports’
@dataclass
class SessionReport:
timestamp: str
ram_percent: float
top_processes: list
llm_advice: str
def save_report(report: SessionReport, report_dir: Path = None) -> Path:
"""Save a session report as JSON, appending to today’s file."""
dir_to_use = report_dir if report_dir is not None else REPORT_DIR
# 💡 : hint : -010 xp Call dir_to_use.mkdir(parents=True, exist_ok=True) here — crashes with FileNotFoundError if directory doesn’t exist
filename = dir_to_use / f’session_{report.timestamp[:10]}.json’
<pre class="giallo" style="color-scheme: light dark; color: light-dark(#24292E, #E1E4E8); background-color: light-dark(#FFFFFF, #24292E);"><code data-lang="plain">existing = []
if filename.exists():
 existing = json.loads(filename.read_text())

existing.append(asdict(report))

# 💡 : hint : -010 xp Write to a .tmp file first, then call tmp.replace(filename) — direct write corrupts on crash
filename.write_text(json.dumps(existing, indent=2))
return filename</code></pre>
def load_history(days: int = 7, report_dir: Path = None) -> list:
"""Load the last N days of session reports."""
dir_to_use = report_dir if report_dir is not None else REPORT_DIR
all_sessions = []
# 💡 : hint : -010 xp Use sorted(dir_to_use.glob(…))[-days:] — glob returns unsorted, so slice the sorted file list not the flat session list
for path in dir_to_use.glob(‘session_*.json’):
all_sessions.extend(json.loads(path.read_text()))
return all_sessions[-days:]
def report_summary(reports: list) -> dict:
"""Summarise a list of report dicts."""
if not reports:
return {‘count’: 0, ‘avg_ram’: 0.0, ‘max_ram’: 0.0}
percents = [r[‘ram_percent’] for r in reports]
return {
‘count’: len(percents),
# BUG 5: Not rounding — produces very long floats
‘avg_ram’: sum(percents) / len(percents),
‘max_ram’: max(percents),
}

Concept detail

File I/O and JSON Reports

Reading and writing files with pathlib

from pathlib import Path

path = Path.home() / '.myapp' / 'data.json'

# Create directory (safe even if it already exists)
path.parent.mkdir(parents=True, exist_ok=True)

# Write
path.write_text(json.dumps(data, indent=2))

# Read
data = json.loads(path.read_text())

# Append pattern (read → modify → write)
existing = json.loads(path.read_text()) if path.exists() else []
existing.append(new_item)
path.write_text(json.dumps(existing, indent=2))

Atomic writes — prevent corruption on crash

# WRONG: direct write — reader may see partial content if process dies
path.write_text(json.dumps(data))

# RIGHT: write to .tmp, then atomically rename
tmp = path.with_suffix('.tmp')
tmp.write_text(json.dumps(data, indent=2))
tmp.replace(path)   # atomic on POSIX (Linux, macOS)

Sorting glob results

# Arbitrary order — don't rely on it
for f in path.glob('*.json'):
    ...

# Sorted (alphabetical = chronological for ISO date filenames)
for f in sorted(path.glob('*.json')):
    ...

# Last N files
recent = sorted(path.glob('session_*.json'))[-7:]

dataclasses.asdict()

from dataclasses import dataclass, asdict

@dataclass
class Report:
    timestamp: str
    value: float

r = Report('2024-01-15', 72.3)
asdict(r)  # → {'timestamp': '2024-01-15', 'value': 72.3}
json.dumps(asdict(r))  # fully JSON-serializable

Solution

import json
from pathlib import Path
from dataclasses import dataclass, asdict
from datetime import datetime

REPORT_DIR = Path.home() / '.ram_manager' / 'reports'

@dataclass
class SessionReport:
    timestamp: str
    ram_percent: float
    top_processes: list
    llm_advice: str

def save_report(report: SessionReport, report_dir: Path = None) -> Path:
    """Save a session report as JSON, appending to today's file."""
    dir_to_use = report_dir if report_dir is not None else REPORT_DIR
    dir_to_use.mkdir(parents=True, exist_ok=True)  # FIX 1: create if missing
    filename = dir_to_use / f'session_{report.timestamp[:10]}.json'

    existing = []
    if filename.exists():
        existing = json.loads(filename.read_text())

    existing.append(asdict(report))

    # FIX 2: Atomic write — write to .tmp then rename
    # rename() is atomic on POSIX — reader never sees partial content
    tmp = filename.with_suffix('.tmp')
    tmp.write_text(json.dumps(existing, indent=2))
    tmp.replace(filename)
    return filename

def load_history(days: int = 7, report_dir: Path = None) -> list:
    """Load the last N days of session reports."""
    dir_to_use = report_dir if report_dir is not None else REPORT_DIR
    all_sessions = []
    # FIX 3+4: Sort by filename (ISO date prefix), slice to last N files
    for path in sorted(dir_to_use.glob('session_*.json'))[-days:]:
        all_sessions.extend(json.loads(path.read_text()))
    return all_sessions

def report_summary(reports: list) -> dict:
    """Summarise a list of report dicts."""
    if not reports:
        return {'count': 0, 'avg_ram': 0.0, 'max_ram': 0.0}
    percents = [r['ram_percent'] for r in reports]
    return {
        'count': len(percents),
        'avg_ram': round(sum(percents) / len(percents), 1),  # FIX 5: round result
        'max_ram': max(percents),
    }

Tests

import json
import tempfile
from pathlib import Path
from datetime import date

def _make_report(percent=60.0, date_str=None):
    return SessionReport(
        timestamp=f'{date_str or date.today().isoformat()}T10:00:00',
        ram_percent=percent,
        top_processes=[{'name': 'chrome', 'rss_mb': 500}],
        llm_advice='Kill Chrome to free RAM.',
    )

def test_save_report_creates_directory():
    with tempfile.TemporaryDirectory() as tmp:
        missing_dir = Path(tmp) / 'deep' / 'nested' / 'reports'
        report = _make_report()
        save_report(report, report_dir=missing_dir)
        assert missing_dir.exists(), 'save_report must create the directory'

def test_save_report_appends_to_existing():
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        r1 = _make_report(percent=50.0)
        r2 = _make_report(percent=75.0)
        save_report(r1, report_dir=tmp_path)
        save_report(r2, report_dir=tmp_path)
        today = date.today().isoformat()
        file = tmp_path / f'session_{today}.json'
        data = json.loads(file.read_text())
        assert len(data) == 2
        assert data[0]['ram_percent'] == 50.0
        assert data[1]['ram_percent'] == 75.0

def test_save_report_atomic_no_tmp_on_success():
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        save_report(_make_report(), report_dir=tmp_path)
        tmp_files = list(tmp_path.glob('*.tmp'))
        assert tmp_files == [], '.tmp file should be deleted after successful save'

def test_load_history_returns_recent_days():
    with tempfile.TemporaryDirectory() as tmp:
        tmp_path = Path(tmp)
        # Create 10 days of reports
        for i in range(10):
            d = f'2024-01-{i+1:02d}'
            f = tmp_path / f'session_{d}.json'
            f.write_text(json.dumps([{
                'timestamp': f'{d}T00:00:00',
                'ram_percent': float(i * 10),
                'top_processes': [],
                'llm_advice': '',
            }]))
        history = load_history(days=3, report_dir=tmp_path)
        # Should include only the 3 most recent files' sessions (1 per file)
        assert len(history) == 3

def test_report_summary_empty():
    s = report_summary([])
    assert s == {'count': 0, 'avg_ram': 0.0, 'max_ram': 0.0}

def test_report_summary_calculates_correctly():
    reports = [
        {'ram_percent': 50.0},
        {'ram_percent': 70.0},
        {'ram_percent': 90.0},
    ]
    s = report_summary(reports)
    assert s['count'] == 3
    assert s['avg_ram'] == 70.0  # (50+70+90)/3
    assert s['max_ram'] == 90.0

def test_report_summary_rounding():
    reports = [{'ram_percent': 33.3}, {'ram_percent': 33.3}, {'ram_percent': 33.4}]
    s = report_summary(reports)
    # Without round(), result is 33.333333... — must be rounded to 1 decimal
    assert isinstance(s['avg_ram'], float)
    assert s['avg_ram'] == round(sum([33.3, 33.3, 33.4]) / 3, 1)