Ch 10 — Standard Library & Capstones
Ch 10 — Standard Library & Capstones
Python’s standard library is enormous — and Aryan doesn’t need all of it, but he definitely needs the modules that handle configuration files, timestamps, file paths, hashing, structured data, and logging. This chapter tours the modules that elevate the RAM manager from a script to production-quality software.
json — Config & Data Exchange
Read and write JSON configuration files.
import json
# Write config
config = {"threshold_pct": 80, "poll_interval_s": 1, "log_path": "/tmp/ram.log"}
with open("config.json", "w") as f:
json.dump(config, f, indent=2)
# Read config
with open("config.json") as f:
cfg = json.load(f)
print(cfg["threshold_pct"]) # 80
# Serialize to string
json_str = json.dumps(config)
cfg2 = json.loads(json_str)re — Regular Expressions
Parse process names, filter log lines, validate input.
import re
# Extract PID from a ps-style output line
line = " 4821 chrome 1024 MB"
m = re.search(r"^\s*(\d+)\s+(\w+)", line)
if m:
pid, name = m.group(1), m.group(2)
# Find all hex addresses in a memory dump
addrs = re.findall(r"0x[0-9a-fA-F]+", "addr 0xDEAD, addr 0xBEEF")
# ['0xDEAD', '0xBEEF']
# Validate threshold input
re.fullmatch(r"\d{1,3}", "85") # Match object — valid
re.fullmatch(r"\d{1,3}", "abc") # None — invaliddatetime — Timestamps
Tag every snapshot with a time.
from datetime import datetime, timedelta
now = datetime.now()
print(now.isoformat()) # '2024-06-15T14:30:00.123456'
print(now.strftime("%H:%M:%S")) # '14:30:00'
# Time deltas
one_minute_ago = now - timedelta(minutes=1)
print((now - one_minute_ago).seconds) # 60pathlib — File Paths
Modern, OS-agnostic path handling.
from pathlib import Path
log_dir = Path.home() / ".ram_manager" / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
log_file = log_dir / "session.log"
log_file.write_text("RAM Manager started\n")
content = log_file.read_text()
# Glob for all log files
for f in log_dir.glob("*.log"):
print(f.name, f.stat().st_size)csv — Report Files
Export snapshots to spreadsheet-friendly format.
import csv
rows = [
{"time": "14:30:00", "process": "chrome", "rss_mb": 1024.5},
{"time": "14:30:01", "process": "electron", "rss_mb": 512.0},
]
with open("report.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["time", "process", "rss_mb"])
writer.writeheader()
writer.writerows(rows)hashlib — Snapshot Integrity
Hash a snapshot to detect changes.
import hashlib, json
def snapshot_hash(snap: dict) -> str:
data = json.dumps(snap, sort_keys=True).encode()
return hashlib.sha256(data).hexdigest()[:12]base64 — Safe Encoding
Encode binary data for transport or display.
import base64
encoded = base64.b64encode(b"\xde\xad\xbe\xef").decode() # '3q2+7w=='
decoded = base64.b64decode(encoded)dataclasses — Clean Data Models
Replace verbose __init__ boilerplate with a decorator.
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ProcessSnapshot:
pid: int
name: str
rss_mb: float
pct: float
timestamp: datetime = field(default_factory=datetime.now)
def is_heavy(self) -> bool:
return self.rss_mb > 512
snap = ProcessSnapshot(4821, "chrome", 1024.5, 73.5)
print(snap) # ProcessSnapshot(pid=4821, name='chrome', ...)enum — Named Constants
Replace magic strings/numbers with typed enums.
from enum import Enum
class AlertLevel(Enum):
OK = "ok"
WARNING = "warning"
CRITICAL = "critical"
def classify(pct: float) -> AlertLevel:
if pct >= 95: return AlertLevel.CRITICAL
if pct >= 80: return AlertLevel.WARNING
return AlertLevel.OKfunctools — Functional Utilities
from functools import lru_cache, partial, reduce
@lru_cache(maxsize=128)
def expensive_lookup(pid: int):
return get_process_name(pid) # cached after first call
# partial — pre-fill arguments
warn_check = partial(lambda limit, v: v > limit, 80)
warn_check(87) # True
# reduce — fold a list
from functools import reduce
total = reduce(lambda acc, s: acc + s.rss_mb, snapshots, 0.0)logging — Structured Logs
Replace print() with proper logging.
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler("ram_manager.log"),
logging.StreamHandler(),
]
)
log = logging.getLogger(__name__)
log.info("RAM Manager started")
log.warning("RAM usage at %.1f%%", 87.3)
log.error("Failed to read PID %d", 4821)typing — Type Hints
Document expected types and catch errors earlier with tools like mypy.
from typing import Optional, List, Dict
def get_top_processes(
snapshots: List[ProcessSnapshot],
n: int = 5,
threshold_mb: float = 256.0,
) -> List[ProcessSnapshot]:
heavy = [s for s in snapshots if s.rss_mb >= threshold_mb]
return sorted(heavy, key=lambda s: s.rss_mb, reverse=True)[:n]
def find_process(pid: int) -> Optional[ProcessSnapshot]:
...sqlite3 — Persistent Storage
Store RAM history in a local database.
import sqlite3
conn = sqlite3.connect("ram_history.db")
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS snapshots (
id INTEGER PRIMARY KEY,
ts TEXT,
pid INTEGER,
name TEXT,
rss_mb REAL
)
""")
cur.execute(
"INSERT INTO snapshots (ts, pid, name, rss_mb) VALUES (?, ?, ?, ?)",
(datetime.now().isoformat(), 4821, "chrome", 1024.5)
)
conn.commit()
conn.close()Standard Library Module Map
flowchart LR
A[RAM Manager] --> B[Data I/O]
A --> C[Modeling]
A --> D[System & Time]
A --> E[Observability]
B --> F[json]
B --> G[csv]
B --> H[pathlib]
B --> I[sqlite3]
C --> J[dataclasses]
C --> K[enum]
C --> L[typing]
D --> M[datetime]
D --> N[re]
D --> O[hashlib / base64]
E --> P[logging]
E --> Q[functools.lru_cache]Key Takeaways
jsonhandles config file read/write; useindent=2for human-readable files.reprovides full regex power for parsing system output and validating input.datetime+timedeltagive accurate, timezone-aware timestamps for every log entry.pathlib.Pathreplacesos.pathwith a cleaner, object-oriented API.@dataclasseliminates__init__boilerplate — the go-to for data-holding classes.enum.Enumreplaces magic strings with typed, documented constants.loggingreplacesprint()for production code — structured, leveled, file-aware.typinghints document intent and enable static analysis;sqlite3persists history across restarts.