Working with CSV, JSON and .env Files

Web applications deal with structured text data constantly โ€” CSV files uploaded by users, JSON configuration files, API payloads, and environment variable files. Python’s standard library provides robust tools for each format: the csv module for tabular data, the json module for serialisation and deserialisation, and python-dotenv for loading .env files. In FastAPI, these appear together in upload endpoints that accept CSV files, configuration systems that load settings from .env, and response serialisation that produces JSON from Python objects.

Working with CSV

import csv
from pathlib import Path
from io import StringIO

# โ”€โ”€ Read CSV from file โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
with open("users.csv", "r", encoding="utf-8", newline="") as f:
    reader = csv.DictReader(f)   # reads into dicts using first row as header
    for row in reader:
        print(row)
        # {"name": "Alice", "email": "alice@example.com", "role": "admin"}

# Access headers
with open("users.csv", encoding="utf-8", newline="") as f:
    reader = csv.DictReader(f)
    headers = reader.fieldnames   # ["name", "email", "role"]
    rows    = list(reader)        # list of dicts

# โ”€โ”€ Write CSV to file โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
users = [
    {"name": "Alice", "email": "alice@example.com", "role": "admin"},
    {"name": "Bob",   "email": "bob@example.com",   "role": "user"},
]

with open("output.csv", "w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "email", "role"])
    writer.writeheader()
    writer.writerows(users)

# โ”€โ”€ FastAPI file upload โ€” process CSV bytes from UploadFile โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
from fastapi import UploadFile

async def process_csv_upload(file: UploadFile) -> list[dict]:
    content = await file.read()
    text    = content.decode("utf-8")
    reader  = csv.DictReader(StringIO(text))   # in-memory โ€” no disk write
    return [
        {"name": row["name"].strip(), "email": row["email"].strip().lower()}
        for row in reader
        if row.get("email")   # skip rows with empty email
    ]
Note: Always pass newline="" when opening CSV files โ€” the csv module handles line endings itself, and without this argument, universal newline mode in Python may cause empty rows or double-newlines on Windows. Also always specify encoding="utf-8" explicitly. These two parameters together prevent 90% of CSV parsing bugs encountered in production when files come from different operating systems.
Tip: For FastAPI file upload endpoints, use StringIO (from the io module) to wrap the decoded CSV bytes in a file-like object that the csv module can read. This avoids writing the uploaded file to disk just to read it back โ€” the whole processing happens in memory. For very large CSV uploads (hundreds of MB), stream the file instead using await file.read(chunk_size) in a loop.
Warning: Validate CSV data thoroughly before inserting into the database. Users will upload malformed CSVs, files with the wrong column names, extra whitespace in fields, missing required values, and wrong data types. Always validate each row and collect errors rather than failing on the first bad row โ€” returning a list of validation errors per row (with row numbers) is much more useful to the user than a single 400 error for the first problem found.

Working with JSON

import json
from pathlib import Path
from datetime import datetime

# โ”€โ”€ Parse JSON string โ†’ Python โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
json_str = '{"name": "Alice", "age": 30, "active": true}'
data     = json.loads(json_str)   # loads = load string
print(data)         # {"name": "Alice", "age": 30, "active": True}
print(type(data))   # dict

# โ”€โ”€ Convert Python โ†’ JSON string โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
user = {"name": "Bob", "age": 25, "tags": ["python", "fastapi"]}
json_str = json.dumps(user)                      # compact
json_pretty = json.dumps(user, indent=2)         # formatted
json_sorted = json.dumps(user, sort_keys=True)   # sorted keys

print(json_pretty)
# {
#   "name": "Bob",
#   "age": 25,
#   "tags": ["python", "fastapi"]
# }

# โ”€โ”€ Read/write JSON files โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Write
with open("config.json", "w", encoding="utf-8") as f:
    json.dump(user, f, indent=2, ensure_ascii=False)

# Read
with open("config.json", "r", encoding="utf-8") as f:
    config = json.load(f)   # load = load file

# pathlib shortcut
Path("config.json").write_text(json.dumps(user, indent=2), encoding="utf-8")
config = json.loads(Path("config.json").read_text(encoding="utf-8"))

# โ”€โ”€ Handling non-serialisable types โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def json_serialiser(obj):
    """Custom serialiser for types that json module cannot handle."""
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Type {type(obj)} not serialisable")

data = {"name": "Event", "timestamp": datetime.now()}
json_str = json.dumps(data, default=json_serialiser)
# {"name": "Event", "timestamp": "2025-08-06T14:30:00.000000"}

Working with .env Files

# .env file format:
# DATABASE_URL=postgresql://localhost/blogdb
# SECRET_KEY=your-secret-key-here
# DEBUG=true
# ALLOWED_ORIGINS=http://localhost:5173,http://localhost:3000

# โ”€โ”€ python-dotenv (install: pip install python-dotenv) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
from dotenv import load_dotenv
import os

# Load .env into os.environ โ€” call early, before other imports
load_dotenv()   # looks for .env in current directory
load_dotenv(".env.local")          # load from specific path
load_dotenv(override=True)         # override existing env vars

db_url = os.getenv("DATABASE_URL", "postgresql://localhost/dev")
debug  = os.getenv("DEBUG", "false").lower() == "true"
port   = int(os.getenv("PORT", "8000"))

# โ”€โ”€ pydantic-settings (recommended for FastAPI) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url:  str  = "postgresql://localhost/blogdb"
    secret_key:    str  = "changeme"
    debug:         bool = False
    port:          int  = 8000
    allowed_origins: list[str] = ["http://localhost:5173"]

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}

settings = Settings()   # reads from .env AND environment variables automatically
print(settings.database_url)

# โ”€โ”€ .env file for different environments โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# .env                  โ† development defaults (NOT committed to git)
# .env.example          โ† template with dummy values (committed to git)
# .env.test             โ† test database settings
# .env.production       โ† NEVER committed โ€” only on the server

Complete FastAPI CSV Upload Endpoint

from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel, EmailStr
from typing import List
from io import StringIO
import csv

app = FastAPI()

class UserImportRow(BaseModel):
    name:  str
    email: str
    role:  str = "user"

class ImportResult(BaseModel):
    imported: int
    errors:   List[dict]

@app.post("/users/import", response_model=ImportResult)
async def import_users(file: UploadFile = File(...)):
    # Validate file type
    if not file.filename.endswith(".csv"):
        raise HTTPException(400, "Only CSV files are accepted")

    # Read and decode
    content = await file.read()
    try:
        text = content.decode("utf-8")
    except UnicodeDecodeError:
        raise HTTPException(400, "File must be UTF-8 encoded")

    # Parse CSV
    reader    = csv.DictReader(StringIO(text))
    imported  = 0
    row_errors = []

    for row_num, row in enumerate(reader, start=2):   # start=2 (row 1 is header)
        try:
            # Validate using Pydantic
            user_data = UserImportRow(
                name  = row.get("name", "").strip(),
                email = row.get("email", "").strip().lower(),
                role  = row.get("role", "user").strip(),
            )
            # TODO: save to database
            imported += 1
        except Exception as e:
            row_errors.append({"row": row_num, "error": str(e), "data": dict(row)})

    return ImportResult(imported=imported, errors=row_errors)

Common Mistakes

Mistake 1 โ€” Missing newline=”” when opening CSV files

โŒ Wrong โ€” universal newline mode interferes with csv module:

with open("data.csv", "r") as f:   # missing newline="" and encoding!
    reader = csv.reader(f)
    # May produce empty rows on Windows files

โœ… Correct:

with open("data.csv", "r", encoding="utf-8", newline="") as f:   # โœ“
    reader = csv.DictReader(f)

Mistake 2 โ€” Serialising datetime with json.dumps without a custom serialiser

โŒ Wrong โ€” TypeError on datetime objects:

data = {"created": datetime.now()}
json.dumps(data)   # TypeError: Object of type datetime is not JSON serializable

โœ… Correct โ€” use the default parameter:

json.dumps(data, default=lambda o: o.isoformat() if isinstance(o, datetime) else str(o))

Mistake 3 โ€” Committing .env with real secrets to git

โŒ Wrong โ€” .env with production credentials committed:

git add .env   # exposes SECRET_KEY, DATABASE_URL to anyone with repo access

โœ… Correct โ€” commit only .env.example with dummy values, add .env to .gitignore.

Quick Reference

Task Code
Read CSV to dicts csv.DictReader(f)
Write CSV from dicts csv.DictWriter(f, fieldnames=[...])
Parse CSV from string csv.DictReader(StringIO(text))
Parse JSON string json.loads(text)
Serialise to JSON string json.dumps(obj, indent=2)
Read JSON file json.load(f)
Write JSON file json.dump(obj, f, indent=2)
Load .env file load_dotenv() or BaseSettings()
Read env var os.getenv("KEY", "default")

🧠 Test Yourself

A FastAPI upload endpoint receives a CSV file as an UploadFile. You call await file.read() and get bytes. What is the correct sequence to parse it as CSV without writing to disk?