CSV Processing
Reading CSV Files
import csv
# Basic reading
with open('data.csv', 'r') as f:
reader = csv.reader(f)
header = next(reader) # First row
for row in reader:
print(row) # ['value1', 'value2', 'value3']
# Reading as dictionary
with open('data.csv', 'r') as f:
reader = csv.DictReader(f)
for row in reader:
print(row['name'], row['age']) # Access by column name
Writing CSV Files
import csv
# Basic writing
with open('output.csv', 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['Name', 'Age', 'City']) # Header
writer.writerow(['Alice', 30, 'NYC'])
writer.writerow(['Bob', 25, 'LA'])
# Writing as dictionary
with open('output.csv', 'w', newline='') as f:
fieldnames = ['name', 'age', 'city']
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({'name': 'Alice', 'age': 30, 'city': 'NYC'})
CSV with Different Formats
import csv
# Tab-separated
with open('data.tsv', 'r') as f:
reader = csv.reader(f, delimiter='\t')
for row in reader:
print(row)
# Custom delimiter
with open('data.csv', 'r') as f:
reader = csv.reader(f, delimiter='|')
for row in reader:
print(row)
# Quoting
with open('data.csv', 'r') as f:
reader = csv.reader(f, quotechar='"', quoting=csv.QUOTE_MINIMAL)
for row in reader:
print(row)
Large CSV Files
import csv
# Process line by line (memory efficient)
def process_large_csv(filename):
with open(filename, 'r') as f:
reader = csv.DictReader(f)
for row in reader: # Lazy iteration
process(row)
# Using pandas for large files
import pandas as pd
# Read in chunks
for chunk in pd.read_csv('large.csv', chunksize=10000):
process(chunk)
# Or specify dtypes to reduce memory
df = pd.read_csv('data.csv', dtype={'id': 'int32', 'name': 'str'})
CSV Best Practices
# ✅ Always specify newline='' when writing
with open('output.csv', 'w', newline='') as f:
pass
# ✅ Use DictReader/DictWriter for named columns
# ✅ Handle encoding (utf-8-sig for Excel)
with open('data.csv', 'r', encoding='utf-8-sig') as f:
pass
# ✅ Validate data before processing
# ❌ Don't trust CSV data - sanitize input
JSON Processing
Reading JSON
import json
# From string
data = json.loads('{"name": "Alice", "age": 30}')
print(data) # {'name': 'Alice', 'age': 30}
# From file
with open('data.json', 'r') as f:
data = json.load(f)
print(data)
Writing JSON
import json
# To string
data = {'name': 'Alice', 'age': 30}
json_str = json.dumps(data)
print(json_str) # {"name": "Alice", "age": 30}
# Pretty print
json_str = json.dumps(data, indent=2)
print(json_str)
# {
# "name": "Alice",
# "age": 30
# }
# To file
with open('output.json', 'w') as f:
json.dump(data, f, indent=2)
JSON Types
import json
# Python -> JSON
# dict -> object
# list, tuple -> array
# str -> string
# int, float -> number
# True -> true
# False -> false
# None -> null
data = {
'name': 'Alice',
'age': 30,
'scores': [95, 87, 92],
'active': True,
'address': None
}
json_str = json.dumps(data)
print(json_str)
# {"name": "Alice", "age": 30, "scores": [95, 87, 92], "active": true, "address": null}
JSON with Custom Objects
import json
from datetime import datetime
# Custom encoder
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
data = {
'name': 'Alice',
'created_at': datetime.now()
}
json_str = json.dumps(data, cls=DateTimeEncoder, indent=2)
print(json_str)
# Custom decoder
def datetime_decoder(dct):
for key, value in dct.items():
if isinstance(value, str) and 'T' in value:
try:
dct[key] = datetime.fromisoformat(value)
except ValueError:
pass
return dct
# Use with object_hook
data = json.loads(json_str, object_hook=datetime_decoder)
JSON Best Practices
# ✅ Use indent for readability
json.dumps(data, indent=2)
# ✅ Use sort_keys for consistent output
json.dumps(data, sort_keys=True)
# ✅ Handle errors
try:
data = json.loads(invalid_json)
except json.JSONDecodeError as e:
print(f'Invalid JSON: {e}')
# ✅ Use appropriate encoding
with open('data.json', 'r', encoding='utf-8') as f:
data = json.load(f)
# ❌ Don't use eval() on JSON
# ❌ Don't trust untrusted JSON
Data Processing Patterns
CSV to JSON
import csv
import json
def csv_to_json(csv_file, json_file):
with open(csv_file, 'r') as f:
reader = csv.DictReader(f)
data = list(reader)
with open(json_file, 'w') as f:
json.dump(data, f, indent=2)
csv_to_json('data.csv', 'data.json')
JSON to CSV
import csv
import json
def json_to_csv(json_file, csv_file):
with open(json_file, 'r') as f:
data = json.load(f)
if not data:
return
with open(csv_file, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
json_to_csv('data.json', 'data.csv')
Processing Large Files
import csv
import json
from typing import Iterator
# Generator for large CSV
def read_large_csv(filename: str) -> Iterator[dict]:
with open(filename, 'r') as f:
reader = csv.DictReader(f)
for row in reader:
yield row
# Process in chunks
def process_in_chunks(filename: str, chunk_size: int = 1000):
chunk = []
for row in read_large_csv(filename):
chunk.append(row)
if len(chunk) >= chunk_size:
process_chunk(chunk)
chunk = []
if chunk:
process_chunk(chunk)
# Streaming JSON
def stream_json(filename: str):
with open(filename, 'r') as f:
decoder = json.JSONDecoder()
content = f.read()
idx = 0
while idx < len(content):
obj, end = decoder.raw_decode(content, idx)
yield obj
idx = end
Error Handling
import csv
import json
def safe_process_csv(filename: str):
try:
with open(filename, 'r') as f:
reader = csv.DictReader(f)
for i, row in enumerate(reader, start=2): # Line 2+ (1-indexed, header is line 1)
try:
process(row)
except KeyError as e:
print(f'Line {i}: Missing column {e}')
except ValueError as e:
print(f'Line {i}: Invalid value - {e}')
except FileNotFoundError:
print(f'File not found: {filename}')
except csv.Error as e:
print(f'CSV error: {e}')
# JSON validation
def validate_json(json_str: str) -> bool:
try:
json.loads(json_str)
return True
except json.JSONDecodeError:
return False