Python Stdlib for Data Engineers cheat sheet

pathlib, itertools, collections, datetime, serialization, and concurrency patterns that replace most one-off pipeline utilities.

ProgrammingIntermediate7 sections

Paths and files

Path("staging") / run_date.isoformat() / "orders.parquet": The slash operator builds platform-correct paths, which is what keeps a pipeline written on Windows working on a Linux runner.
p.mkdir(parents=True, exist_ok=True): The idempotent directory create. Without exist_ok two concurrent tasks writing the same partition race and one of them raises.
with tempfile.TemporaryDirectory() as tmp: out = Path(tmp) / "staging.parquet": Cleans up even when the task raises. On long-lived workers, leftover temp files are one of the most common causes of a full disk.
os.replace(tmp_path, final_path): An atomic rename within one filesystem, the write-then-swap trick that stops a reader from ever seeing a half-written file.
shutil.disk_usage("/tmp").free: Check before spilling a large sort or shuffle to local disk. Container scratch space is often far smaller than the memory limit implies.

itertools.batched(rows, 1000): Fixed-size batches from any iterable, added in Python 3.12. The final batch is short unless you pass strict=True, available since 3.13.
itertools.islice(reader, 1, None): Skips the header without materializing the file. Slicing an iterator this way keeps memory flat no matter how large the source is.
itertools.chain.from_iterable(pages): Flattens one level lazily, which is the right way to consume a paginated API. sum(pages, []) does the same thing in quadratic time.
for key, group in itertools.groupby(rows, key=itemgetter("customer_id")): ...: Groups only consecutive runs, so the input must already be sorted by the key. This is the single most common stdlib misuse in ETL code.
a, b = itertools.tee(stream, 2): Two independent views of one iterator, but tee buffers everything the faster branch has consumed. Reading the source twice is usually cheaper.

buckets = collections.defaultdict(list): Removes the setdefault dance when partitioning rows in memory. Reading a missing key creates it, so never use one for lookups you also probe.
collections.Counter(r["status"] for r in rows).most_common(5): One-line cardinality profiling over a stream. most_common with no argument sorts the entire mapping, which is wasteful on high-cardinality columns.
recent_errors = collections.deque(maxlen=1000): A fixed-size ring buffer with constant-time appends at both ends. The right structure for keeping the last N bad records without unbounded growth.
@functools.lru_cache(maxsize=1024): Memoizes dimension lookups. Arguments must be hashable, and the cache lives per process, so it does nothing across Spark executors or forked workers.
@dataclasses.dataclass(frozen=True, slots=True) class TableSpec: name: str partition_keys: tuple[str, ...] = (): frozen makes the config hashable and safe to share across threads, slots cuts per-instance memory. Mutable defaults need field(default_factory=list).

datetime.now(timezone.utc): utcnow is deprecated as of Python 3.12 because it returned a naive datetime that looked like UTC. Carry tzinfo end to end.
ZoneInfo("America/New_York"): The stdlib IANA database since 3.9, so pytz is no longer needed. On Windows there is no system database, so install the tzdata package.
datetime.fromisoformat("2026-08-30T12:00:00Z"): Python 3.11 taught this the full ISO 8601 grammar including the trailing Z. On 3.10 and earlier the exact same string raises ValueError.
dt.astimezone(ZoneInfo("UTC")).isoformat(): Convert first and serialize second, never the reverse. A naive datetime silently assumes the machine zone, which differs between laptop and container.
run_date - timedelta(days=1): timedelta has no month unit because months are not a fixed length. Do calendar arithmetic on explicit year and month integers instead of guessing 30 days.

with gzip.open("events.json.gz", "rt", encoding="utf-8") as f: rows = (json.loads(line) for line in f): Streams NDJSON without decompressing to disk. Text mode is required, otherwise the parser receives bytes and fails on the first line.
json.dumps(obj, default=str, sort_keys=True, separators=(",", ":")): default=str rescues datetimes and Decimals. sort_keys with compact separators makes the output byte-stable, which is what content hashing depends on.
with open("out.csv", "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore") w.writeheader(): newline="" is mandatory or Windows writes a blank line between every row. extrasaction stops a crash when a dict carries an unexpected key.
csv.field_size_limit(10**7): The default cap is 128 KB per field and raises on a single embedded JSON blob. Raise the limit rather than pre-splitting the file.
hashlib.sha256(canonical_bytes).hexdigest(): Deterministic row fingerprints for change detection. Hash a canonical serialization, never repr or str, which are not stable across Python versions.

with ThreadPoolExecutor(max_workers=16) as ex: results = list(ex.map(fetch, urls)): Threads are correct for API and object store calls because the GIL is released during I/O. ex.map re-raises the first exception when you iterate it.
futures = {ex.submit(load, t): t for t in tables} for fut in as_completed(futures): ...: Handles results as they finish and lets you attribute a failure back to its input. With map you lose which item raised.
ProcessPoolExecutor(max_workers=os.cpu_count()): For CPU-bound parsing and compression. Arguments and returns are pickled, so shipping large frames between processes often costs more than it saves.
sem = asyncio.Semaphore(20): Bounds concurrency against a rate-limited API. Without it asyncio.gather will happily open ten thousand sockets and get the whole job throttled.
concurrent.futures.wait(futures, timeout=600): A pool with no timeout hangs the pipeline forever on one stuck socket. Set an explicit timeout on the pool and on every HTTP client underneath it.

logger = logging.getLogger(__name__): Module-level loggers inherit handlers from the root, which is what lets an orchestrator capture library output without any extra wiring.
logger.info("loaded table", extra={"rows": n, "table": name}): Structured fields survive into JSON log shipping and stay queryable. Formatting the count into the message string throws that away.
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")): Call it once at the entry point only. A library that calls basicConfig itself silently steals your formatting for the whole process.
subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=600): Without check=True a failing command returns quietly and the task reports success. Never use shell=True with an interpolated value.
parser.add_argument("--run-date", type=date.fromisoformat, required=True): Parsing the type at the argument boundary means bad input fails in milliseconds instead of forty minutes into an extract.

From DataLane — tutorials at/blog, practice SQL live in theplayground.