150. Stdlib Toolkit (App V1.0)

The app works. Time to make it production-quality.

🤔 Socratic question: Why do companies pay engineers just to “add logging”? Why is print() considered unprofessional? Because when something breaks at 3am on a production server with 10 million users, you need to know exactly what happened — and print() statements don’t have timestamps, levels, or go to log files.

🤯 Scale check: Google processes more than 8.5 billion searches per day. Every single one generates a log entry. Google has engineers whose entire job is just making logging fast enough. You’re learning the same concepts they use — just at a much smaller scale.

Three stdlib tools that separate amateur code from professional code:

1. logging instead of print() 📝

import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(name)s %(levelname)s %(message)s')
logger = logging.getLogger('ram_manager')

logger.debug('Taking snapshot')           # only visible at DEBUG level
logger.info('RAM: %.1f%%', pct)           # informational
logger.warning('RAM high: %.1f%%', pct)   # visible by default
logger.error('LLM call failed: %s', e)   # always visible

2. Counter for frequency tracking 📊

from collections import Counter

offender_counts: Counter = Counter()

def record_offenders(processes, threshold_mb=100):
    names = [p['name'] for p in processes if p['rss_mb'] > threshold_mb]
    offender_counts.update(names)          # increment all at once
    top = offender_counts.most_common(3)   # [(name, count), ...]
    logger.info('Top offenders: %s', top)

3. @wraps for decorator identity 🎭

from functools import wraps

def retry(fn):
    @wraps(fn)                    # preserves fn.__name__ and __doc__
    def wrapper(*args, **kwargs):
        ...
    return wrapper

Without @wraps, every decorated function shows up as 'wrapper' in logs and tracebacks. Debugging becomes guesswork.

💡 Real-world: Every production Python service at companies like Stripe, Airbnb, and Uber uses logging (not print), Counter for metrics, and decorators for retry logic. This chapter is your graduation from “it works on my machine” to “it works in production.”

💡 Fun fact: Counter.most_common() uses a heap sort internally (heapq.nlargest) — O(n log k) instead of O(n log n) for a full sort. When you’re counting millions of log events and only need the top 10, this matters. The same algorithm powers the “trending topics” feature in every major social platform.

⚠️ Watch out: When you put @retry on ask_llm, the retry decorator needs @wraps(fn) or ask_llm.__name__ will become 'wrapper'. Then when the retry logs “wrapper attempt 2 failed”, you have no idea which function it is. In a codebase with many decorated functions, this makes debugging near-impossible.

🤔 Think about it: The solution uses Counter.update(names) where names is a list comprehension. What’s the difference between Counter.update([...]) and Counter([...]) as a constructor? Why does offender_counts need to be a module-level variable rather than local to record_offenders — and what are the tradeoffs?

🏁 Aryan polishes the RAM manager to production quality. He adds logging, tracks which processes are repeat offenders with Counter, and wraps the LLM call in a retry decorator. His print() calls need to become logging calls, his retry decorator is missing @wraps, and he uses a plain dict for counting instead of Counter.

Learning objectives

Use logging.getLogger() instead of print() for operational messages
Use Counter for frequency tracking (update(), most_common())
Apply @wraps(fn) inside decorators to preserve identity
Combine all v0.x features into a single coherent app

Key concepts

logging.getLogger(name) — named logger
logger.debug/info/warning/error() — leveled messages
Counter.update(iterable) — increment multiple counts
Counter.most_common(n) — top-N by count
@wraps(fn) — preserve name and doc

Try it

import time
import psutil
import os
import requests
import tomllib
import click
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
from rich.console import Console
from rich.table import Table
from rich import box
<p>load_dotenv()
console = Console()
CONFIG_PATH = Path.home() / ‘.ram_manager’ / ‘config.toml’
DEFAULT_CONFIG = {‘threshold’: 80, ‘top_n’: 10, ‘model’: ‘claude-haiku-4-5-20251001’}</p>
<h1 id="bulb-hint-010-xp-replace-print-with-logger-debug-add-logging-getlogger-ram-manager-at-module-level">💡 : hint : -010 xp Replace print() with logger.debug() — add logging.getLogger(‘ram_manager’) at module level</h1>
<p>def take_snapshot(n: int = 10) -> dict:
print(‘Taking RAM snapshot…’)
mem = psutil.virtual_memory()
processes = []
for proc in psutil.process_iter([‘pid’, ‘name’, ‘memory_info’]):
try:
rss_mb = proc.info[‘memory_info’].rss / 1e6
processes.append({‘name’: proc.info[‘name’], ‘pid’: proc.info[‘pid’], ‘rss_mb’: rss_mb})
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
return {
‘percent’: mem.percent,
‘used_gb’: mem.used / 1e9,
‘total_gb’: mem.total / 1e9,
‘processes’: sorted(processes, key=lambda p: p[‘rss_mb’], reverse=True)[:n],
‘timestamp’: datetime.now().isoformat(),
}</p>
<h1 id="bulb-hint-010-xp-use-counter-as-the-type-and-offender-counts-update-names-manual-dict-counting-is-verbose-and-error-prone">💡 : hint : -010 xp Use Counter as the type and offender_counts.update(names) — manual dict counting is verbose and error-prone</h1>
<p>offender_counts = {}
def record_offenders(processes: list[dict], threshold_mb: float = 100) -> None:
for p in processes:
if p[‘rss_mb’] > threshold_mb:
if p[‘name’] not in offender_counts:
offender_counts[p[‘name’]] = 0
offender_counts[p[‘name’]] += 1</p>
<h1 id="bulb-hint-010-xp-add-wraps-fn-inside-the-wrapper-without-it-ask-llm-name-becomes-wrapper-and-tracebacks-are-unreadable">💡 : hint : -010 xp Add @wraps(fn) inside the wrapper — without it ask_llm.<strong>name</strong> becomes ‘wrapper’ and tracebacks are unreadable</h1>
<p>def retry(fn):
def wrapper(*args, **kwargs):
for attempt in range(3):
try:
return fn(*args, **kwargs)
except Exception as e:
if attempt == 2:
raise
print(f’Retry {attempt + 1}: {e}’)
time.sleep(2 ** attempt)
return wrapper</p>
<p>@retry
def ask_llm(snapshot: dict, model: str) -> str:
api_key = os.environ.get(‘ANTHROPIC_API_KEY’)
if not api_key:
raise EnvironmentError(‘ANTHROPIC_API_KEY not set’)
prompt = f"RAM at {snapshot[‘percent’]:.1f}%. Advise briefly."
response = requests.post(
‘https://api.anthropic.com/v1/messages’,
json={‘model’: model, ‘max_tokens’: 200, ‘messages’: [{‘role’: ‘user’, ‘content’: prompt}]},
headers={‘x-api-key’: api_key, ‘anthropic-version’: ‘2023-06-01’},
timeout=30,
)
response.raise_for_status()
return response.json()[‘content’][0][‘text’]</p>
<p>def load_config(path: Path = CONFIG_PATH) -> dict:
config = DEFAULT_CONFIG.copy()
if path.exists():
with open(path, ‘rb’) as f:
for k, v in tomllib.load(f).items():
if k in config and v is not None:
config[k] = v
return config</p>
<p>@click.group()
def cli():
‘’‘RAM Manager v1.0’‘’</p>
<p>@cli.command()
@click.option(‘–threshold’, type=int, default=None)
@click.option(‘–top-n’, type=int, default=None)
@click.option(‘–verbose’, is_flag=True)
def monitor(threshold, top_n, verbose):
config = load_config()
t = threshold if threshold is not None else config[‘threshold’]
n = top_n if top_n is not None else config[‘top_n’]
snapshot = take_snapshot(n=n)
record_offenders(snapshot[‘processes’])
pct = snapshot[‘percent’]
color = ‘red’ if pct > t else ‘green’
console.print(f’[{color}]RAM: {pct:.1f}%[/]’)</p>
<p>if <strong>name</strong> == ‘<strong>main</strong>’:
cli()

Concept detail

App v1.0 — Production polish

The final RAM manager has:

Logging instead of print() — respects log levels
Counter tracking repeat RAM offenders across checks
Retry decorator with exponential backoff for the LLM call
Click subcommands with config file + CLI override
Rich table showing process history (“Hits” column)

$ python ram_manager.py monitor --ask-ai
2024-03-15 14:30:22 ram_manager INFO Top RAM offenders: [('chrome', 3), ...]
╭─────────────────────────────────────────────╮
│                RAM: 87.3%                   │
├─────────────────┬────────┬────────┬─────────┤
│ Process         │    PID │ RSS MB │    Hits │
├─────────────────┼────────┼────────┼─────────┤
│ chrome          │    812 │    800 │       3 │
│ python          │    421 │    150 │       1 │
╰─────────────────┴────────┴────────┴─────────╯

AI advice:
Chrome is a repeat offender (3 hits). Consider closing unused tabs...

Solution

import time
import logging
import psutil
import os
import requests
import tomllib
import click
from collections import Counter
from functools import wraps
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
from rich.console import Console
from rich.table import Table
from rich import box

load_dotenv()
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(name)s %(levelname)s %(message)s',
)
logger = logging.getLogger('ram_manager')
console = Console()

CONFIG_PATH = Path.home() / '.ram_manager' / 'config.toml'
DEFAULT_CONFIG = {'threshold': 80, 'top_n': 10, 'model': 'claude-haiku-4-5-20251001'}

# Track repeat RAM offenders across runs in this session
offender_counts: Counter = Counter()

def take_snapshot(n: int = 10) -> dict:
    logger.debug('Taking RAM snapshot')
    mem = psutil.virtual_memory()
    processes = []
    for proc in psutil.process_iter(['pid', 'name', 'memory_info']):
        try:
            rss_mb = proc.info['memory_info'].rss / 1e6
            processes.append({'name': proc.info['name'], 'pid': proc.info['pid'], 'rss_mb': rss_mb})
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            pass
    return {
        'percent': mem.percent,
        'used_gb': mem.used / 1e9,
        'total_gb': mem.total / 1e9,
        'processes': sorted(processes, key=lambda p: p['rss_mb'], reverse=True)[:n],
        'timestamp': datetime.now().isoformat(),
    }

def record_offenders(processes: list[dict], threshold_mb: float = 100) -> None:
    names = [p['name'] for p in processes if p['rss_mb'] > threshold_mb]
    offender_counts.update(names)   # Counter.update() increments counts
    if offender_counts:
        top = offender_counts.most_common(3)
        logger.info('Top RAM offenders this session: %s', top)

def retry(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        for attempt in range(3):
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                if attempt == 2:
                    raise
                wait = 2 ** attempt
                logger.warning('Attempt %d failed: %s. Retrying in %ds', attempt + 1, e, wait)
                time.sleep(wait)
    return wrapper

@retry
def ask_llm(snapshot: dict, model: str) -> str:
    api_key = os.environ.get('ANTHROPIC_API_KEY')
    if not api_key:
        raise EnvironmentError('ANTHROPIC_API_KEY not set')
    prompt = (
        f"RAM at {snapshot['percent']:.1f}% "
        f"({snapshot['used_gb']:.1f}/{snapshot['total_gb']:.1f} GB). "
        f"Top process: {snapshot['processes'][0]['name'] if snapshot['processes'] else 'none'}. "
        f"Give brief advice."
    )
    response = requests.post(
        'https://api.anthropic.com/v1/messages',
        json={'model': model, 'max_tokens': 200, 'messages': [{'role': 'user', 'content': prompt}]},
        headers={'x-api-key': api_key, 'anthropic-version': '2023-06-01'},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()['content'][0]['text']

def load_config(path: Path = CONFIG_PATH) -> dict:
    config = DEFAULT_CONFIG.copy()
    if path.exists():
        with open(path, 'rb') as f:
            for k, v in tomllib.load(f).items():
                if k in config and v is not None:
                    config[k] = v
    return config

def make_process_table(snapshot: dict) -> Table:
    table = Table(box=box.ROUNDED, title=f"RAM: {snapshot['percent']:.1f}%")
    table.add_column('Process', style='cyan')
    table.add_column('PID', justify='right')
    table.add_column('RSS MB', justify='right')
    table.add_column('Hits', justify='right', style='yellow')
    for p in snapshot['processes']:
        rss = p['rss_mb']
        color = 'red' if rss > 500 else ('yellow' if rss > 100 else 'green')
        hits = str(offender_counts.get(p['name'], 0)) or ''
        table.add_row(p['name'], str(p['pid']), f'[{color}]{rss:.0f}[/]', hits)
    return table

@click.group()
def cli():
    '''RAM Manager v1.0 — monitor memory, get AI advice.'''

@cli.command()
@click.option('--threshold', type=int, default=None)
@click.option('--top-n', type=int, default=None)
@click.option('--ask-ai', is_flag=True, help='Get LLM advice')
def monitor(threshold, top_n, ask_ai):
    '''Monitor RAM usage.'''
    config = load_config()
    t = threshold if threshold is not None else config['threshold']
    n = top_n if top_n is not None else config['top_n']
    snapshot = take_snapshot(n=n)
    record_offenders(snapshot['processes'])
    if snapshot['percent'] > t:
        logger.warning('RAM above threshold: %.1f%% > %d%%', snapshot['percent'], t)
    console.print(make_process_table(snapshot))
    if ask_ai:
        try:
            advice = ask_llm(snapshot, config['model'])
            console.print(f'\n[bold]AI advice:[/]\n{advice}')
        except Exception as e:
            logger.error('LLM call failed: %s', e)

if __name__ == '__main__':
    cli()

Tests

import logging
from collections import Counter

def test_uses_logger_not_print():
    import inspect
    src = inspect.getsource(take_snapshot)
    assert 'logger.' in src, 'Use logger.debug/info/warning() not print()'
    assert 'print(' not in src

def test_record_offenders_uses_counter():
    import inspect
    src = inspect.getsource(record_offenders)
    assert 'Counter' in src or 'offender_counts' in src

def test_record_offenders_increments():
    global offender_counts
    offender_counts.clear()
    procs = [{'name': 'chrome', 'rss_mb': 200}, {'name': 'python', 'rss_mb': 50}]
    record_offenders(procs, threshold_mb=100)
    assert offender_counts['chrome'] == 1
    record_offenders(procs, threshold_mb=100)
    assert offender_counts['chrome'] == 2

def test_retry_preserves_name():
    @retry
    def my_func():
        '''My docstring.'''
        pass
    assert my_func.__name__ == 'my_func', 'Use @wraps(fn) in retry decorator'

def test_retry_retries_on_failure():
    calls = []
    @retry
    def flaky():
        calls.append(1)
        if len(calls) < 3:
            raise ValueError('not ready')
        return 'done'
    result = flaky()
    assert result == 'done'
    assert len(calls) == 3

def test_ask_llm_is_retried():
    import inspect
    src = inspect.getsource(ask_llm)
    # The @retry decorator should be applied
    assert True   # structure test — covered by decorator being present