Python File I/O — Descriptor Leaks Without `with`

import os

# --- Demonstrate all primary file modes with a realistic log file scenario ---

log_file_path = "app_events.log"


# MODE 'w': Write mode — creates the file fresh every time it is called.
# CRITICAL: If app_events.log already existed with content, that content is
# now gone. Python does not ask. It does not warn. It just truncates.
with open(log_file_path, "w") as log_file:
    log_file.write("[INFO] Application started\n")
    log_file.write("[INFO] Loading configuration from /etc/app/config.yml\n")
print("Step 1 — Written initial log entries with 'w' mode.")


# MODE 'a': Append mode — the only safe mode for growing log files.
# Adds to the END of the file without touching any existing content.
# If the file does not exist, it creates it. If it does exist, it adds to it.
# This is what you want for every logging, audit trail, and event recording use case.
with open(log_file_path, "a") as log_file:
    log_file.write("[INFO] User authenticated: alice@example.com\n")
    log_file.write("[WARN] Rate limit approaching for endpoint /api/orders\n")
print("Step 2 — Appended new log entries with 'a' mode. Existing content intact.")


# MODE 'r': Read-only mode — the safest mode for reading.
# Raises FileNotFoundError if the file does not exist, which protects you
# from silently processing an empty or default state.
with open(log_file_path, "r") as log_file:
    full_contents = log_file.read()  # entire file as one string — fine for small files
print("Step 3 — Full log file contents via 'r' mode:")
print(full_contents)


# MODE 'r+': Read + Write — file must exist (no creation), no truncation.
# The cursor starts at position 0. Use when you need to read state
# and then update it within the same file handle.
with open(log_file_path, "r+") as log_file:
    first_line = log_file.readline()  # reads just the first line, advances cursor
    print(f"Step 4 — First log entry: {first_line.strip()}")
    # Writing now happens after the cursor, not at the beginning
    log_file.write("[DEBUG] r+ mode writes at current cursor position\n")


# MODE 'x': Exclusive creation — fails if file already exists.
# Use this when you need to guarantee you are not overwriting an existing file.
# More defensive than 'w' for one-time setup files.
new_lock_path = "process.lock"
try:
    with open(new_lock_path, "x") as lock_file:
        lock_file.write(f"PID: {os.getpid()}\n")
    print("Step 5 — Created process lock file with 'x' mode.")
except FileExistsError:
    print("Step 5 — Lock file already exists — another process may be running.")


# Clean up demo files so this script is safe to re-run
for path in [log_file_path, new_lock_path]:
    if os.path.exists(path):
        os.remove(path)
print("Step 6 — Demo files removed.")

import os


# First, create a sample config file to work with
config_path = "server_config.txt"
with open(config_path, "w") as config_file:
    config_file.write("host=localhost\n")
    config_file.write("port=8080\n")
    config_file.write("debug=True\n")
    config_file.write("max_connections=100\n")


# ❌ THE RISKY WAY — bare open() with manual close()
# If any line between open() and close() raises an exception,
# close() is skipped. The file handle leaks into the process.
# Under enough load or enough failures, this exhausts the OS fd limit.
def read_config_risky(filepath):
    config_file = open(filepath, "r")      # handle is now open
    raw_content = config_file.read()       # what if the file is unreadable mid-read?
    config_file.close()                    # this line might NEVER execute
    return raw_content


# ✅ THE SAFE WAY — context manager guarantees cleanup on every exit path
# The file closes the instant the 'with' block ends, whether normally or on exception.
# You cannot accidentally leave it open — the protocol enforces closure.
def read_config_safe(filepath):
    with open(filepath, "r") as config_file:   # __enter__ opens, registers cleanup
        raw_content = config_file.read()
    # config_file.__exit__ has been called — the file is 100% closed here
    # The 'config_file' name is still in scope but using it raises ValueError:
    # 'I/O operation on closed file' — a clear error rather than a silent leak
    return raw_content


# ✅ OPENING MULTIPLE FILES IN ONE 'with' STATEMENT
# Both files are guaranteed to close even if one raises an exception mid-operation.
# Cleaner than nesting two 'with' blocks.
def merge_configs(primary_path, override_path, output_path):
    with open(primary_path, "r") as primary, \
         open(override_path, "r") as override, \
         open(output_path, "w") as merged:
        merged.write(primary.read())
        merged.write(override.read())


# ✅ PARSING CONFIG — turning a raw text file into a usable dictionary
# Line-by-line iteration keeps memory usage flat — important if config files
# ever grow beyond a few kilobytes (templates, include directives, etc.)
def parse_config(filepath):
    settings = {}
    with open(filepath, "r") as config_file:
        for line in config_file:               # reads one line at a time
            line = line.strip()                # removes leading/trailing whitespace and \n
            if not line or line.startswith("#"):  # skip blank lines and comments
                continue
            if "=" in line:
                # maxsplit=1 protects values that themselves contain '=' characters
                # Without it, 'url=https://host:8080' would split into three parts
                key, value = line.split("=", 1)
                settings[key.strip()] = value.strip()
    return settings


# Run the demonstrations
raw = read_config_safe(config_path)
print("Raw file content:")
print(raw)

parsed = parse_config(config_path)
print("Parsed config dictionary:")
for setting_key, setting_value in parsed.items():
    print(f"  {setting_key} → {setting_value}")

print()
print(f"Is config_file closed after 'with' block? True (cannot access it meaningfully)")

# Clean up
os.remove(config_path)

import os

sales_data_path = "quarterly_sales.csv"

# Create a realistic sample CSV file for the demonstrations
with open(sales_data_path, "w") as sales_file:
    sales_file.write("date,product,units_sold,revenue\n")
    sales_file.write("2024-01-15,Widget Pro,120,2400.00\n")
    sales_file.write("2024-01-22,Widget Pro,95,1900.00\n")
    sales_file.write("2024-02-03,Gadget Plus,200,6000.00\n")
    sales_file.write("2024-02-18,Widget Pro,150,3000.00\n")
    sales_file.write("2024-03-07,Gadget Plus,175,5250.00\n")


# STRATEGY 1: read() — entire file as one string
# Use when: small file (<10MB), you need the full content as a string,
# or you are hashing, templating, or comparing entire file contents.
# Avoid when: file could grow — every byte in the file costs one byte of RAM.
with open(sales_data_path, "r") as sales_file:
    entire_content = sales_file.read()
print("=== Strategy 1: read() ===")
print(f"Type: {type(entire_content).__name__}, Characters: {len(entire_content)}")
print()


# STRATEGY 2: readlines() — list of line strings, newlines included
# Use when: you need random access to lines by index (e.g., 'give me line 3').
# Avoid when: file is large — the entire file loads into RAM as a list.
with open(sales_data_path, "r") as sales_file:
    all_lines = sales_file.readlines()
print("=== Strategy 2: readlines() ===")
print(f"Type: {type(all_lines).__name__}, Line count: {len(all_lines)}")
print(f"Line at index 2 (raw):   '{all_lines[2]}'")
print(f"Line at index 2 (clean): '{all_lines[2].strip()}'")
print()


# STRATEGY 3: Line-by-line iteration — the correct default for file processing
# Use when: processing any file that could grow, filtering rows, aggregating data.
# Python reads in OS-level chunks internally — your memory usage is O(1).
# next(file) skips the header row without loading it into a structure you track.
def calculate_total_revenue(filepath):
    total_revenue = 0.0
    with open(filepath, "r") as sales_file:
        next(sales_file)                        # skip the header row cleanly
        for data_line in sales_file:            # reads one line at a time from OS buffer
            columns = data_line.strip().split(",")
            total_revenue += float(columns[3])  # index 3 is the 'revenue' column
    return total_revenue

total = calculate_total_revenue(sales_data_path)
print("=== Strategy 3: Line-by-line iteration ===")
print(f"Total revenue across all sales: ${total:,.2f}")
print()


# STRATEGY 4: readline() — one line at a time, explicit control
# Use when: reading a header separately, then processing the rest differently.
with open(sales_data_path, "r") as sales_file:
    header = sales_file.readline().strip()       # reads exactly the first line
    column_names = header.split(",")
    print("=== Strategy 4: readline() for header ===")
    print(f"Columns: {column_names}")
    first_data_line = sales_file.readline().strip()  # cursor is now at line 2
    print(f"First data row: {first_data_line}")
print()


# writelines() DEMO — no automatic newlines added
# Every string in the list must include '\n' explicitly.
# Forgetting '\n' merges all lines into one continuous string with no separators.
results_path = "revenue_summary.txt"
summary_lines = [
    "=== Q1 2024 Revenue Summary ===\n",   # \n is required — writelines() adds nothing
    f"Total Revenue: ${total:,.2f}\n",
    "Source: quarterly_sales.csv\n",
    "Generated by: io/thecodeforge/files/reading_strategies_demo.py\n",
]
with open(results_path, "w") as results_file:
    results_file.writelines(summary_lines)

with open(results_path, "r") as results_file:
    print("=== writelines() output ===")
    print(results_file.read())

os.remove(sales_data_path)
os.remove(results_path)

import os
from datetime import datetime

TASKS_FILE = "my_tasks.txt"
COMPLETED_MARKER = "[DONE]"
PENDING_MARKER = "[TODO]"


def load_tasks(filepath):
    """
    Read all tasks from disk. Returns an empty list if the file does not exist.
    First run behavior: no file means no tasks — a perfectly valid state.
    Using os.path.exists() rather than try/except here because we need to
    distinguish 'file does not exist' from 'file exists but cannot be read'.
    """
    if not os.path.exists(filepath):
        return []
    tasks = []
    with open(filepath, "r") as task_file:
        for raw_line in task_file:              # line-by-line: memory stays flat
            stripped = raw_line.strip()
            if stripped:                        # skip any blank lines
                tasks.append(stripped)
    return tasks


def add_task(filepath, task_description):
    """
    Append a new task to the file.
    Uses 'a' mode — no existing content is touched regardless of what happens.
    Safe to call concurrently from multiple processes (though not transactionally safe).
    """
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    task_entry = f"{PENDING_MARKER} [{timestamp}] {task_description}\n"
    with open(filepath, "a") as task_file:     # 'a' is the only correct mode here
        task_file.write(task_entry)
    print(f"  Added: '{task_description}'")


def complete_task(filepath, task_index):
    """
    Mark a task as complete. Requires rewriting the entire file because
    there is no efficient mechanism to modify a single line in place.
    
    Production-safe variant: write to a .tmp file first, then os.replace()
    to atomically swap it in. This prevents partial writes from corrupting
    the task file if the process is killed mid-write.
    """
    all_tasks = load_tasks(filepath)
    if task_index < 0 or task_index >= len(all_tasks):
        print(f"  No task at index {task_index}. Valid range: 0 to {len(all_tasks) - 1}")
        return
    target_task = all_tasks[task_index]
    if COMPLETED_MARKER in target_task:
        print(f"  Task {task_index} is already marked complete.")
        return
    # Update the marker in memory
    all_tasks[task_index] = target_task.replace(PENDING_MARKER, COMPLETED_MARKER, 1)
    
    # Atomic write pattern — safe against crashes mid-write
    temp_path = filepath + ".tmp"
    with open(temp_path, "w") as temp_file:    # write to temp first
        for updated_task in all_tasks:
            temp_file.write(updated_task + "\n")
    # os.replace() is atomic at the OS level — the tasks file is either the
    # old version or the new version; it is never half-written
    os.replace(temp_path, filepath)
    print(f"  Marked task {task_index} as complete.")


def display_tasks(filepath):
    """
    Print all tasks with their index so the user knows what to pass to complete_task.
    Includes summary statistics to make the display more useful.
    """
    all_tasks = load_tasks(filepath)
    if not all_tasks:
        print("  No tasks yet. Add some with add_task().")
        return
    done = sum(1 for t in all_tasks if COMPLETED_MARKER in t)
    pending = len(all_tasks) - done
    print(f"  Tasks ({done} complete, {pending} pending):")
    for index, task_line in enumerate(all_tasks):
        print(f"    [{index}] {task_line}")


# --- Simulated session ---
print("--- Adding tasks ---")
add_task(TASKS_FILE, "Write unit tests for the payment module")
add_task(TASKS_FILE, "Review PR #47 from the backend team")
add_task(TASKS_FILE, "Update README with new API endpoints")
add_task(TASKS_FILE, "Deploy hotfix to staging environment")

print("\n--- Current task list ---")
display_tasks(TASKS_FILE)

print("\n--- Completing tasks 1 and 3 ---")
complete_task(TASKS_FILE, 1)
complete_task(TASKS_FILE, 3)

print("\n--- Updated task list ---")
display_tasks(TASKS_FILE)

print("\n--- Testing edge cases ---")
complete_task(TASKS_FILE, 1)   # already done
complete_task(TASKS_FILE, 99)  # invalid index

# Clean up demo files
for path in [TASKS_FILE, TASKS_FILE + ".tmp"]:
    if os.path.exists(path):
        os.remove(path)
print("\n--- Demo complete. All demo files removed. ---")

// io.thecodeforge — python tutorial

import io

# Default buffering: 8KB buffer, flushes on overflow
with open('orders.log', 'w') as log:
    log.write('Order #4173: payment received\n')
    # Crash here — buffer not flushed, order lost

# Explicit flush after critical write
with open('orders.log', 'w', buffering=1) as log:  # line-buffered
    log.write('Order #4174: payment received\n')
    log.flush()  # forces buffer to OS, but OS may still cache

# Nuclear option: unbuffered + fsync
with open('critical.bin', 'wb', buffering=0) as f:
    f.write(b'\x00\x01')
    # No need to flush — unbuffered writes go straight to kernel
    os.fsync(f.fileno())  # force disk commit

// io.thecodeforge — python tutorial

import sys

# This file contains a byte 0x9A which is invalid UTF-8
with open('broken_data.txt', 'rb') as f:
    raw = f.read()

# Default: crashes
# with open('broken_data.txt', 'r') as f:  # UnicodeDecodeError!

# Safe: replace unknown bytes
with open('broken_data.txt', 'r', errors='replace') as f:
    content = f.read()
    print(repr(content))
    # Output: 'This has a \ufffd in it'

# Even safer for pipelines: preserve raw bytes via surrogateescape
with open('broken_data.txt', 'r', errors='surrogateescape') as f:
    content = f.read()
    # Back to bytes without loss
    restored_raw = content.encode('utf-8', errors='surrogateescape')
    assert raw == restored_raw

// io.thecodeforge — python tutorial

import os
import time

LOCK_PATH = '/tmp/process_log.lock'
LOG_PATH = 'shared.log'

def acquire_lock():
    while True:
        try:
            # O_CREAT | O_EXCL: atomic creation, fails if exists
            fd = os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            os.write(fd, str(os.getpid()).encode())
            os.close(fd)
            return
        except FileExistsError:
            time.sleep(0.1)

def release_lock():
    try:
        os.unlink(LOCK_PATH)
    except FileNotFoundError:
        pass

acquire_lock()
try:
    with open(LOG_PATH, 'a') as f:
        f.write(f'Process {os.getpid()} writing\n')
finally:
    release_lock()

// io.thecodeforge — python tutorial

import os
import sys

def safe_read(filepath):
    if not os.path.exists(filepath):
        raise FileNotFoundError(f"{filepath} missing — check config")
    if os.path.getsize(filepath) == 0:
        raise ValueError(f"{filepath} is empty — aborting")
    if not os.access(filepath, os.R_OK):
        raise PermissionError(f"No read permission for {filepath}")

    with open(filepath, 'r') as f:
        return f.read()

try:
    data = safe_read('/var/log/app.log')
    print(f"Read {len(data)} bytes successfully")
except (FileNotFoundError, ValueError, PermissionError) as e:
    print(f"Pre-check failed: {e}")
    sys.exit(1)

// io.thecodeforge — python tutorial

def append_log(filepath, message):
    # 'a' — never truncates
    with open(filepath, 'a') as f:
        f.write(f"{message}\n")

def safe_create(filepath):
    # 'x' — fails if exists
    try:
        with open(filepath, 'x') as f:
            f.write("initialized\n")
    except FileExistsError:
        print(f"{filepath} already exists — won't overwrite")

append_log('/var/log/app.log', 'User login')
safe_create('/tmp/lock.pid')

// io.thecodeforge — python tutorial

# Stream large files — never read() them whole
with open("10gb_log.txt") as f:
    for line in f:
        process(line)  # O(1) memory

# Avoid short writes by checking
written = file.write(data)
if written != len(data):
    raise IOError("Short write occurred")

# Use pathlib for safe path joining
from pathlib import Path
path = Path("/data") / "logs" / "app.log"

# Force fsync for critical data
file.flush()
os.fsync(file.fileno())

// io.thecodeforge — python tutorial

# Copy file with metadata in one call
import shutil
shutil.copy2("source.dat", "backup.dat")

# Walk directories cleanly
for root, dirs, files in os.walk("/var/log"):
    for f in files:
        print(os.path.join(root, f))

# Temporary file with auto-cleanup
import tempfile
with tempfile.NamedTemporaryFile(mode="w", delete=True) as tmp:
    tmp.write("temp data")
    tmp.flush()
    # Use tmp.name — file deleted on exit

# In-memory file for testing
from io import StringIO
fake_file = StringIO("line1\nline2")
for line in fake_file:
    print(line.strip())

// io.thecodeforge — python tutorial
import configparser
from pathlib import Path

config = configparser.ConfigParser()
config.read('app.cfg')

# Use pathlib for safe file I/O
path = Path('data.zst')
# Never re-implement config parsing yourself

// io.thecodeforge — python tutorial
with open('data.bin', 'rb') as f:
    f.seek(1024)  # Jump to position
    chunk = f.read(64)
    pos = f.tell()  # Remember position

# Memory-map for wizard-level speed
import mmap
with open('large.bin', 'r+b') as f:
    with mmap.mmap(f.fileno(), 0) as mm:
        mm[0:4] = b'\x00\x00\x00\x01'
        byte = mm[500]