Atomic File Writes

Rewrite save_expenses to use atomic file operations that prevent data corruption

💻

Writing code and entering commands is only available on desktop. Open this page on a larger screen to complete this chapter.

The corruption problem

Your save_expenses() function writes data directly to expenses.json. If the program crashes or loses power while writing, the file can end up partially written — valid JSON at the top, truncated garbage at the bottom. When the API restarts, load_expenses() fails because it cannot parse the corrupted file. All your data is lost.

This is not a theoretical risk. Production servers crash, disks fill up, and processes get killed. Any file write that is not atomic is a data loss risk.

The atomic write pattern

The fix is a two-step process:

Write the data to a temporary file in the same directory
Use os.replace() to swap the temp file into the real file path

os.replace() is atomic on all major operating systems. It either completes fully or does not happen at all. If the program crashes during step 1, the original file is untouched. If it crashes during step 2, the operating system guarantees the swap either completes or rolls back.

Why the same directory matters

The temporary file must be in the same directory as the target file. os.replace() is only guaranteed to be atomic when both files are on the same filesystem. Writing to a different directory might cross filesystem boundaries.

Instructions

Rewrite save_expenses() to use atomic file writes.

Add import os to the imports at the top of the file.
Replace the body of save_expenses() with three parts:
- Set temp_path = str(DATA_FILE) + ".tmp" — this is the temporary file you will write to first.
- Open it with with open(temp_path, "w") as f: and write the data with json.dump({"counter": counter, "expenses": expenses}, f, indent=2).
- After the with block, call os.replace(temp_path, str(DATA_FILE)) — this atomically swaps the temp file into place.

← Previous Chapter Next Chapter →

import json
import time
from datetime import datetime
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional

# Step 1: Add import os

app = FastAPI()
DATA_FILE = Path("expenses.json")

class Expense(BaseModel):
    description: str = Field(min_length=1)
    amount: float = Field(gt=0)
    category: Literal["food", "transport", "entertainment", "utilities", "other"]
    date: Optional[str] = None

    @field_validator("date")
    @classmethod
    def validate_date_format(cls, v):
        if v is not None:
            try:
                datetime.strptime(v, "%Y-%m-%d")
            except ValueError:
                raise ValueError("Date must be in YYYY-MM-DD format")
        return v

class ExpenseUpdate(BaseModel):
    description: Optional[str] = Field(default=None, min_length=1)
    amount: Optional[float] = Field(default=None, gt=0)
    category: Optional[Literal["food", "transport", "entertainment", "utilities", "other"]] = None

class ExpenseOut(BaseModel):
    id: int
    description: str
    amount: float
    category: str
    date: Optional[str] = None

class ErrorResponse(BaseModel):
    detail: str

def save_expenses():
    # Step 2: Replace this function body with atomic write
    #   Set temp_path = str(DATA_FILE) + ".tmp"
    #   Open temp_path and json.dump the data
    #   Call os.replace(temp_path, str(DATA_FILE))
    data = {"counter": counter, "expenses": expenses}
    DATA_FILE.write_text(json.dumps(data, indent=2))

def load_expenses():
    global expenses, counter
    if DATA_FILE.exists():
        data = json.loads(DATA_FILE.read_text())
        expenses = {int(k): v for k, v in data["expenses"].items()}
        counter = data["counter"]

expenses = {}
counter = 0
load_expenses()

@app.middleware("http")
async def log_requests(request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = (time.time() - start) * 1000
    print(f"{request.method} {request.url.path} {response.status_code} {duration:.1f}ms")
    return response

@app.post("/expenses", status_code=201, response_model=ExpenseOut)
def create_expense(expense: Expense):
    global counter
    counter += 1
    expenses[counter] = {"id": counter, **expense.model_dump()}
    save_expenses()
    return expenses[counter]

@app.get("/expenses", response_model=list[ExpenseOut])
def list_expenses(category: str = None):
    if category:
        return [e for e in expenses.values() if e["category"] == category]
    return list(expenses.values())

@app.get("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def get_expense(expense_id: int):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    return expenses[expense_id]

@app.delete("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def delete_expense(expense_id: int):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    save_expenses()
    return expenses.pop(expense_id)

@app.patch("/expenses/{expense_id}", response_model=ExpenseOut, responses={404: {"model": ErrorResponse}})
def update_expense(expense_id: int, updates: ExpenseUpdate):
    if expense_id not in expenses:
        raise HTTPException(status_code=404, detail="Expense not found")
    expenses[expense_id].update(updates.model_dump(exclude_unset=True))
    save_expenses()
    return expenses[expense_id]

@app.get("/summary")
def spending_summary():
    summary = {}
    for expense in expenses.values():
        cat = expense["category"]
        summary[cat] = summary.get(cat, 0) + expense["amount"]
    return summary

Atomic File Writes

The corruption problem

The atomic write pattern

Why the same directory matters

Instructions

Interactive Code Editor