← Home

Ch 11 — Real-World Tooling

Ch 11 — Real-World Tooling

The RAM manager is almost feature-complete. Now Aryan needs to connect it to the real world: actually reading system memory with psutil, fetching data over HTTP, handling network failures with retries, streaming large responses, persisting conversation history for an AI assistant integration, going async for non-blocking I/O, and generating file reports. This chapter bridges the gap between Python exercises and production software.

psutil — System Resource Access

psutil is the backbone of any system monitor written in Python.

import psutil

# Overall memory
mem = psutil.virtual_memory()
print(f"Total: {mem.total / (1024**3):.1f} GB")
print(f"Used:  {mem.used  / (1024**3):.1f} GB  ({mem.percent:.1f}%)")
print(f"Free:  {mem.available / (1024**3):.1f} GB")

# Per-process
for proc in psutil.process_iter(["pid", "name", "memory_info"]):
    try:
        rss_mb = proc.info["memory_info"].rss / (1024 ** 2)
        print(f"{proc.info['pid']:>6}  {proc.info['name']:<25}  {rss_mb:>8.1f} MB")
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        continue

HTTP Client — requests

Fetch external data (e.g., send alerts via webhook, pull version info).

import requests

def send_slack_alert(webhook_url: str, message: str) -> bool:
    payload = {"text": message}
    resp = requests.post(webhook_url, json=payload, timeout=5)
    resp.raise_for_status()   # raises HTTPError for 4xx/5xx
    return True

Environment Variables

Keep secrets and deployment-specific config out of source code.

import os

SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL", "")
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL_S", "1"))
LOG_LEVEL     = os.environ.get("LOG_LEVEL", "INFO")

if not SLACK_WEBHOOK:
    raise RuntimeError("SLACK_WEBHOOK_URL environment variable is not set")

Use python-dotenv in development to load a .env file automatically.

Retry / Backoff

Transient network failures are normal. Retry with exponential backoff.

import time, requests
from requests.exceptions import RequestException

def post_with_retry(url: str, payload: dict, max_retries: int = 3) -> dict:
    wait = 1.0
    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.post(url, json=payload, timeout=5)
            resp.raise_for_status()
            return resp.json()
        except RequestException as exc:
            if attempt == max_retries:
                raise
            print(f"Attempt {attempt} failed: {exc}. Retrying in {wait}s…")
            time.sleep(wait)
            wait *= 2   # exponential backoff

The tenacity library provides a decorator-based version of this pattern for production use.

Streaming Responses

When the API response is large (e.g., an LLM generating a long report), stream it instead of waiting.

import requests

def stream_report(url: str) -> str:
    lines = []
    with requests.get(url, stream=True, timeout=30) as resp:
        resp.raise_for_status()
        for chunk in resp.iter_lines():
            if chunk:
                line = chunk.decode()
                lines.append(line)
                print(line, flush=True)   # live output
    return "\n".join(lines)

Conversation History

Maintain a rolling history buffer for LLM-powered chat assistants embedded in the manager.

from collections import deque
from dataclasses import dataclass, field
from typing import List

@dataclass
class Message:
    role:    str   # "user" | "assistant" | "system"
    content: str

class ConversationHistory:
    def __init__(self, max_turns: int = 20):
        self._history: deque[Message] = deque(maxlen=max_turns * 2)

    def add(self, role: str, content: str) -> None:
        self._history.append(Message(role, content))

    def as_list(self) -> List[dict]:
        return [{"role": m.role, "content": m.content} for m in self._history]

asyncio + httpx — Non-Blocking I/O

Poll multiple endpoints concurrently without spinning up threads.

import asyncio
import httpx

async def fetch_ram_api(client: httpx.AsyncClient, host: str) -> dict:
    resp = await client.get(f"http://{host}/api/ram", timeout=5)
    return resp.json()

async def poll_all_hosts(hosts: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [fetch_ram_api(client, h) for h in hosts]
        return await asyncio.gather(*tasks)

# Entry point
results = asyncio.run(poll_all_hosts(["host1", "host2", "host3"]))

Typed API Models

Use dataclasses (or pydantic for validation) to define the shape of API responses.

from dataclasses import dataclass

@dataclass
class RamApiResponse:
    host:       str
    total_mb:   float
    used_mb:    float
    percent:    float
    timestamp:  str

def parse_response(raw: dict) -> RamApiResponse:
    return RamApiResponse(**raw)

File I/O Reports

Write a formatted text report to disk.

from pathlib import Path
from datetime import datetime

def write_report(snapshots: list, output_dir: Path) -> Path:
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = output_dir / f"ram_report_{ts}.txt"

    lines = [f"RAM Report — {ts}", "=" * 50]
    for snap in sorted(snapshots, key=lambda s: s.rss_mb, reverse=True):
        lines.append(f"{snap.name:<25} {snap.rss_mb:>8.1f} MB  {snap.pct:>6.1f}%")

    report_path.write_text("\n".join(lines))
    return report_path

Real-World Component Flow

flowchart TD
    A[psutil reads system RAM] --> B[ProcessSnapshot objects]
    B --> C{Alert threshold crossed?}
    C -->|Yes| D[post_with_retry to Slack webhook]
    C -->|No| E[Append to history buffer]
    D --> F{Retry needed?}
    F -->|Yes| G[Exponential backoff]
    G --> D
    F -->|No| H[Log success]
    E --> I[write_report to disk]
    I --> J[CSV / text file saved]

Async vs Sync Decision

flowchart LR
    A[Task] --> B{I/O bound?}
    B -->|Yes — network / disk| C{Many concurrent tasks?}
    B -->|No — CPU bound| D[Multiprocessing or sync]
    C -->|Yes| E[asyncio + httpx]
    C -->|No| F[requests — sync is fine]
    E --> G[asyncio.gather for parallel fetches]
    F --> H[Simple sequential calls]

Key Takeaways

  • psutil provides cross-platform access to system memory data with minimal setup.
  • requests handles HTTP; always set a timeout and call raise_for_status().
  • Environment variables keep secrets out of code; os.environ.get(key, default) is safe for optional values.
  • Retry with exponential backoff makes the manager resilient to transient network failures.
  • Streaming (iter_lines()) handles large or incremental responses without buffering the full body.
  • A deque(maxlen=N) is the cleanest rolling conversation history buffer.
  • asyncio + httpx enable concurrent polling of multiple hosts without threading complexity.
  • Typed API models (@dataclass) enforce response shape and make downstream code self-documenting.