Python File Handling: Reading, Writing, CSV, JSON, Context Managers, pathlib, Encoding, Large Files, and Every File Pattern Data Engineers Use

Every data engineering pipeline starts and ends with files. Read a CSV from a data lake. Write a JSON config. Parse a log file. Export a report. Move files between directories. Check if a file exists before processing. Append rows to an output file. Every single one of these operations is file handling.

Yet most tutorials stop at open("file.txt", "r"). They skip context managers (the with statement that prevents resource leaks), skip CSV and JSON (the two most common data formats), skip pathlib (the modern way to handle file paths), and skip encoding issues (the silent data corruption that breaks production pipelines).

Think of file handling like a library visit. You check out a book (open a file), read it (read data), take notes (write data), and return the book (close the file). If you forget to return the book (forget to close the file), you are hogging a resource that others cannot use — and eventually the library runs out of available books (the OS runs out of file handles). The with statement is like an auto-return policy — the book is automatically returned when you leave the library, even if you trip on the way out (an exception occurs).

Table of Contents

• Opening and Closing Files
• The Basic Way (and Why It Is Dangerous)
• Context Managers — the with Statement
• Why You Should Always Use with
• File Modes
• All File Modes Explained
• Choosing the Right Mode
• Reading Files
• read() — Entire File as One String
• readline() — One Line at a Time
• readlines() — All Lines as a List
• Iterating Line by Line (Best for Large Files)
• Writing Files
• write() — Write a String
• writelines() — Write a List of Strings
• Appending to Files
• Print to a File
• Working with File Paths — pathlib
• Why pathlib Over os.path
• Common pathlib Operations
• Building Paths Safely
• Listing Files in a Directory
• Checking if Files Exist
• CSV Files
• Reading CSV with csv.reader
• Reading CSV with csv.DictReader
• Writing CSV with csv.writer
• Writing CSV with csv.DictWriter
• Handling Edge Cases (Commas, Quotes, Encoding)
• JSON Files
• Reading JSON from a File
• Writing JSON to a File
• JSON Strings (loads and dumps)
• Pretty Printing JSON
• Handling Nested JSON
• JSON with Non-Serializable Types
• Working with Large Files
• Line-by-Line Processing
• Chunked Reading
• Counting Lines Without Loading
• Encoding and Unicode
• What Is Encoding
• Common Encoding Problems
• Detecting and Fixing Encoding Issues
• File Operations — Copy, Move, Delete
• Data Engineering Patterns
• Config File Reader
• Processing All Files in a Directory
• Safe File Writer (Temp File Pattern)
• File-Based Logging
• Common Mistakes
• Interview Questions
• Wrapping Up

Opening and Closing Files

The Basic Way (and Why It Is Dangerous)

# The dangerous way — manually opening and closing
f = open("data.txt", "r")
content = f.read()
print(content)
f.close()         # Easy to forget!

# What if an error occurs BEFORE close()?
f = open("data.txt", "r")
content = f.read()
result = 10 / 0   # ZeroDivisionError — crashes here
f.close()         # NEVER REACHED — file stays open!

# The file handle leaks. Do this enough times and your process
# runs out of file handles (OS limit, usually 1024).
# The program slows down, then crashes with "Too many open files."

Context Managers — the with Statement

# The safe way — with statement (context manager)
with open("data.txt", "r") as f:
    content = f.read()
    print(content)
# File is AUTOMATICALLY closed here — even if an error occurs inside the block

# Even with an error, the file is closed:
with open("data.txt", "r") as f:
    content = f.read()
    result = 10 / 0   # ZeroDivisionError — but the file is STILL closed!
# f.close() is called automatically by the with statement before the error propagates

# Multiple files at once:
with open("input.txt", "r") as src, open("output.txt", "w") as dst:
    dst.write(src.read())

Real-life analogy: The with statement is like a hotel check-in/check-out system. You check in (open the file), use the room (read/write data), and when you leave the building — whether you walk out normally or the fire alarm goes off — the room is automatically cleaned and made available (file closed). Without with, you have to remember to check out manually, and if you forget, the room stays occupied forever.

Why You Should Always Use with

1. Automatic cleanup — file is closed even if an exception occurs.
2. No resource leaks — prevents “Too many open files” errors in long-running pipelines.
3. Cleaner code — no try/finally block needed for cleanup.
4. PEP 8 standard — the with statement is the expected way to handle files in Python.
5. Works with more than files — database connections, network sockets, locks all use the same pattern.

File Modes

All File Modes Explained

Mode	Name	Description	File Exists?	File Missing?
"r"	Read	Read only	Opens file	FileNotFoundError
"w"	Write	Write only (overwrites!)	Erases content	Creates new file
"a"	Append	Adds to end	Keeps content, adds to end	Creates new file
"x"	Exclusive create	Write only, fails if exists	FileExistsError	Creates new file
"r+"	Read + Write	Both read and write	Opens file	FileNotFoundError
"w+"	Write + Read	Write and read (overwrites!)	Erases content	Creates new file
"a+"	Append + Read	Append and read	Keeps content	Creates new file
"rb"	Read binary	Read bytes (images, PDFs)	Opens file	FileNotFoundError
"wb"	Write binary	Write bytes	Erases content	Creates new file

Choosing the Right Mode

# Reading data → "r" (default)
with open("data.csv", "r") as f:
    content = f.read()

# Writing a fresh output file → "w" (WARNING: erases existing content!)
with open("output.csv", "w") as f:
    f.write("id,name
")

# Adding to a log file → "a" (keeps existing content)
with open("pipeline.log", "a") as f:
    f.write("2026-07-07 14:30:00 | Pipeline completed
")

# Creating a file only if it does not exist → "x" (safe creation)
try:
    with open("report.csv", "x") as f:
        f.write("id,name
")
except FileExistsError:
    print("File already exists — skipping creation")

# Reading binary (images, PDFs, Parquet) → "rb"
with open("image.png", "rb") as f:
    binary_data = f.read()

Reading Files

read() — Entire File as One String

# Reads the ENTIRE file into memory as a single string
with open("data.txt", "r") as f:
    content = f.read()
print(content)       # "Hello
World
Python
"
print(len(content))  # Total characters including newlines

# Read only the first N characters
with open("data.txt", "r") as f:
    first_100 = f.read(100)   # First 100 characters only

# WARNING: Do NOT use read() on large files (GBs)
# It loads the ENTIRE file into memory — can crash your process

readline() — One Line at a Time

# Reads one line at a time (includes the 
 newline character)
with open("data.txt", "r") as f:
    line1 = f.readline()    # "Hello
"
    line2 = f.readline()    # "World
"
    line3 = f.readline()    # "Python
"
    line4 = f.readline()    # "" — empty string means end of file

# Strip the newline
line = f.readline().strip()  # "Hello" (no 
)

readlines() — All Lines as a List

# Reads ALL lines into a list (each line is a string with 
)
with open("data.txt", "r") as f:
    lines = f.readlines()
print(lines)  # ["Hello
", "World
", "Python
"]

# Strip newlines from each line
clean_lines = [line.strip() for line in lines]
print(clean_lines)  # ["Hello", "World", "Python"]

# WARNING: Like read(), this loads the entire file into memory

Iterating Line by Line (Best for Large Files)

# The BEST way to read files — one line at a time, constant memory
with open("data.txt", "r") as f:
    for line in f:                # File object is an ITERATOR
        print(line.strip())       # Process one line, then discard it

# This works on files of ANY size — 1 KB or 100 GB
# Only ONE line is in memory at a time
# The file object is lazy — it reads the next line only when asked

# Practical: count lines matching a pattern
error_count = 0
with open("pipeline.log", "r") as f:
    for line in f:
        if "ERROR" in line:
            error_count += 1
print(f"Errors found: {error_count}")

# Practical: find the first matching line
with open("config.txt", "r") as f:
    for line in f:
        if line.startswith("db_host="):
            db_host = line.strip().split("=")[1]
            break

Real-life analogy: read() is like photocopying an entire 500-page book and carrying all the pages home. readline() is like reading one page at the library and taking notes. Iterating line by line is like scanning one page, writing down what you need, then flipping to the next page — you never carry the whole book.

Writing Files

write() — Write a String

# write() does NOT add newlines — you must add 
 yourself
with open("output.txt", "w") as f:
    f.write("Hello, World!
")
    f.write("Python is great.
")
    f.write(f"Today's date: 2026-07-07
")

# Write formatted data
with open("report.txt", "w") as f:
    f.write("=" * 40 + "
")
    f.write("PIPELINE REPORT
")
    f.write("=" * 40 + "
")
    f.write(f"Tables processed: 15
")
    f.write(f"Rows loaded: 1,250,000
")
    f.write(f"Duration: 45 seconds
")

writelines() — Write a List of Strings

# writelines() does NOT add newlines between items — you must include them
lines = ["Hello
", "World
", "Python
"]
with open("output.txt", "w") as f:
    f.writelines(lines)

# Common pattern: add newlines during write
data = ["Alice", "Bob", "Charlie"]
with open("names.txt", "w") as f:
    f.writelines(name + "
" for name in data)

Appending to Files

# "a" mode adds to the END of the file without erasing existing content
with open("pipeline.log", "a") as f:
    f.write("2026-07-07 14:30:00 | Pipeline started
")
    f.write("2026-07-07 14:31:30 | 15 tables loaded
")
    f.write("2026-07-07 14:31:35 | Pipeline completed
")

# Next run — appends after the previous entries
with open("pipeline.log", "a") as f:
    f.write("2026-07-08 14:30:00 | Pipeline started
")

Print to a File

# print() has a file parameter — writes to a file instead of the screen
with open("output.txt", "w") as f:
    print("Hello, World!", file=f)       # Adds 
 automatically
    print("Python", "is", "great", file=f)  # Adds spaces between args
    print(f"Count: {42}", file=f)

# Useful for quick debugging output to a file
with open("debug.log", "w") as f:
    for i, item in enumerate(data):
        print(f"[{i}] {item}", file=f)

Working with File Paths — pathlib

Why pathlib Over os.path

# os.path (old way) — string manipulation, platform-dependent
import os
path = os.path.join("data", "bronze", "customers.csv")
exists = os.path.exists(path)
name = os.path.basename(path)
ext = os.path.splitext(path)[1]

# pathlib (modern way) — object-oriented, cross-platform, readable
from pathlib import Path
path = Path("data") / "bronze" / "customers.csv"
exists = path.exists()
name = path.name          # "customers.csv"
ext = path.suffix         # ".csv"
stem = path.stem          # "customers"

# pathlib uses / operator for joining — no os.path.join() needed
# Works on Windows, Mac, and Linux automatically

Real-life analogy: os.path is like giving directions with raw street names — “turn left on Main, then right on Oak, then left on 3rd.” pathlib is like using a GPS — you say “take me to 123 Main Street” and it figures out the platform-specific details (forward slash on Mac/Linux, backslash on Windows).

Common pathlib Operations

from pathlib import Path

p = Path("/data/bronze/customers/2026/07/customers_20260707.csv")

p.name         # "customers_20260707.csv"
p.stem         # "customers_20260707"
p.suffix       # ".csv"
p.parent       # Path("/data/bronze/customers/2026/07")
p.parents[0]   # Path("/data/bronze/customers/2026/07")
p.parents[1]   # Path("/data/bronze/customers/2026")
p.parts        # ('/', 'data', 'bronze', 'customers', '2026', '07', 'customers_20260707.csv')
p.is_file()    # True if it is a file
p.is_dir()     # True if it is a directory
p.exists()     # True if it exists on disk

# Change extension
new_path = p.with_suffix(".parquet")   # .../customers_20260707.parquet

# Change filename
new_path = p.with_name("orders_20260707.csv")   # .../orders_20260707.csv

Building Paths Safely

# Build paths with / operator — cross-platform safe
data_dir = Path("/data/bronze")
file_path = data_dir / "customers" / "2026" / "07" / "customers.csv"
print(file_path)  # /data/bronze/customers/2026/07/customers.csv

# Dynamic paths based on date
from datetime import date
today = date.today()
output_path = Path("/data/silver") / f"year={today.year}" / f"month={today.month:02d}"
output_path.mkdir(parents=True, exist_ok=True)   # Create directories if missing

# Home directory
home = Path.home()                   # /Users/naveen or /home/naveen
config = home / ".config" / "pipeline" / "settings.json"

# Current working directory
cwd = Path.cwd()                     # Where the script is running

Listing Files in a Directory

# List all files in a directory
data_dir = Path("/data/bronze/customers")

# All items (files + directories)
for item in data_dir.iterdir():
    print(item.name)

# Only CSV files
csv_files = list(data_dir.glob("*.csv"))
print(f"Found {len(csv_files)} CSV files")

# Recursive search (all subdirectories)
all_csvs = list(data_dir.rglob("*.csv"))
print(f"Found {len(all_csvs)} CSV files in all subdirectories")

# Filter by pattern
parquet_files = sorted(data_dir.glob("*.parquet"))
log_files = list(Path("/var/log").glob("pipeline_*.log"))

# Practical: process all CSVs in a directory
for csv_file in sorted(data_dir.glob("*.csv")):
    print(f"Processing: {csv_file.name} ({csv_file.stat().st_size:,} bytes)")
    # process(csv_file)

Checking if Files Exist

# Always check before reading
path = Path("config.json")
if path.exists():
    with open(path, "r") as f:
        config = json.load(f)
else:
    print("Config file not found — using defaults")
    config = {"host": "localhost", "port": 5432}

# Check type
path = Path("/data/bronze")
print(path.is_file())     # False — it is a directory
print(path.is_dir())      # True

# Create directory if missing
output_dir = Path("/data/silver/customers")
output_dir.mkdir(parents=True, exist_ok=True)
# parents=True: creates all intermediate directories
# exist_ok=True: no error if directory already exists

CSV Files

Reading CSV with csv.reader

import csv

# Basic CSV reading — returns each row as a list of strings
with open("employees.csv", "r") as f:
    reader = csv.reader(f)
    header = next(reader)        # First row = column names
    print(f"Columns: {header}")  # ['id', 'name', 'department', 'salary']

    for row in reader:
        print(f"  {row[1]} in {row[2]} earns ${row[3]}")
        # row[0] = "1001", row[1] = "Naveen", row[2] = "Engineering", row[3] = "95000"
        # Everything is a STRING — cast to int/float as needed
        salary = int(row[3])

# With different delimiter (TSV = tab-separated)
with open("data.tsv", "r") as f:
    reader = csv.reader(f, delimiter="  ")
    for row in reader:
        print(row)

Reading CSV with csv.DictReader

# DictReader returns each row as a DICTIONARY — access by column name
with open("employees.csv", "r") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"  {row['name']} in {row['department']} earns ${row['salary']}")
        # row = {'id': '1001', 'name': 'Naveen', 'department': 'Engineering', 'salary': '95000'}

# DictReader is preferred over reader because:
# - Access by name: row['salary'] instead of row[3]
# - Column order does not matter — works even if columns are reordered
# - Self-documenting — you see what each field means

# Filter while reading
engineers = []
with open("employees.csv", "r") as f:
    for row in csv.DictReader(f):
        if row["department"] == "Engineering":
            engineers.append(row)
print(f"Engineers: {len(engineers)}")

Real-life analogy: csv.reader is like a spreadsheet with hidden column headers — you access data by column number (A, B, C). csv.DictReader is like a spreadsheet with visible headers — you access data by column name (Name, Department, Salary). When someone reorders the columns, DictReader still works; reader breaks.

Writing CSV with csv.writer

import csv

# Write rows as lists
with open("output.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["id", "name", "city"])            # Header
    writer.writerow([1001, "Naveen", "Toronto"])        # Data row
    writer.writerow([1002, "Alice", "London"])

    # Write multiple rows at once
    rows = [[1003, "Bob", "Delhi"], [1004, "Charlie", "Vancouver"]]
    writer.writerows(rows)

# newline="" is REQUIRED on Windows to prevent extra blank lines

Writing CSV with csv.DictWriter

# DictWriter writes from dictionaries — cleaner for complex data
records = [
    {"id": 1001, "name": "Naveen", "city": "Toronto", "salary": 95000},
    {"id": 1002, "name": "Alice", "city": "London", "salary": 88000},
    {"id": 1003, "name": "Bob", "city": "Delhi", "salary": 72000},
]

with open("output.csv", "w", newline="") as f:
    fieldnames = ["id", "name", "city", "salary"]
    writer = csv.DictWriter(f, fieldnames=fieldnames)

    writer.writeheader()     # Writes the header row
    writer.writerows(records)  # Writes all data rows

# DictWriter ensures columns are always in the correct order
# Even if your dicts have keys in a different order

Handling Edge Cases (Commas, Quotes, Encoding)

# Values with commas — csv module handles quoting automatically
with open("addresses.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["name", "address"])
    writer.writerow(["Naveen", "123 Main St, Suite 456, Toronto"])
    # Output: Naveen,"123 Main St, Suite 456, Toronto"
    # The csv module automatically quotes fields containing commas

# Values with quotes
writer.writerow(["She said", 'He said "hello"'])
# Output: She said,"He said ""hello"""
# Double quotes are escaped as ""

# Specifying encoding (important for international characters)
with open("data.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row)

JSON Files

Reading JSON from a File

import json

# json.load() reads from a FILE object
with open("config.json", "r") as f:
    config = json.load(f)

print(type(config))         # <class 'dict'>
print(config["db_host"])    # "server.database.windows.net"
print(config["tables"][0])  # First table name

# JSON arrays become Python lists
# JSON objects become Python dicts
# JSON strings → str, numbers → int/float, true/false → True/False, null → None

Writing JSON to a File

# json.dump() writes to a FILE object
pipeline_result = {
    "status": "success",
    "tables_loaded": 15,
    "rows_total": 1250000,
    "duration_seconds": 45,
    "errors": [],
    "timestamp": "2026-07-07T14:30:00"
}

with open("result.json", "w") as f:
    json.dump(pipeline_result, f, indent=2)  # indent=2 for readable formatting

# Without indent — single line (compact, good for APIs)
# With indent=2 — pretty-printed (good for config files and debugging)

JSON Strings (loads and dumps)

# json.loads() parses a JSON STRING (not a file)
json_str = '{"name": "Naveen", "age": 30, "active": true}'
data = json.loads(json_str)
print(data["name"])   # "Naveen"
print(data["active"]) # True (Python bool, not string "true")

# json.dumps() converts Python to a JSON STRING
data = {"name": "Naveen", "scores": [90, 85, 92], "active": True}
json_str = json.dumps(data)
print(json_str)  # '{"name": "Naveen", "scores": [90, 85, 92], "active": true}'

# Memory trick:
# load / dump → FILE (no s)
# loads / dumps → STRING (s = string)

Pretty Printing JSON

# indent for formatting, sort_keys for consistent output
data = {"name": "Naveen", "city": "Toronto", "skills": ["Python", "SQL", "Azure"]}

pretty = json.dumps(data, indent=2, sort_keys=True)
print(pretty)
# {
#   "city": "Toronto",
#   "name": "Naveen",
#   "skills": [
#     "Python",
#     "SQL",
#     "Azure"
#   ]
# }

# ensure_ascii=False for international characters
data = {"name": "Naveen", "greeting": "Namaste"}
print(json.dumps(data, ensure_ascii=False, indent=2))

Handling Nested JSON

# API responses are often deeply nested
response = {
    "data": {
        "users": [
            {"id": 1, "name": "Alice", "address": {"city": "Toronto"}},
            {"id": 2, "name": "Bob", "address": {"city": "London"}},
        ]
    },
    "meta": {"total": 2, "page": 1}
}

# Navigate nested structure
users = response["data"]["users"]
for user in users:
    name = user["name"]
    city = user["address"]["city"]
    print(f"{name} lives in {city}")

# Safe nested access (avoid KeyError)
city = response.get("data", {}).get("users", [{}])[0].get("address", {}).get("city", "Unknown")
print(city)  # "Toronto"

JSON with Non-Serializable Types

from datetime import datetime, date
from decimal import Decimal

# These types are NOT JSON serializable by default:
# datetime, date, Decimal, set, bytes, custom objects

# json.dumps({"date": datetime.now()})  # TypeError!

# Fix: custom serializer
def json_serializer(obj):
    if isinstance(obj, (datetime, date)):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    if isinstance(obj, set):
        return list(obj)
    raise TypeError(f"Type {type(obj)} is not JSON serializable")

data = {
    "timestamp": datetime.now(),
    "amount": Decimal("99.99"),
    "tags": {"python", "sql"},
}

json_str = json.dumps(data, default=json_serializer, indent=2)
print(json_str)
# {
#   "timestamp": "2026-07-07T14:30:00.123456",
#   "amount": 99.99,
#   "tags": ["python", "sql"]
# }

Working with Large Files

Line-by-Line Processing

# Process a 10 GB log file — only ONE line in memory at a time
total_errors = 0
with open("huge_log.txt", "r") as f:
    for line in f:
        if "ERROR" in line:
            total_errors += 1
print(f"Total errors: {total_errors}")
# Works on files of ANY size — 1 MB or 100 GB
# Memory usage stays constant regardless of file size

Chunked Reading

# Read binary files in chunks (good for copying large files)
def copy_file(src, dst, chunk_size=1024 * 1024):  # 1 MB chunks
    with open(src, "rb") as f_in, open(dst, "wb") as f_out:
        while True:
            chunk = f_in.read(chunk_size)
            if not chunk:
                break
            f_out.write(chunk)

# Read CSV in chunks with pandas (when you need DataFrames)
import pandas as pd
chunks = pd.read_csv("huge_file.csv", chunksize=10000)
for chunk_df in chunks:
    # Process 10,000 rows at a time
    clean = chunk_df.dropna()
    clean.to_csv("output.csv", mode="a", header=False, index=False)

Counting Lines Without Loading

# Count lines efficiently — never loads the whole file
def count_lines(filepath):
    count = 0
    with open(filepath, "r") as f:
        for _ in f:
            count += 1
    return count

total = count_lines("huge_file.csv")
print(f"Total lines: {total:,}")

# Even faster with sum + generator
def count_lines_fast(filepath):
    with open(filepath, "r") as f:
        return sum(1 for _ in f)

Encoding and Unicode

What Is Encoding

Encoding is how text is converted to bytes for storage. UTF-8 is the universal standard — it supports every character in every language. latin-1 (ISO-8859-1) handles Western European characters. cp1252 is Windows’ default. When encoding is wrong, you see garbled text: é instead of e, or ??? for non-ASCII characters.

# Always specify encoding explicitly — never rely on system default
with open("data.csv", "r", encoding="utf-8") as f:
    content = f.read()

# If you get UnicodeDecodeError, the file is not UTF-8
# Try common alternatives:
try:
    with open("data.csv", "r", encoding="utf-8") as f:
        content = f.read()
except UnicodeDecodeError:
    with open("data.csv", "r", encoding="latin-1") as f:
        content = f.read()
    print("Warning: file is not UTF-8, read as latin-1")

Common Encoding Problems

# Problem 1: Source system exports in cp1252, your pipeline expects UTF-8
# Fix: specify encoding when reading
with open("windows_export.csv", "r", encoding="cp1252") as f:
    data = f.read()

# Problem 2: Writing UTF-8 but forgetting to specify encoding
# On Windows, the default might be cp1252 — silently corrupts non-ASCII characters
with open("output.csv", "w", encoding="utf-8") as f:  # Always specify!
    f.write("Naveen
")

# Problem 3: BOM (Byte Order Mark) in UTF-8 files from Excel
# Excel saves CSV as "UTF-8 with BOM" — the BOM character appears as garbage
with open("excel_export.csv", "r", encoding="utf-8-sig") as f:  # utf-8-sig strips BOM
    reader = csv.DictReader(f)
    for row in reader:
        print(row)

Detecting and Fixing Encoding Issues

# Read as bytes first, then decode
with open("mystery.csv", "rb") as f:
    raw_bytes = f.read(1000)  # Read first 1000 bytes

print(raw_bytes[:50])  # See the raw bytes — look for patterns

# Use chardet to auto-detect encoding (pip install chardet)
import chardet
result = chardet.detect(raw_bytes)
print(result)  # {'encoding': 'ISO-8859-1', 'confidence': 0.73}

# Then read with the detected encoding
with open("mystery.csv", "r", encoding=result["encoding"]) as f:
    content = f.read()

File Operations — Copy, Move, Delete

import shutil
from pathlib import Path

# Copy a file
shutil.copy2("source.csv", "backup.csv")          # Preserves metadata
shutil.copy2("source.csv", "/data/archive/")       # Copy to directory

# Copy an entire directory
shutil.copytree("/data/bronze/", "/data/bronze_backup/")

# Move/rename a file
shutil.move("temp.csv", "final.csv")                  # Rename
shutil.move("report.csv", "/data/archive/report.csv") # Move

# Delete a file
Path("temp.csv").unlink(missing_ok=True)  # No error if file does not exist

# Delete a directory (even if not empty)
shutil.rmtree("/data/temp/", ignore_errors=True)

# Get file size
size = Path("data.csv").stat().st_size
print(f"File size: {size:,} bytes ({size / 1024 / 1024:.1f} MB)")

Data Engineering Patterns

Config File Reader

import json
from pathlib import Path

def load_config(config_path="config.json"):
    """Load pipeline configuration from JSON file."""
    path = Path(config_path)
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {config_path}")

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

    # Validate required keys
    required = ["source_host", "target_path", "tables"]
    missing = [k for k in required if k not in config]
    if missing:
        raise ValueError(f"Missing config keys: {missing}")

    return config

# Usage
config = load_config("pipeline_config.json")

Processing All Files in a Directory

from pathlib import Path
import csv

def process_csv_directory(input_dir, output_file):
    """Read all CSVs in a directory and combine into one output file."""
    input_path = Path(input_dir)
    csv_files = sorted(input_path.glob("*.csv"))

    if not csv_files:
        print(f"No CSV files found in {input_dir}")
        return

    total_rows = 0
    header_written = False

    with open(output_file, "w", newline="", encoding="utf-8") as out_f:
        for csv_file in csv_files:
            print(f"Processing: {csv_file.name}")
            with open(csv_file, "r", encoding="utf-8") as in_f:
                reader = csv.reader(in_f)
                header = next(reader)

                if not header_written:
                    writer = csv.writer(out_f)
                    writer.writerow(header)
                    header_written = True

                for row in reader:
                    writer.writerow(row)
                    total_rows += 1

    print(f"Combined {len(csv_files)} files → {total_rows:,} rows → {output_file}")

process_csv_directory("/data/incoming/", "/data/combined/all_customers.csv")

Safe File Writer (Temp File Pattern)

import tempfile
import shutil
from pathlib import Path

def safe_write(filepath, content):
    """Write to a temp file first, then rename — prevents partial writes."""
    target = Path(filepath)

    # Write to temp file in the same directory
    with tempfile.NamedTemporaryFile(
        mode="w", dir=target.parent, suffix=".tmp", delete=False
    ) as tmp:
        tmp.write(content)
        tmp_path = tmp.name

    # Atomic rename — either the file is complete or unchanged
    shutil.move(tmp_path, filepath)

# Why? If the process crashes DURING writing, the original file is intact.
# Without this pattern, a crash mid-write leaves a half-written, corrupted file.

File-Based Logging

from datetime import datetime

def log_to_file(message, log_file="pipeline.log"):
    """Append a timestamped log entry to a file."""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(log_file, "a", encoding="utf-8") as f:
        f.write(f"{timestamp} | {message}
")

# Usage
log_to_file("Pipeline started")
log_to_file("Loaded 15 tables, 1.2M rows")
log_to_file("Pipeline completed in 45 seconds")

Common Mistakes

1. Not using the with statement — f = open(...) without with leaks file handles if an exception occurs. Always use with open(...) as f:. There is no valid reason to skip it.
2. Using "w" when you mean "a" — "w" erases the entire file before writing. If you want to add to an existing file, use "a" (append). This mistake silently destroys data.
3. Reading a huge file with read() or readlines() — loads the entire file into memory. A 10 GB file needs 10 GB of RAM. Iterate line by line instead: for line in f:.
4. Not specifying encoding — open("file.csv") uses the system default encoding, which varies by OS. Always specify encoding="utf-8" for consistent behavior across platforms.
5. Forgetting newline="" in csv.writer on Windows — without it, CSV files get extra blank lines between rows on Windows. Always include newline="" in the open() call.
6. Confusing json.load vs json.loads — load reads from a file object. loads parses a string. Using the wrong one gives a confusing TypeError. Memory trick: the s in loads stands for string.
7. Using string concatenation for file paths — "data/" + "bronze/" + "file.csv" breaks on Windows (needs backslashes). Use pathlib.Path with the / operator for cross-platform paths.

Interview Questions

Q: What is a context manager and why should you use one for file handling? A: A context manager (the with statement) automatically closes the file when the block exits — even if an exception occurs. Without it, forgetting to call f.close() leaks file handles, which can crash long-running processes with “Too many open files” errors. The with statement guarantees cleanup.

Q: What is the difference between "w" and "a" file modes? A: "w" (write) creates a new file or erases an existing file’s content before writing. "a" (append) creates a new file or adds to the end of an existing file without erasing. Use "w" for fresh output files. Use "a" for log files and incremental data collection.

Q: How do you read a very large file without running out of memory? A: Iterate over the file object directly: for line in f:. The file object is a lazy iterator that reads one line at a time and discards it after processing. Memory usage stays constant regardless of file size. Never use read() or readlines() on large files.

Q: What is the difference between json.load() and json.loads()? A: json.load(f) reads JSON from a file object. json.loads(s) parses JSON from a string. Similarly, json.dump(data, f) writes to a file and json.dumps(data) returns a string. The s suffix stands for “string.”

Q: Why should you use pathlib instead of os.path? A: pathlib provides an object-oriented API that is more readable (path.name vs os.path.basename(path)), cross-platform (handles forward/backward slashes automatically), and supports the / operator for joining paths. It is the modern standard since Python 3.4.

Q: What is encoding and why does it matter for file handling? A: Encoding defines how text characters are converted to bytes for storage. UTF-8 is the universal standard supporting all languages. If you read a file with the wrong encoding, you get garbled text or UnicodeDecodeError. Always specify encoding="utf-8" explicitly, and use utf-8-sig for Excel-exported CSVs with BOM markers.

Wrapping Up

File handling is the foundation of every data pipeline. Master the with statement for safe resource management, pathlib for cross-platform file paths, csv.DictReader and csv.DictWriter for structured data, and json.load/json.dump for configuration and API data. For large files, always iterate line by line — never load the entire file into memory. And always specify encoding="utf-8" explicitly.

These patterns carry directly into data engineering work — reading configs, processing incoming files, writing pipeline outputs, and building ETL scripts that handle real-world data at scale.

Previous in this series: Lambda Functions, map(), filter(), reduce() (Section 1)

Next in this series: Error Handling — try/except/finally, Custom Exceptions

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.