File I/O & Context Managers
15 questions found
How would you implement a context manager for managing a lock (like threading.Lock) while ALSO adding a timeout, raising an error if the lock can't be acquired quickly enough?
Advanced
Implement __enter__ to attempt lock.acquire(timeout=N), raising a custom TimeoutError if it fails, and __exit__ to always release the lock — combining the standard locking pattern with custom timeout-handling logic that plain 'with lock:' doesn't provide out of the box.
import threading
class TimedLock:
def __init__(self, lock, timeout):
self.lock = lock
self.timeout = timeout
def __enter__(self):
if not self.lock.acquire(timeout=self.timeout):
raise TimeoutError("Could not acquire lock in time")
return self.lock
def __exit__(self, *args):
self.lock.release()
lock = threading.Lock()
with TimedLock(lock, timeout=5):
critical_section()
Real-world example
Preventing a thread from hanging indefinitely waiting for a lock, instead failing fast with a clear timeout error after a reasonable wait.
Common follow-ups: Why does __exit__ need to accept *args (exc_type, exc_value, traceback) even if it doesn't use them?
Concurrency (asyncio/threading/multiprocessing)
How do you correctly handle character encoding when reading text files that might not be UTF-8, and what error occurs if you guess wrong?
Advanced
Explicitly pass the 'encoding' parameter to open() (e.g., encoding='latin-1' or 'utf-8') rather than relying on the platform-dependent default — reading a file with the WRONG encoding raises a UnicodeDecodeError (for incompatible byte sequences) or silently produces GARBLED text (for compatible-but-wrong encodings that don't error).
try:
with open("legacy_data.txt", encoding="utf-8") as f:
content = f.read()
except UnicodeDecodeError:
with open("legacy_data.txt", encoding="latin-1") as f: # fallback encoding
content = f.read()
Real-world example
Reading a legacy data file of unknown or non-UTF-8 encoding without crashing or silently corrupting non-ASCII characters.
Common follow-ups: Why does Python's open() NOT default to UTF-8 consistently across all platforms in older versions, and how does PEP 686 address this?
Debugging & Profiling
How would you implement asynchronous file I/O using aiofiles alongside an async context manager, to avoid blocking the event loop during file operations?
Advanced
Standard Python file I/O is BLOCKING (no native async file API in the standard library), so aiofiles wraps file operations in a thread pool internally while exposing an async context manager interface (`async with aiofiles.open(...) as f:`), letting file reads/writes coexist with other async work without blocking the event loop.
import aiofiles
async def read_file_async(path):
async with aiofiles.open(path, mode="r") as f:
content = await f.read()
return content
async def main():
content = await read_file_async("data.txt")
print(content)
Real-world example
Reading configuration or data files inside an async web server or scraper without blocking the event loop during disk I/O.
Common follow-ups: Why doesn't the Python standard library provide a NATIVE async file I/O API, unlike its native support for async networking?
Async Generators & Async Context Managers
How would you write a context manager that atomically writes to a file, avoiding leaving a corrupted or partially-written file if an exception occurs mid-write?
Advanced
Write to a TEMPORARY file first, and only rename it to the FINAL destination (an atomic OS-level operation) in __exit__ if NO exception occurred — if an exception did occur, clean up the temp file instead, ensuring the original (or no) file is ever left in a partially-written, corrupted state.
import os, tempfile
from contextlib import contextmanager
@contextmanager
def atomic_write(path):
tmp_fd, tmp_path = tempfile.mkstemp(dir=os.path.dirname(path) or ".")
try:
with os.fdopen(tmp_fd, "w") as f:
yield f
os.replace(tmp_path, path) # atomic rename, only on success
except Exception:
os.remove(tmp_path)
raise
with atomic_write("config.json") as f:
json.dump(data, f)
Real-world example
Safely updating a critical configuration file that other processes might read concurrently, without ever exposing a half-written state.
Common follow-ups: Why is os.replace() specifically chosen here over a plain rename or direct write to the final path?
Serialization (json
pickle)
How do you check whether a file or directory exists before operating on it, using pathlib?
Intermediate
Path.exists() returns True if the path exists at all (file or directory); Path.is_file() and Path.is_dir() narrow the check to specifically a regular file or a directory respectively, letting you branch logic based on exactly what kind of filesystem entry is present.
from pathlib import Path
p = Path("data.txt")
if p.exists() and p.is_file():
content = p.read_text()
else:
print("File not found")
Real-world example
Checking whether a configuration file exists before attempting to read it, falling back to defaults if it doesn't.
Common follow-ups: Is there a race condition risk between checking p.exists() and later opening the file, and how would EAFP-style code avoid it?
Exception Handling