Standard library tour: itertools, functools, pathlib, and friends

from collections import Counter, defaultdict, deque, OrderedDict, ChainMap, namedtuple

# Frequency
Counter("abracadabra").most_common(3)         # [('a', 5), ('b', 2), ('r', 2)]

# Grouping
buckets = defaultdict(list)
for w in words: buckets[len(w)].append(w)

# Fixed-size queue (rolling window)
window = deque(maxlen=10)
for x in stream: window.append(x); use(list(window))

# Layered config
cfg = ChainMap(env, file_cfg, defaults)

from itertools import chain, accumulate, count, cycle, islice, groupby, batched, pairwise

list(chain([1,2], [3,4]))                     # [1,2,3,4]
list(accumulate([1,2,3,4]))                   # [1,3,6,10] (cumulative)
list(islice(count(10, 2), 5))                 # [10,12,14,16,18]
list(pairwise([1,2,3,4]))                     # [(1,2),(2,3),(3,4)]
list(batched(range(10), 3))                   # [(0,1,2),(3,4,5),(6,7,8),(9,)]   (3.12+)

for k, group in groupby(sorted(items, key=key), key=key):
    process(k, list(group))

from functools import partial, reduce, lru_cache, cache, wraps, cached_property, singledispatch, total_ordering

@cache
def fib(n): return n if n < 2 else fib(n-1) + fib(n-2)

@lru_cache(maxsize=1024)
def slow(x): ...

multiply_by_3 = partial(operator.mul, 3)

@singledispatch
def serialise(x): raise TypeError
@serialise.register
def _(x: int): return str(x)
@serialise.register
def _(x: list): return "[" + ",".join(serialise(i) for i in x) + "]"

@total_ordering
class Version: ...                            # define __eq__ + __lt__; get the rest

from datetime import datetime, date, time, timedelta, timezone
from zoneinfo import ZoneInfo

# Now — always be timezone-aware
now = datetime.now(timezone.utc)
ny = datetime.now(ZoneInfo("America/New_York"))

# Construct
d = date(2026, 6, 7)
dt = datetime(2026, 6, 7, 14, 30, tzinfo=ZoneInfo("UTC"))

# Arithmetic
dt + timedelta(days=7, hours=3)
later - earlier                               # timedelta

# Convert tz
ny_time = dt.astimezone(ZoneInfo("America/New_York"))

# Format / parse
dt.isoformat()                                # "2026-06-07T14:30:00+00:00"
datetime.fromisoformat("2026-06-07T14:30:00+00:00")
dt.strftime("%Y-%m-%d %H:%M")
datetime.strptime("2026-06-07 14:30", "%Y-%m-%d %H:%M")

import time
time.sleep(0.5)                               # seconds
time.monotonic()                              # for measuring intervals
time.perf_counter()                           # highest precision; for benchmarking

import os, sys

os.environ.get("OPENAI_API_KEY", "")
os.environ["MY_VAR"] = "x"                   # set
os.getcwd()                                   # use Path.cwd() instead
os.cpu_count()                                # logical cores
os.urandom(16)                                # 16 bytes of entropy (use `secrets` instead)

sys.argv                                       # list of CLI args
sys.exit(0)                                    # exit
sys.platform                                   # 'linux', 'darwin', 'win32'
sys.version_info                               # (3, 12, 4, ...)
sys.stdin / sys.stdout / sys.stderr            # streams

import subprocess

# Simple: check exit code, capture output
r = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True)
print(r.stdout.strip())                       # commit SHA

# Don't use shell=True with user input (shell injection)
# bad:  subprocess.run(f"grep {user_input} file.txt", shell=True)
# good: subprocess.run(["grep", user_input, "file.txt"])

# Stream output as it arrives
proc = subprocess.Popen(
    ["python", "long_script.py"],
    stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
    text=True, bufsize=1,
)
for line in proc.stdout:
    print(line, end="")
proc.wait()
if proc.returncode:
    raise RuntimeError("script failed")

from tempfile import TemporaryDirectory, NamedTemporaryFile, mkstemp
import shutil

with TemporaryDirectory() as tmp:
    work = Path(tmp) / "work"
    work.mkdir()
    ...
# directory auto-deleted

with NamedTemporaryFile(delete=False, suffix=".csv") as f:
    f.write(b"a,b,c\n")
    path = f.name
# delete manually if delete=False

shutil.copy(src, dst); shutil.copytree(src, dst)
shutil.move(src, dst)
shutil.rmtree(path)
shutil.disk_usage(".").free
shutil.which("git")                           # locate executable
shutil.make_archive("backup", "zip", "src/")

import hashlib, secrets, hmac

hashlib.sha256(b"data").hexdigest()
hashlib.blake2b(b"data", digest_size=16).hexdigest()  # faster than SHA-512

# Cryptographic randomness — for tokens, passwords, IDs
secrets.token_urlsafe(32)                     # URL-safe random string
secrets.token_hex(16)                         # 32 hex chars
secrets.choice(["a", "b", "c"])
secrets.compare_digest("expected", "candidate")    # constant-time

# HMAC signature
mac = hmac.new(key, msg, hashlib.sha256).hexdigest()
hmac.compare_digest(mac, received_mac)

import argparse

p = argparse.ArgumentParser(description="ETL")
p.add_argument("input", type=Path, help="input file")
p.add_argument("-o", "--output", type=Path, default=Path("out.json"))
p.add_argument("--limit", type=int, default=1000)
p.add_argument("-v", "--verbose", action="store_true")
args = p.parse_args()

print(args.input, args.output, args.limit, args.verbose)

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

# I/O-bound: threads
with ThreadPoolExecutor(max_workers=20) as ex:
    futures = {ex.submit(fetch, url): url for url in urls}
    for fut in as_completed(futures):
        url = futures[fut]
        try: data = fut.result()
        except Exception as e: log.warning("%s failed: %s", url, e)
        else: process(data)

# CPU-bound: processes
with ProcessPoolExecutor() as ex:
    for result in ex.map(heavy_compute, items, chunksize=100):
        ...

from enum import Enum, auto, IntEnum, StrEnum                   # StrEnum: 3.11+

class Color(Enum):
    RED = auto()
    GREEN = auto()
    BLUE = auto()

class HttpStatus(IntEnum):
    OK = 200
    NOT_FOUND = 404

class Mode(StrEnum):
    READ = "r"
    WRITE = "w"

Color.RED.name, Color.RED.value
HttpStatus(404) is HttpStatus.NOT_FOUND
"r" == Mode.READ                              # True (StrEnum is also str)

import hashlib
from collections import defaultdict
from pathlib import Path

def hash_file(p: Path, chunk: int = 65536) -> str:
    h = hashlib.blake2b(digest_size=16)
    with p.open("rb") as f:
        while data := f.read(chunk):
            h.update(data)
    return h.hexdigest()

def find_duplicates(root: Path) -> dict[str, list[Path]]:
    buckets: dict[str, list[Path]] = defaultdict(list)
    for p in root.rglob("*"):
        if p.is_file():
            buckets[hash_file(p)].append(p)
    return {h: ps for h, ps in buckets.items() if len(ps) > 1}

if __name__ == "__main__":
    import sys
    root = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
    for h, paths in find_duplicates(root).items():
        print(h)
        for p in paths: print(" ", p)