DEV Community: Developer Service

Your RAG Pipeline Is Lying to You

Developer Service — Thu, 14 May 2026 13:28:41 +0000

You built a RAG pipeline. You tested it manually with a few queries, got reasonable answers, and shipped it. Your stakeholders are happy. The chatbot returns answers that sound authoritative.

But here is the uncomfortable question: do you actually know if it's right?

Not "does it return something", but does it return the correct thing, consistently, across the full distribution of queries your users will actually ask?

If you can't answer that with a number, you don't have a production system. You have a demo with a deployment URL.

This article is for senior engineers who build RAG systems and want to know if they actually work. We will walk through a concrete RAG example - a pipeline over a corporate annual report - and build the testing layer that most teams skip entirely. The code is real and runnable. The failures are not hypothetical.

All code in this article is available in the companion GitHub repository: github.com/nunombispo/rag-pipeline-article. The repository includes a sample golden dataset, the full pytest suite, and a minimal monitoring wrapper.

Why RAG Is Harder Than It Looks

RAG sounds simple on paper: retrieve relevant context, pass it to the model, get a grounded answer. The reality is that there are three independent failure layers, any one of which can silently corrupt your results.

Layer 1 - Retrieval failure: The vector search returns chunks that are topically adjacent to the query but don't contain the actual answer. Similarity is not the same as relevance. A question about Q3 revenue may surface chunks about Q3 strategy, not the income statement.

Layer 2 - Context failure: The right information is in the document, but it was split across a chunk boundary during preprocessing. The number is on page 47, the label is on page 46, and your chunker put them in separate vectors. Neither chunk alone answers the question correctly.

Layer 3 - Generation failure: The model receives good context but still produces a wrong answer - either by hallucinating when the context is ambiguous, by confusing units or time periods, or by confidently synthesizing an answer from two chunks that should not be combined.

Each layer fails independently and silently. There is no exception thrown. The user gets an answer. The answer looks plausible.

Query ──► [Embedding] ──► [Vector Search] ──► [Context Assembly] ──► [LLM] ──► Answer
              │                  │                     │               │
           Layer 0            Layer 1               Layer 2         Layer 3
         (usually OK)      (common fail)         (common fail)   (occasional fail)

Layer 0 is the embedding model itself: given a query string, does it produce a vector? This step almost never fails in practice — hosted embedding APIs are reliable, and vector dimension mismatches surface immediately at index time. Most teams verify it once and move on.

The layers that actually kill production systems are 1 and 2. They are typically invisible until a user complains - and most users don't complain. They just stop trusting the system.

RAG failures are not crashes. They are confidence-eroding wrong answers delivered at scale. The risk is not a system outage - it is the gradual destruction of user trust in your AI investment.

The Concrete Example: RAG Over an Annual Report

The easiest way to see all three layers fail is on a real system, with a real document, running real queries. Annual reports are the canonical enterprise RAG use case. Dense prose, financial tables, footnotes, forward-looking statements, and regulatory boilerplate - all in one PDF. They are also a perfect failure showcase because the data is precise, verifiable, and publicly available.

We will use Microsoft's 2025 Annual Report (publicly available). It's 83 pages of shareholder letters, financial tables, segment breakdowns, accounting notes, and legal disclosures - exactly the kind of document that gets thrown at enterprise RAG pipelines and produces interesting failures. The principles apply to any dense financial document.

Setup

pip install anthropic pydantic>=2 chromadb pypdf python-dotenv rich pytest

Step 1: Ingest the PDF

# ingest.py
import hashlib
from pypdf import PdfReader
import chromadb

CHUNK_SIZE = 800        # characters
CHUNK_OVERLAP = 100


def load_pdf(path: str) -> list[dict]:
    """Extract text from PDF, page by page."""
    reader = PdfReader(path)
    pages = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text() or ""
        pages.append({"page": i + 1, "text": text.strip()})
    return pages


def chunk_text(pages: list[dict]) -> list[dict]:
    """Naive fixed-size chunking with overlap."""
    chunks = []
    for page in pages:
        text = page["text"]
        start = 0
        while start < len(text):
            end = start + CHUNK_SIZE
            chunk = text[start:end]
            chunk_id = hashlib.md5(chunk.encode()).hexdigest()
            chunks.append({
                "id": chunk_id,
                "text": chunk,
                "page": page["page"],
                "start": start,
            })
            start += CHUNK_SIZE - CHUNK_OVERLAP
    return chunks


def build_collection(pdf_path: str, collection_name: str = "annual_report"):
    client = chromadb.Client()
    collection = client.get_or_create_collection(collection_name)

    pages = load_pdf(pdf_path)
    chunks = chunk_text(pages)

    collection.add(
        ids=[c["id"] for c in chunks],
        documents=[c["text"] for c in chunks],
        metadatas=[{"page": c["page"]} for c in chunks],
    )

    print(f"Indexed {len(chunks)} chunks from {len(pages)} pages.")
    return collection

Step 2: Query the Pipeline

# rag.py
import anthropic
import chromadb
import os
from dotenv import load_dotenv

MODEL = "claude-sonnet-4-6"
TOP_K = 3

load_dotenv()
API_KEY = os.getenv("ANTHROPIC_API_KEY")
client = anthropic.Anthropic(api_key=API_KEY)


def retrieve(collection, query: str, top_k: int = TOP_K) -> list[dict]:
    results = collection.query(query_texts=[query], n_results=top_k)
    chunks = []
    for doc, meta, distance in zip(
        results["documents"][0],
        results["metadatas"][0],
        results["distances"][0],
    ):
        chunks.append({"text": doc, "page": meta["page"], "score": 1 - distance})
    return chunks


def answer(query: str, chunks: list[dict]) -> str:
    context = "\n\n---\n\n".join(
        f"[Page {c['page']}]\n{c['text']}" for c in chunks
    )
    system = (
        "You are a financial analyst assistant. Answer questions based strictly "
        "on the provided context. If the answer is not present in the context, "
        "say 'I could not find this in the provided document.' "
        "Do not speculate or use external knowledge."
    )
    response = client.messages.create(
        model=MODEL,
        max_tokens=512,
        thinking={"type": "disabled"},  # disable extended thinking; not needed here
        system=system,
        messages=[
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}",
            }
        ],
    )
    parts = [b.text for b in response.content if getattr(b, "type", None) == "text"]
    return "\n".join(parts).strip() if parts else ""


def rag_query(collection, query: str) -> dict:
    chunks = retrieve(collection, query)
    response = answer(query, chunks)
    return {
        "query": query,
        "chunks": chunks,
        "answer": response,
    }

Step 3: Run It

# main.py
from ingest import build_collection
from rag import rag_query

collection = build_collection("2025_AnnualReport.pdf")  # optional: collection_name="my_docs"
result = rag_query(collection, "What were the main risks discussed?")

print(result["answer"])
# result also has "chunks" (text, page, score) and "query"

This is a typical minimal RAG implementation. Straightforward, readable - and full of silent failure modes. Let's expose them.

What Happens Without Evaluation

Before building the test suite, let's make the failures concrete by running the pipeline against four representative queries. When retrieval and generation align, the pipeline delivers a clean, grounded answer with no intervention needed - Query 1 shows this. But "correct" and "useful" are not the same thing, and the remaining queries show how quietly the gap between them can widen.

Query 1: "What was Microsoft's total revenue in fiscal year 2025?"

Based on the provided context, Microsoft's total revenue in fiscal year 2025 was
**$281,724 million ($281.7 billion)**, representing a **15% increase** compared
to fiscal year 2024's revenue of $245,122 million.

The answer is correct. But the same figure also appears narratively in the shareholder letter on page 2 - "revenue was $281.7 billion, up 15 percent" - without the structured financial context an analyst would expect. If the retriever ranks the letter chunk higher than the financial tables on page 25, the answer is technically right but stripped of the gross margin, operating income, and EPS data that make it meaningful. It passes a keyword check. It fails a usefulness check.

Query 2: "How many employees does Microsoft have?"

I could not find this in the provided document. The provided context does not
contain any information about Microsoft's employee headcount.

The answer - 228,000 full-time employees as of June 30, 2025 - is in the Human Capital Resources section on page 16. But the section spans pages 15–17, and naive fixed-size chunking splits it mid-paragraph. Depending on where the chunk boundary falls, the retriever may rank pages 15 or 17 higher than page 16, returning context about compensation programs or operations centers rather than the headcount figure. The model refuses. The refusal looks correct. The underlying cause is a retrieval failure.

A less disciplined model - or the same model without an explicit "do not speculate" instruction in the system prompt - may hallucinate a figure by latching onto other numeric patterns in the retrieved chunks. The correct behavior is not guaranteed; it is a product of your system prompt design, and without a test asserting it, a prompt change can silently remove it.

Query 3: "What percentage of revenue came from cloud services?"

Based on the provided context, I can identify the following cloud-related revenue
items for fiscal year 2025:

| Product/Service                                    | Revenue (in millions) |
|----------------------------------------------------|-----------------------|
| Server products and cloud services                 | $98,435               |
| Microsoft 365 Commercial products and cloud services| $87,767              |
| Microsoft 365 Consumer products and cloud services | $7,404                |

**However, I cannot provide a precise "cloud-only" percentage** for the following
reasons:

1. Several line items **bundle** cloud services with non-cloud products (e.g.,
   "Server products **and** cloud services"), making it impossible to isolate
   pure cloud revenue from the data provided.
2. Total revenue is stated as **$281,724 million** for FY2025.

If the bundled categories above are used as a rough proxy, those three lines sum
to approximately **$193,606 million**, representing roughly **~69%** of total
revenue — but this would **overstate** true cloud-only revenue since non-cloud
products are included in those figures.

The retriever surfaces chunks that bundle cloud and non-cloud revenue under the same line items. The model correctly flags the ambiguity, but still calculates a ~69% figure using the bundled totals as a proxy. The answer looks rigorous: it has a table, a caveat, and a percentage. It is also wrong in a way that is easy to miss. A reader who skims the headline number walks away with a figure that overstates cloud revenue by an unknown margin. Transparency about uncertainty is not the same as a correct answer.

Query 4: "Did Microsoft mention any regulatory risks related to AI?"

Based on the provided context, **no specific mention of regulatory risks related
to AI** is made in the excerpts provided. The pages included focus primarily on:

- Microsoft's mission and AI platform strategy (Page 22)
- The company's positioning in the AI platform shift (Page 2)
- The Microsoft Elevate initiative and AI skills investments (Page 4)

**I could not find this in the provided document.** To get a comprehensive view
of Microsoft's AI-related regulatory risks, you would likely need to refer to the
**Risk Factors section** of their Annual Report on Form 10-K, which is not
included in the provided context.

The model correctly declines - and it is correct to do so. The actual regulatory risk language lives in the 10-K filing, a separate document not in the pipeline. The refusal is technically right. But the user doesn't know that. From their perspective, they asked a reasonable question and got a dead end. The real failure is architectural: the pipeline has no mechanism to tell the user why it can't answer - whether the information doesn't exist, exists in the document but wasn't retrieved, or exists in a document that was never ingested. All three scenarios return the same response. Without instrumentation, you cannot distinguish between them.

Without a ground truth dataset and a structured test harness, you have no way to know which of these answers are right - or whether a change to your chunking strategy or system prompt just made things worse.

The Testing Layer

Good RAG evaluation has three stages. Each catches a different class of failure.

Stage 1: Chunk Quality

Before any query is run, validate that your chunking strategy produces coherent, complete units of information.

# tests/test_chunks.py
import pytest
from ingest import load_pdf, chunk_text

PDF_PATH = "2025_AnnualReport.pdf"

@pytest.fixture(scope="module")
def chunks():
    pages = load_pdf(PDF_PATH)
    return chunk_text(pages)


def test_no_empty_chunks(chunks):
    empty = [c for c in chunks if not c["text"].strip()]
    assert len(empty) == 0, f"{len(empty)} empty chunks found"


def test_chunk_size_bounds(chunks):
    oversized = [c for c in chunks if len(c["text"]) > 1000]
    assert len(oversized) == 0, f"{len(oversized)} chunks exceed 1000 chars"


def test_no_orphaned_numbers(chunks):
    """Flag chunks that are pure numeric tables with no prose context.
    These are extraction artifacts from PDF tables and answer nothing reliably."""
    import re
    suspicious = []
    for c in chunks:
        words = re.findall(r"[a-zA-Z]{3,}", c["text"])
        if len(words) < 5 and len(c["text"]) > 100:
            suspicious.append(c)
    assert len(suspicious) < 10, (
        f"{len(suspicious)} chunks look like structureless table extracts. "
        "Consider a table-aware PDF parser."
    )


def test_chunk_overlap_preserves_context(chunks):
    """Verify that consecutive chunks share some overlapping text."""
    # Sample the first 50 chunk pairs
    misses = 0
    for i in range(min(50, len(chunks) - 1)):
        a = chunks[i]["text"]
        b = chunks[i + 1]["text"]
        # Overlap window: last N chars of a should appear in start of b
        overlap_window = a[-100:]
        if overlap_window not in b:
            misses += 1
    # Allow some misses (page boundaries)
    assert misses < 15, f"Too many chunk pairs with no detectable overlap ({misses})"

These tests catch configuration mistakes before they touch the vector store.

Stage 2: Retrieval Evaluation

This is the most important stage and the most commonly skipped. Build a golden dataset: a small set of (query, expected page, expected content) pairs where you know where the answer lives.

# tests/golden_dataset.py
GOLDEN_QA = [
    {
        "query": "What was Microsoft's total revenue in fiscal year 2025?",
        "expected_page": 25,        # Summary Results of Operations table
        "expected_answer_contains": ["281", "billion"],
    },
    {
        "query": "How many employees does Microsoft have?",
        "expected_page": 16,        # Human Capital Resources section
        "expected_answer_contains": ["228,000", "employees"],
    },
    {
        "query": "What was the net income for fiscal year 2025?",
        "expected_page": 37,        # Income Statements
        "expected_answer_contains": ["101", "billion"],
    },
    {
        "query": "What was Microsoft Cloud revenue in fiscal year 2025?",
        "expected_page": 22,        # Overview highlights section
        "expected_answer_contains": ["168.9", "billion"],
    },
    {
        "query": "Did Microsoft pay dividends in fiscal year 2025?",
        "expected_page": 8,         # Dividends table
        "expected_answer_contains": ["3.32", "dividend"],
    },
]

And the test for the retrieval:

# tests/test_retrieval.py
import pytest
from ingest import build_collection
from rag import retrieve
from tests.golden_dataset import GOLDEN_QA

PDF_PATH = "2025_AnnualReport.pdf"
# Chroma similarity can rank adjacent prose above the exact table row; keep k generous.
TOP_K = 12


@pytest.fixture(scope="module")
def collection():
    return build_collection(PDF_PATH)


def test_retrieval_hit_rate(collection):
    """Aggregate metric: what fraction of golden queries retrieve the right page?"""
    hits = 0
    for item in GOLDEN_QA:
        chunks = retrieve(collection, item["query"], top_k=TOP_K)
        if item["expected_page"] in [c["page"] for c in chunks]:
            hits += 1
    hit_rate = hits / len(GOLDEN_QA)
    print(f"\nRetrieval hit rate: {hit_rate:.0%} ({hits}/{len(GOLDEN_QA)})")
    assert hit_rate >= 0.80, f"Retrieval hit rate {hit_rate:.0%} is below the 80% threshold"

The 80% threshold is the key line. It gives you a regression gate: if a change to your chunking strategy, embedding model, or vector store configuration degrades retrieval below that threshold, CI fails. You know before it reaches users.

Stage 3: End-to-End Answer Quality

LLM-as-judge is the pragmatic approach here. You don't need a separate expensive evaluation model - the same Claude API call that powers your pipeline can evaluate its own answers against expected criteria.

# tests/test_answers.py
from typing import Literal

import anthropic
import pytest
from pydantic import BaseModel, Field
from ingest import build_collection
from rag import rag_query
from tests.golden_dataset import GOLDEN_QA

PDF_PATH = "2025_AnnualReport.pdf"
client = anthropic.Anthropic()
JUDGE_MODEL = "claude-sonnet-4-6"


class JudgeEvaluation(BaseModel):
    """Schema for Claude JSON / structured output (output_format)."""

    contains_expected_info: bool = Field(
        description="Whether the answer includes the expected factual content."
    )
    is_hallucinated: bool = Field(
        description="Whether the answer invents facts not grounded in the question/context."
    )
    confidence: Literal["high", "medium", "low"] = Field(
        description="How confident the assessment is."
    )
    reason: str = Field(description="Brief explanation of the assessment.")


def llm_judge(query: str, answer: str, expected_keywords: list[str]) -> dict:
    """Use Claude structured outputs to evaluate answer quality."""
    prompt = f"""You are evaluating a RAG system answer for factual correctness.

Question: {query}
Answer: {answer}
Expected to contain information about: {", ".join(expected_keywords)}

Fill the structured fields: whether the answer contains the expected information,
whether it appears hallucinated, your confidence level, and a short reason."""

    response = client.messages.parse(
        model=JUDGE_MODEL,
        max_tokens=512,
        temperature=0,
        thinking={"type": "disabled"},  # disable extended thinking; not needed here
        messages=[{"role": "user", "content": prompt}],
        output_format=JudgeEvaluation,
    )
    parsed = response.parsed_output
    if parsed is None:
        raise RuntimeError(
            "Structured judge output missing; check model support for output_format "
            "and response content."
        )
    return parsed.model_dump()


@pytest.fixture(scope="module")
def collection():
    return build_collection(PDF_PATH)


@pytest.mark.parametrize("item", GOLDEN_QA, ids=[q["query"][:40] for q in GOLDEN_QA])
def test_answer_quality(collection, item):
    result = rag_query(collection, item["query"])
    evaluation = llm_judge(
        item["query"],
        result["answer"],
        item["expected_answer_contains"],
    )
    assert not evaluation["is_hallucinated"], (
        f"Hallucination detected.\nQuery: {item['query']}\n"
        f"Answer: {result['answer']}\nReason: {evaluation['reason']}"
    )
    assert evaluation["contains_expected_info"], (
        f"Answer missing expected content.\nQuery: {item['query']}\n"
        f"Answer: {result['answer']}\nExpected: {item['expected_answer_contains']}\n"
        f"Reason: {evaluation['reason']}"
    )

Run the full suite with:

pytest tests/ -v --tb=short

A representative run looks like this:

collected 10 items                                                                            

tests/test_answers.py::test_answer_quality[What was Microsoft's total revenue in fi] PASSED       [ 10%]
tests/test_answers.py::test_answer_quality[How many employees does Microsoft have?] FAILED        [ 20%]
tests/test_answers.py::test_answer_quality[What was the net income for fiscal year ] PASSED       [ 30%]
tests/test_answers.py::test_answer_quality[What was Microsoft Cloud revenue in fisc] PASSED       [ 40%]
tests/test_answers.py::test_answer_quality[Did Microsoft pay dividends in fiscal ye] FAILED       [ 50%]
tests/test_chunks.py::test_no_empty_chunks PASSED                                                 [ 60%]
tests/test_chunks.py::test_chunk_size_bounds PASSED                                               [ 70%]
tests/test_chunks.py::test_no_orphaned_numbers PASSED                                             [ 80%]
tests/test_chunks.py::test_chunk_overlap_preserves_context PASSED                                 [ 90%]
tests/test_retrieval.py::test_retrieval_hit_rate PASSED

That failure is not a bug in the test. That is the test working. The headcount figure - 228,000 employees as of June 30, 2025 - sits in the middle of the Human Capital Resources section on page 16. The chunker split the paragraph across the page 15–16 boundary, and the retriever ranked adjacent chunks higher. You know this now, before a user discovers it in production.

The Monitoring Layer

Testing covers the queries you anticipated. Monitoring covers the queries you didn't.

In production, instrument every request with structured logging:

# monitoring.py
import time
import json
import logging
from rag import retrieve, answer

logger = logging.getLogger("rag.monitor")


def monitored_rag_query(collection, query: str) -> dict:
    t0 = time.perf_counter()
    chunks = retrieve(collection, query)
    retrieval_time = time.perf_counter() - t0

    # Flag low-confidence retrievals before they reach the model
    top_score = chunks[0]["score"] if chunks else 0.0
    low_confidence = top_score < 0.75

    t1 = time.perf_counter()
    response = answer(query, chunks)
    generation_time = time.perf_counter() - t1

    event = {
        "query": query,
        "retrieval_time_ms": round(retrieval_time * 1000, 1),
        "generation_time_ms": round(generation_time * 1000, 1),
        "top_retrieval_score": round(top_score, 4),
        "low_confidence_retrieval": low_confidence,
        "num_chunks_retrieved": len(chunks),
        "retrieved_pages": [c["page"] for c in chunks],
    }

    logger.info(json.dumps(event))

    if low_confidence:
        logger.warning(f"Low-confidence retrieval for query: {query!r} (score={top_score:.3f})")

    return {"query": query, "chunks": chunks, "answer": response, "meta": event}

Here is what the monitoring output looks like for the headcount query we already know fails:

INFO:httpx:HTTP Request: POST https://api.anthropic.com/v1/messages "HTTP/1.1 200 OK"
INFO:rag.monitor:{"query": "How many employees does Microsoft have?", "retrieval_time_ms": 244.1, "generation_time_ms": 1893.5, "top_retrieval_score": 0.1758, "low_confidence_retrieval": true, "num_chunks_retrieved": 3, "retrieved_pages": [81, 22, 75]}
WARNING:rag.monitor:Low-confidence retrieval for query: 'How many employees does Microsoft have?' (score=0.176)

The retriever returned pages 81, 22, and 75 - investor relations, the shareholder letter overview, and segment geographic data. Page 16, where the headcount figure lives, is not in the list. The top_retrieval_score of 0.176 is far below the 0.75 threshold, and the warning fires before the answer even reaches the model. In production, this log line is your early warning system.

The fields that matter most:

top_retrieval_score - the cosine similarity of the best-matching chunk. The 0.176 above is an extreme case; in practice, scores below 0.70 should be treated with suspicion. Track the distribution over time; a shift downward usually means document quality or query distribution has changed.
low_confidence_retrieval - a boolean flag you can aggregate in your monitoring dashboard. If 15% of production queries are flagging this, you have a systematic retrieval problem, not random noise.
retrieval_time_ms vs generation_time_ms - when latency degrades, these tell you which layer is the bottleneck. Retrieval slowdowns point to index size or embedding model throughput. Generation slowdowns point to context length or model tier.

Monitoring a RAG pipeline is not fundamentally different from monitoring an API. You need latency percentiles, error rates, and a signal for "this request probably didn't go well". The only RAG-specific addition is retrieval score distribution - and that is one structured log field.

The Organizational Gap

But instrumentation alone doesn't close the loop. The failure mode that none of the code above addresses is not technical - evaluation is treated as a one-time task rather than a discipline.

A team ships the pipeline with a passing test suite. Three months later, the PDF is updated to reflect new financials. Nobody reruns the golden dataset against the new document. The chunk boundaries shift. The retrieval hit rate drops from 85% to 60%. Users notice the answers getting worse but attribute it to "AI being unreliable" rather than a specific regression with a specific cause and a specific fix.

This is not a technical failure. It is an organizational failure. Evaluation was never wired into the deployment process.

The fix is structural: treat RAG evaluation the same way you treat API contract testing. Every time the underlying document changes, every time the embedding model is updated, and every time the chunking parameters are tuned, the test suite runs. A score below threshold blocks the deployment.

That requires someone to own the golden dataset, keep it current, and expand it as new query patterns emerge in production logs. In most organizations, that role does not exist today. It needs to.

If no one on your team owns the golden dataset, no one owns the quality of your RAG system. That is not an AI problem. It is a product ownership problem.

What Good Looks Like: A Minimal Viable Test Stack

You do not need a dedicated MLOps platform to do this properly.

Start with the golden dataset: 50 hand-verified (query, expected page, expected content) pairs covering the most common query categories in your use case. If resources are tight, 20 is enough to get started; the dataset grows as production logs reveal new failure patterns. The investment is one senior engineer and one afternoon. Once it exists, wire the retrieval hit rate test into every pull request that touches chunking, embedding config, or document ingestion. A drop below threshold blocks merge. The test runs in under two minutes against a local ChromaDB instance.

Then instrument production. Log retrieval scores on every request and alert when the 7-day moving average of low_confidence_retrieval exceeds 10% - that's the signal that your document or query distribution has shifted. Every quarter, pull the top 20 low-confidence queries from those logs, verify the expected answers manually, and add them to the golden dataset. The test suite grows. Coverage improves. The system gets harder to break without anyone noticing.

That is the full stack. No specialized tooling required beyond what most engineering teams already have.

Conclusion: The Trust Problem

A RAG system that is not evaluated is not a production system. It is a demo with a deployment URL and a confidence problem.

The danger is not that it fails catastrophically. The danger is that it fails quietly, at scale, on exactly the queries that matter most to your users - the precise financial figures, the specific regulatory disclosures, the numbers a decision might actually rest on. And because there is no alarm, no exception, no dashboard that goes red, the failure is invisible until trust has already eroded.

The patterns in this article are not novel. They are the same patterns that mature engineering organizations apply to APIs, databases, and distributed systems: define what correct looks like, measure it continuously, and alert when it degrades.

RAG pipelines deserve the same discipline. The code above is a starting point. The golden dataset is the actual investment. Build it this week.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Image by Gerd Altmann from Pixabay

Your Python Tool Needs Persistence - It Doesn't Need a Database Server

Developer Service — Thu, 07 May 2026 06:36:28 +0000

At some point, every internal tool, CLI utility, or developer script needs to remember something between runs. A list of environments. A job queue. A cache of API responses. A set of user preferences.

The default response is to reach for a database. Set up SQLite, write a schema, maybe add an ORM. That's fifteen minutes of infrastructure for a problem that might not warrant it, and fifteen minutes that turns into an hour once you factor in migrations, connection handling, and test fixtures.

There's a narrower tool for this class of problem. It's called TinyDB.

What TinyDB Actually Is

TinyDB is a Python library that stores structured data as documents in a plain JSON file. No server process. No connection string. No migration history. The entire database is a file on disk that you can open in a text editor.

It's not a replacement for PostgreSQL. It's not trying to be. But for internal tooling, CLI applications, local configuration storage, and developer utilities - the kind of code your team writes to support the product rather than ship as the product - it covers most persistence needs with a fraction of the operational surface area.

It can be installed it with a single command:

pip install tinydb

No credentials to rotate. No infrastructure to provision. No service to monitor.

The Real Cost of Over-Engineering Internal Tools

Internal tools accumulate infrastructure debt quietly. A script that needed a database gets one. The database needs to live somewhere, so it gets a connection string in the environment config. The connection string needs to be kept out of version control. Now you have secrets management for a tool that three engineers use twice a week.

TinyDB short-circuits that chain. The database is a file. It lives next to the code, or in a known path on the machine that runs it. There is nothing to provision, nothing to rotate, and nothing to monitor.

The tradeoff is real: no concurrency, no transactions, no relational integrity. If two processes write to the same file simultaneously, you will corrupt data. If you need joins, TinyDB is the wrong tool. These are known, documented limitations, not surprises that emerge in production.

The decision is straightforward: if the data fits in a file a human could read, and a single process owns the writes, TinyDB is appropriate. If either of those conditions doesn't hold, use something else.

A Concrete Example: CLI Task Manager

To make this tangible, here's how TinyDB works in a minimal CLI task manager, the kind of internal tool that actually gets built.

The full source code is in the companion GitHub repository: https://github.com/nunombispo/tinydb-article

Opening the database and organizing data into tables:

# file: db.py

from pathlib import Path

from tinydb import TinyDB
from tinydb.storages import MemoryStorage

# Default path for the production database
DEFAULT_DB_PATH = Path.home() / ".task_manager" / "tasks.json"


def get_db(db_path: Path | None = None, in_memory: bool = False) -> TinyDB:
    if in_memory:
        return TinyDB(storage=MemoryStorage)

    path = db_path or DEFAULT_DB_PATH
    path.parent.mkdir(parents=True, exist_ok=True)
    return TinyDB(path)


def get_tables(db: TinyDB) -> tuple:
    tasks = db.table("tasks")
    notes = db.table("notes")
    return tasks, notes

Both tables live in a single tasks.json file. Open it and you see exactly what's stored:

{
  "tasks": {
    "1": {
      "title": "Write TinyDB article",
      "done": false,
      "created_at": "2026-04-14T08:28:41.294890"
    },
    "2": {
      "title": "Review PR #42",
      "done": false,
      "created_at": "2026-05-04T09:52:27.316977"
    }
  },
  "notes": {
    "1": {
      "task_id": 1,
      "text": "Check MemoryStorage in the TinyDB docs",
      "created_at": "2026-04-14T08:36:32.672970"
    }
  }
}

No binary formats. No opaque files. If something goes wrong, you can read the database with cat.

Inserting a document:

Each insert returns a doc_id, TinyDB's auto-assigned integer key. No schema to define first:

# file: models.py

def add_task(tasks: Table, title: str) -> int:
    doc_id = tasks.insert({
        "title": title,
        "done": False,
        "created_at": datetime.now().isoformat(),
    })
    return doc_id

Querying:

TinyDB's query interface is built around a Query object. Field conditions, regex matching, and logical operators are all supported:

# file: models.py

def get_pending_tasks(tasks: Table) -> list[Document]:
    Task = Query()
    return sorted(tasks.search(Task.done == False), key=lambda t: t.doc_id)

def get_done_tasks(tasks: Table) -> list[Document]:
    Task = Query()
    return sorted(tasks.search(Task.done == True), key=lambda t: t.doc_id)

def search_tasks(tasks: Table, keyword: str) -> list[Document]:
      Task = Query()
      return tasks.search(Task.title.matches(f".*{re.escape(keyword)}.*", flags=re.IGNORECASE))

def get_task(tasks: Table, task_id: int) -> Document | None:
    return tasks.get(doc_id=task_id)

def get_all_tasks(tasks: Table) -> list[Document]:
    return sorted(tasks.all(), key=lambda t: t.doc_id)

Updating and deleting:

Updates and deletes follow the same query pattern. Deleting a task cascades to its notes explicitly in code, TinyDB has no foreign key enforcement, so the application owns that logic:

# file: models.py

def mark_done(tasks: Table, task_id: int) -> bool:
    if tasks.get(doc_id=task_id) is None:
        return False
    tasks.update({"done": True}, doc_ids=[task_id])
    return True

def mark_pending(tasks: Table, task_id: int) -> bool:
    if tasks.get(doc_id=task_id) is None:
        return False
    tasks.update({"done": False}, doc_ids=[task_id])
    return True

def upsert_task(tasks: Table, title: str) -> int:
    Task = Query()
    doc_ids = tasks.upsert(
        {"title": title, "done": False, "created_at": datetime.now().isoformat()},
        Task.title == title,
    )
    return doc_ids[0]

def delete_task(tasks: Table, notes: Table, task_id: int) -> bool:
    if tasks.get(doc_id=task_id) is None:
        return False

    Note = Query()
    notes.remove(Note.task_id == task_id)

    tasks.remove(doc_ids=[task_id])
    return True

The cascade is two lines. It's visible, it's testable, and it does exactly what it says.

Swapping the storage backend for tests:

This is where TinyDB earns its keep in a team setting. The storage backend is fully decoupled from the query layer. Switching to an in-memory store for tests requires changing one line:

# file: db.py

def get_db(db_path: Path | None = None, in_memory: bool = False) -> TinyDB:
    if in_memory:
        return TinyDB(storage=MemoryStorage)

    path = db_path or DEFAULT_DB_PATH
    path.parent.mkdir(parents=True, exist_ok=True)
    return TinyDB(path)

Every insert, query, update, and delete works identically. No test database to set up, no fixtures to seed, no files to clean up. The full test suite for this project, 26 tests, runs against MemoryStorage in under a second.

The CLI in action:

$ tasks add "Write TinyDB article"
✓ Task added (id=1): Write TinyDB article

$ tasks list
                      Pending Tasks
┏━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃    ID ┃ Title                ┃   Status   ┃ Created    ┃
┡━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│     1 │ Write TinyDB article │  pending   │ 2026-04-14 │
│     2 │ Review PR #42        │  pending   │ 2026-04-14 │
│     3 │ Deploy staging       │  pending   │ 2026-04-14 │
└───────┴──────────────────────┴────────────┴────────────┘

$ tasks done 1
✓ Task 1 marked as done.

$ tasks search "article"
               Search results for "article"
┏━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃    ID ┃ Title                ┃   Status   ┃ Created    ┃
┡━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│     1 │ Write TinyDB article │  pending   │ 2026-04-14 │
└───────┴──────────────────────┴────────────┴────────────┘

$ tasks note 1 "Check MemoryStorage in the TinyDB docs"
✓ Note added (id=1) to task 1.

$ tasks delete 2 --yes
✓ Task 2 deleted.

When to Stop Using TinyDB

TinyDB is explicit about its limits. The documentation lists them. That honesty is worth taking at face value.

Stop using TinyDB when any of these conditions appear: more than one process writes to the database concurrently, the dataset grows into the tens of thousands of documents and performance becomes noticeable, or you need relational integrity that the application layer shouldn't have to enforce manually.

The natural upgrade path is SQLite, same "no server required" property, built for performance and concurrency. Beyond that, the usual options apply.

The failure mode to avoid is using TinyDB past the point where it fits and accumulating workarounds instead of upgrading. That's a different kind of infrastructure debt, and it's avoidable if you know the boundary going in.

Conclusion

Most persistence problems in developer tooling aren't database problems, they're "I need this to survive the next run" problems. TinyDB solves exactly that, with no server, no schema, and no ceremony.

The task manager in this article, named tables, cross-table references, cascade deletes, 26 tests on MemoryStorage, fits entirely in a JSON file you can open in a text editor. That's not a limitation. For this class of tool, it's the right design.

Reach for TinyDB when you're building something for developers, by a developer. Reach for something else when the data outgrows a file you can read.

All code is in the companion GitHub repository: https://github.com/nunombispo/tinydb-article

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Serving a Live Battery Dashboard from ESP32-C3 with Microdot and MicroPython

Developer Service — Mon, 13 Apr 2026 06:48:36 +0000

You've been there: the hardware works, the sensors are wired up, and somewhere inside main.py a print() is faithfully reporting voltage every second into a serial terminal that nobody will ever open again.

The gap between "working prototype" and "something you'd actually show someone" is often just a few hundred lines of web code - and on MicroPython, that's easier than you might expect.

This article walks you through building a live battery charge dashboard that runs entirely on an ESP32-C3 Super Mini. The device connects to your local Wi-Fi, hosts a small web page using Microdot, and serves real-time voltage, current, and power readings from an INA219 power monitor while a TP4056 charges an 18650 cell.

Any browser on the same network can open the page and watch the charge progress unfold.

The hardware is a simple, but useful circuit. The core lesson here is Microdot: how to think about routes, JSON APIs, and a minimal browser UI that polls the device.

These patterns apply to anything you'd wire to a micro-controller, like temperature sensors, fuel gauges, motion detectors, so even if battery monitoring isn't your use case, this structure is.

What You Will Build

The finished project is a single-page dashboard served directly from the ESP32-C3. It shows:

Bus voltage (V) - measured by the INA219 on the battery side
Charge current (mA) - derived from the INA219 shunt voltage (in series with battery)
Power (mW) - computed as V × mA
Last update time - so you can immediately see if the connection is stale
Connection status indicator - a clear visual signal that the browser is actively receiving data

It is intentionally small. On microcontrollers, the fastest route to reliability is a small surface area: tiny payloads, tiny pages, and simple control flow.

This is the running dashboard while charging a 18650 battery:

You won't find React, WebSockets, or authentication here - not because those are wrong, but because they distract from the architectural lesson.

Why Microdot?

Before diving into code, it's worth understanding why Microdot is the right tool here rather than, say, raw sockets or a stripped-down asyncio HTTP handler.

MicroPython already trims the Python standard library to fit in a few hundred kilobytes. Most full web frameworks - Flask or FastAPI - assume a runtime environment with considerably more headroom.

Microdot was designed specifically for constrained environments: it's small enough to live comfortably on an ESP32, and it gives you the ergonomics that make web code readable.

The mental model maps directly to Flask if you know it:

from microdot import Microdot

app = Microdot()

@app.route('/')
def index(request):
    return "Hello from ESP32", 200, {'Content-Type': 'text/plain'}

app.run(port=80)

That's it. An app object, route decorators, and sensible response handling. Microdot supports both sync and async handlers, which matters on ESP32 where uasyncio is your concurrency model and blocking the event loop kills Wi-Fi stability.

For this project, the full Microdot API surface we'll use is:

Feature	Used for
`@app.route('/')`	Serving the HTML dashboard page
`@app.route('/api/metrics')`	JSON endpoint the browser polls
Returning tuples `(body, status, headers)`	Setting content type, handling errors
Sync handlers (`def`)	Keeping the project small and readable

Nothing too exotic. That's the point.

System Overview

There are two parallel "flows" in this project, and keeping them separate in your head will save debugging time later.

Data flow:

INA219 (I²C) → ESP32-C3 (MicroPython) → Microdot routes → browser

Power and charge flow:

5V USB → TP4056 → 18650 cell
                 ↓
           INA219 (in series with battery leg)
                 ↓
           ESP32-C3 (powered from USB separately)

The INA219 sits in the battery leg - measuring what actually goes into the cell - while the ESP32 runs the web server and polls the sensor over I²C.

These are logically independent: the charger doesn't care about the web server, and the web server doesn't care about charging chemistry.

Hardware

You need four main components, plus a few passives and connectors.

ESP32-C3 Super Mini: The ESP32-C3 Super Mini is a good choice here for several reasons: it's compact, supports Wi-Fi, runs MicroPython well, and its USB-C connector makes firmware updates painless.
INA219: The INA219 is a current/power monitor that communicates over I²C. It measures two things directly:
- Bus voltage - the voltage on the load side of the shunt resistor (this is your battery voltage)
- Shunt voltage - the tiny voltage drop across the shunt resistor caused by current flowing through it
- From the shunt voltage and a known shunt resistance, you get current: I = V_shunt / R_shunt.
TP4056: The TP4056 is a single-cell Li-ion linear charger. Its charging algorithm follows two phases:
- Constant Current (CC): The charger delivers a fixed current (typically 1A on modules with a 1.2kΩ programming resistor) until the cell voltage reaches approximately 4.2V.
- Constant Voltage (CV): The charger holds 4.2V and the current tapers toward zero as the cell reaches full charge.
18650 Cell: Use a genuine cell from a reputable brand. Counterfeit 18650s with inflated capacity ratings are widespread and can be unsafe.

Safety

Li-ion cells store a significant amount of energy and can fail dangerously if mistreated. Pay attention, especially on a circuit you're building on a breadboard.

Never charge unattended, especially during initial bring-up.
Stop immediately if anything gets warm to the touch, smells unusual, swells, or hisses.
Use a proper enclosure or at minimum keep the cell away from flammable materials on your bench.
Don't short the battery terminals - ever.
Don't let the cell discharge below ~2.5V.

Wiring Summary

In this build, the INA219 is placed in the battery leg (in series), so it measures the current actually going into (or coming out of) the cell:

TP4056 B+ ──── INA219 IN+ ──── INA219 IN- ──── Cell +

All grounds must be common (ESP32, INA219, and TP4056 share GND).

For reference, the full wiring diagram:

This is how it looks like on a breadboard:

Note: You might see in this photo the TP4056 connections and think the polarity is reversed. I have made a mistake and soldered them reversed, in the picture the TP4046 B- is red and B+ is black.

Remember: The TP4056 B- should be to ground and TP4056 B+ goes to INA219 V+. Soldered them correctly, don' follow my example...

Software: Project Structure

The project is split into four files that live on the ESP32 filesystem:

├── main.py      # Wi-Fi connection + server startup
├── web.py       # Microdot app, routes, and inline HTML/JS
├── sensor.py    # INA219 init and read_metrics()
└── ina219.py    # Low-level INA219 driver

This structure makes the project maintainable. web.py owns the HTTP contract. sensor.py owns hardware access. Neither knows about the other's implementation details. The route handlers in web.py call sensor.py functions and translate the results into HTTP responses.

The article only shows only the important snippets below. The complete, working code (including the full HTML/CSS) lives in the repo: https://github.com/nunombispo/microdot-article

First, you will need to install the microdot library, which for this case means copying the microdot.py from the repository to the device. Full install instructions are here:

`sensor.py` - Hardware Reads

The goal is: hide the INA219/I²C setup behind one function, and expose a single read_metrics() that returns a JSON-friendly dict. The sensor is initialized lazily (first call), which keeps main.py minimal.

_ina = None

def _get_ina219():
    global _ina
    if _ina is not None:
        return _ina

    # NOTE: Update pins to match your ESP32-C3 board wiring.
    i2c = I2C(0, scl=Pin(9), sda=Pin(8), freq=400_000)
    _ina = INA219(i2c=i2c, address=0x40)
    _ina.configure_32v_2a()
    return _ina

def read_metrics() -> dict:
    ina = _get_ina219()
    voltage_v = ina.bus_voltage_v()
    current_ma = ina.current_ma()
    power_mw = voltage_v * current_ma
    return {
        "voltage_v": voltage_v,
        "current_ma": current_ma,
        "power_mw": power_mw,
    }

A few deliberate choices here:

_ina is module-level state. The INA219 is initialized once and reused across requests. Reinitializing I²C on every poll is wasted work and can cause intermittent bus issues.
The API adds a timestamp separately (in web.py) so sensor.py stays focused on measurement, not transport/UI concerns.

`web.py` - Routes and Inline UI

There are only two routes: one serves the page, one serves JSON metrics. Microdot supports both sync and async def handlers; this project uses sync handlers to keep things simple.

from microdot import Microdot, Response
from sensor import read_metrics

app = Microdot()
Response.default_content_type = "text/html; charset=utf-8"

def _mono_ms() -> int:
    ticks_ms = getattr(time, "ticks_ms", None)
    return int(ticks_ms()) if ticks_ms else int(time.time() * 1000)

@app.route('/')
def index(_request):
    return HTML  # full HTML/CSS lives in the repo

@app.route('/api/metrics')
def api_metrics(_request):
    try:
        metrics = read_metrics()
        metrics["ts_ms"] = _mono_ms()
        return metrics
    except Exception:
        return {"error": "sensor_unavailable"}, 503

Notice how thin the handlers are. The API route is intentionally just “read the sensor, return JSON”, with a single stable error shape for the UI. That keeps the HTTP layer boring - which is exactly what you want on a microcontroller.

On the browser side, the core loop is a simple poll-and-render (full page in the repo):

async function poll() {
  const r = await fetch('/api/metrics', { cache: 'no-store' });
  if (!r.ok) throw new Error('HTTP ' + r.status);
  const d = await r.json();
  voltageEl.textContent = d.voltage_v.toFixed(2) + ' V';
  currentEl.textContent = d.current_ma.toFixed(2) + ' mA';
}
setInterval(poll, 1000);

`main.py` - Boot and Server Start

The startup sequence is short and deliberate: connect Wi‑Fi, initialize hardware, then start the server.

import time
import network
from web import app


_ENV = _read_dotenv()
WIFI_SSID = _ENV.get("WIFI_SSID", "")
WIFI_PASSWORD = _ENV.get("WIFI_PASSWORD", "")

def connect_wifi(ssid: str, password: str, timeout_s: int = 20) -> str:
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        wlan.connect(ssid, password)
        start = time.time()
        while not wlan.isconnected():
            if time.time() - start > timeout_s:
                raise RuntimeError("Wi-Fi connect timeout")
            time.sleep(0.25)
    return wlan.ifconfig()[0]

def main() -> None:
    if not WIFI_SSID or not WIFI_PASSWORD:
        raise RuntimeError("Missing WIFI_SSID/WIFI_PASSWORD in .env")
    ip = connect_wifi(WIFI_SSID, WIFI_PASSWORD)
    print("Wi-Fi connected, IP:", ip)
    print("Open http://%s/ in a browser" % ip)
    app.run(port=80, debug=False)

main()

The startup sequence matters: connect Wi‑Fi first, then start the server.

What to Expect During a Charge Session

With the INA219 measuring battery-side current, a typical TP4056 session has a recognizable shape:

Constant Current phase (early):

Current is relatively flat - usually 800–1000 mA on a standard 1A module, though weak USB supplies often limit this to 600–700 mA.
Voltage climbs steadily from around 3.6–3.7V toward 4.2V.
This phase lasts 1–2 hours depending on cell capacity and charge current.

Constant Voltage phase (late):

Voltage plateaus near 4.2V.
Current begins a gradual exponential decay.
The TP4056 considers the cell full when current drops below approximately C/10 (for a 2000mAh cell, that's around 200 mA).
This phase can last another hour.

Let' see the CC phase in action:

And if it gets disconnect for some reason:

Conclusion

Micro-controllers don’t need a “real web stack” to feel like a real product. With Microdot, you get the same clean mental model as Flask - routes, JSON, and a tiny UI - without dragging in heavy dependencies or complex infrastructure.

In this build you now have:

A stable contract: GET /api/metrics returns a small JSON payload that any UI can consume.
A maintainable split: sensor logic stays in sensor.py, HTTP logic stays in web.py.

If you build something similar for a different sensor, keep the same pattern: one hardware module that returns a dict, one Microdot JSON route, and a minimal page that polls and renders.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Python Decorators - The Three-Layer Pattern

Developer Service — Tue, 07 Apr 2026 06:42:58 +0000

You've probably used decorators before, @login_required, @cache, @app.route("/").

They feel like magic: one line above a function and its behavior changes entirely. But at some point you need more than an on/off switch. You need @retry(times=3), @cache(maxsize=128), @require_role("admin"). That's when most developers hit a wall.

The good news is that parameterized decorators aren't a different concept, they're just one extra function layer on top of what you already know. Once it clicks, you'll see the pattern everywhere in the Python ecosystem and start reaching for it naturally in your own code.

In this tutorial you'll go from a plain decorator to a fully parameterized one, understand exactly why the three-layer structure is necessary, and build a handful of decorators you could drop into a real project today, including a @retry decorator that handles flaky network calls and a @require_role guard for protecting API endpoints.

Let's start with a quick recap of how decorators work under the hood, because that foundation is what makes everything else make sense.

Decorators Under the Hood

If you've written a decorator before, you know the syntax.

But let's slow down and look at what Python is actually doing, because that mental model is the key to understanding parameterized decorators later.

Here's the simplest possible decorator:

def shout(fn):
    def wrapper(*args, **kwargs):
        result = fn(*args, **kwargs)
        return str(result).upper()
    return wrapper

@shout
def greet(name):
    """Greet the user"""
    return f"hello, {name}"

print(greet("alice"))  # "HELLO, ALICE"

The @shout syntax is pure shorthand. Python translates it into exactly this:

greet = shout(greet)

That's the whole trick.

shout receives the original function, returns a new one (wrapper), and Python rebinds the name greet to point at wrapper from that point on.

Every time you call greet(...), you're actually calling wrapper(...), which calls the original function internally.

Visually, the transformation looks like this:

The identity problem

There's a subtle issue with the above. After decoration, greet is no longer quite itself:

print(greet.__name__)   # "wrapper"  ← wrong
print(greet.__doc__)    # None       ← lost

Python sees wrapper where it expects greet. This breaks help(), logging, stack traces, and any tool that inspects function metadata.

The fix is functools.wraps:

import functools

def shout(fn):
    @functools.wraps(fn)       # copies __name__, __doc__, __annotations__, __wrapped__
    def wrapper(*args, **kwargs):
        result = fn(*args, **kwargs)
        return str(result).upper()
    return wrapper

@shout
def greet(name):
    """Greet the user"""
    return f"hello, {name}"

print(greet.__name__)   # "greet"
print(greet.__doc__)    # "Greet the user"

Now greet.__name__ is "greet" and greet.__wrapped__ holds a reference to the original function, useful for testing, which you'll see later.

Think of @functools.wraps as the professional finish on every decorator you write. It costs one line and saves a lot of confusion.

With that foundation in place, it's time to add parameters and that's where things get genuinely interesting.

All the code of this and the other examples in this article are available at: https://github.com/nunombispo/python-decorators-article

The Three Layers of a Parametrized Decorator

A plain decorator takes a function and returns a function. That's one layer.

The moment you add parameters, you need somewhere to put them and that somewhere is a new outer function that runs before the decorator does.

That gives you three layers total.

Here's the structure, stripped to its bones:

def outer(param):          # layer 1: receives your arguments
    def decorator(fn):     # layer 2: receives the function
        def wrapper(*args, **kwargs):   # layer 3: runs on every call
            # param is available here
            return fn(*args, **kwargs)
        return wrapper
    return decorator

When Python sees @outer(param), it does this in two steps:

decorator = outer(param)   # step 1: call the factory, get a decorator back
greet = decorator(greet)   # step 2: apply that decorator to the function

Which collapses to the familiar one-liner:

greet = outer(param)(greet)

The diagram maps this exactly:

Why the extra layer exists

A plain @decorator works because Python calls it with the function as its only argument.

But @outer(param) is evaluated before Python passes the function, outer(param) runs immediately and must return something callable that accepts a function. That's decorator.

You're not adding complexity for its own sake; the structure is forced by the order Python evaluates things.

Closure capture

The reason param is available inside wrapper, despite outer having already returned, is closures.

When wrapper is defined inside decorator, which is defined inside outer, it captures references to variables in the enclosing scopes. param doesn't get copied into wrapper; it stays alive because wrapper holds a reference to it.

If you want a deeper understanding of how Python resolves variable names across nested scopes, check out A Practical Guide to Python Variable Scope - it covers the LEGB rule and closures in detail and will help you understand it better.

This is what makes the pattern so powerful. You can pass in times=3 or role="admin" at decoration time, and that value is quietly available every single time the wrapped function is called, no global state, no configuration objects, just a variable captured in a closure.

A concrete way to see this:

def outer(param):
    print(f"outer called with {param}")      # runs once, at decoration time
    def decorator(fn):
        print(f"decorator called with {fn}") # runs once, at decoration time
        def wrapper(*args, **kwargs):
            print(f"wrapper called, param={param}")  # runs on every call
            return fn(*args, **kwargs)
        return wrapper
    return decorator

@outer("hello")
def greet(name):
    return name

Running this, you'll see outer and decorator print immediately when the module loads. wrapper only prints when you actually call greet(...).

That timing distinction, decoration time vs call time, is one of the most common sources of bugs when writing parameterized decorators.

With the structure clear, let's build something real with it.

Building Your First Parameterized Decorator

Time to put the three-layer pattern to work.

@repeat is the perfect first example - the logic is simple enough that the decorator structure stays front and center.

Here's the goal:

@repeat(times=3)
def greet(name):
    print(f"Hello, {name}!")

greet("Alice")
# Hello, Alice!
# Hello, Alice!
# Hello, Alice!

And here's the full implementation:

import functools

def repeat(times):                              # layer 1: factory
    def decorator(fn):                          # layer 2: decorator
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):           # layer 3: wrapper
            for _ in range(times):
                result = fn(*args, **kwargs)
            return result
        return wrapper
    return decorator

Let's walk through each layer.

Layer 1 - repeat(times)

This is the factory. It runs exactly once, at decoration time, when Python evaluates @repeat(times=3). Its only job is to receive the argument and return a decorator. After this call, times is captured in the closure and available to everything nested inside.

Layer 2 - decorator(fn)

This is what Python passes the function to. It also runs once, at decoration time, immediately after repeat(times) returns. It receives greet as fn, applies @functools.wraps(fn) to preserve its identity, and returns wrapper as the replacement.

Layer 3 - wrapper(*args, **kwargs)

This is the function your code actually calls every time you write greet("Alice"). It has access to both times (from layer 1's closure) and fn (from layer 2's closure). The loop runs, the original function is called each iteration, and the last result is returned.

A note on return value

Storing result and returning it after the loop means the caller always gets the return value of the last call. For a @repeat decorator that's fine - but in other decorators, returning inside the loop on the first successful call is often the right choice. Always think about what your wrapper should hand back to the caller.

Checking it works

Thanks to functools.wraps, the function's identity is intact:

print(greet.__name__)     # "greet"
print(greet.__wrapped__)  # <function greet at 0x...>

And you can access the original unwrapped function via __wrapped__ - handy when testing, since you can call greet.__wrapped__("Alice") to bypass the decorator entirely.

Decorators You'll Actually Ship

@repeat is a clean teaching example, but the pattern really earns its keep when the logic inside wrapper does something meaningful.

Here are three parameterized decorators that solve real problems.

`@retry` - handling flaky operations

Network calls fail. APIs rate-limit. Database connections drop.

A @retry decorator lets you express resilience in one line at the call site rather than cluttering every function with try/except loops.

import time
import functools

def retry(times=3, delay=1.0, exceptions=(Exception,)):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(1, times + 1):
                print(f"Attempt {attempt} of {times}")
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < times:
                        time.sleep(delay)
            raise last_exc
        return wrapper
    return decorator

@retry(times=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
    ...  # flaky network call

A few design decisions worth noting.

The exceptions parameter lets the caller decide what counts as a retry, you never want to silently swallow a ValueError or KeyError that signals a bug in your own code.

Storing last_exc and re-raising it after the loop preserves the original trace-back rather than replacing it with a generic message.

And returning immediately on success means you don't pay any extra overhead on the happy path.

`@timer` - measuring execution time

Useful during development and in production monitoring.

The unit parameter keeps the output readable whether you're measuring a 2-ms database query or a 45-second batch job.

import time
import functools

def timer(unit="ms"):
    units = {"ms": 1_000, "s": 1, "us": 1_000_000}

    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = fn(*args, **kwargs)
            elapsed = (time.perf_counter() - start) * units[unit]
            print(f"{fn.__name__} took {elapsed:.2f}{unit}")
            return result
        return wrapper
    return decorator

@timer(unit="ms")
def query_database(sql):
    ...

@timer(unit="s")
def generate_report():
    ...

time.perf_counter() is the right choice here - it's the highest-resolution clock available and unaffected by system clock adjustments.

Notice that result is captured before printing, so the timing includes only the function itself, not the print call.

`@require_role` - protecting endpoints

A common pattern in Django, Flask, and FastAPI: restrict a view or route handler to users with a specific role, without repeating the same guard logic in every function body.

import functools
from functools import wraps

class User:
    def __init__(self, name, role):
        self.name = name
        self.role = role

def require_role(role):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            # in a real app, current_user comes from your auth context
            user = kwargs.get("user") or (args[0] if args else None)
            if user is None or user.role != role:
                raise PermissionError(f"Role '{role}' required.")
            return fn(*args, **kwargs)
        return wrapper
    return decorator

@require_role("admin")
def delete_user(user, target_id):
    ...

@require_role("editor")
def publish_post(user, post_id):
    ...

The role string is captured in the closure at decoration time, so delete_user and publish_post each carry their own independent requirement - no shared state, no configuration lookup at runtime.

In a real application you'd pull the current user from a request context or session rather than a function argument, but the decorator structure stays identical.

Stacking Decorators

Nothing stops you from applying multiple decorators to the same function.

In fact, combining small focused decorators is one of the cleanest ways to compose behavior in Python:

@retry(times=3, delay=0.5)
@timer(unit="ms")
def fetch_data(url):
    ...

But the order matters, and it trips people up.

Application order is bottom-up

Python applies decorators from the one closest to def outward. The equivalent explicit form makes this concrete:

fetch_data = retry(times=3, delay=0.5)(timer(unit="ms")(fetch_data))

timer wraps fetch_data first, then retry wraps the result of that. So at runtime, calling fetch_data(url) enters retry's wrapper first, which calls timer's wrapper, which calls the original function.

A useful mental model: read the decorators from top to bottom to understand the call order at runtime - the topmost decorator is entered first.

@retry ← entered first at runtime, sees everything below 
@timer ← entered second def fetch_data

In this arrangement, @timer measures the total time including all retry attempts:

@timer(unit="ms")    ← entered first, measures total time including all retries
@retry(times=3)      ← retries happen inside the timed block
def fetch_data(url):
    ...

Swap the order and it measures only a single attempt:

@retry(times=3)      ← entered first, retries everything below
@timer(unit="ms")    ← measures a single attempt
def fetch_data(url):
    ...

Neither is wrong - they measure different things. Being deliberate about order is part of using stacked decorators well.

Debugging stacked decorators

When something goes wrong in a stack, the first tool to reach for is __wrapped__. Because functools.wraps sets this attribute on every wrapper, you can peel back the layers manually:

fetch_data            # retry's wrapper
fetch_data.__wrapped__            # timer's wrapper
fetch_data.__wrapped__.__wrapped__ # original fetch_data

If one of your decorators is missing functools.wraps, the chain breaks at that point - __wrapped__ won't exist and you can't see past it. This is the most common reason to go back and add @functools.wraps(fn) to an older decorator you didn't write yourself.

For a quicker inspection, inspect.unwrap() does the peeling for you:

import inspect

original = inspect.unwrap(fetch_data)
print(original)  # <function fetch_data at 0x...>

It follows __wrapped__ all the way down to the original function in one call. Pair it with inspect.signature() to confirm the original signature is intact:

print(inspect.signature(fetch_data))          # (*args, **kwargs) if wraps is missing
print(inspect.signature(inspect.unwrap(fetch_data)))  # (url) ← correct

A mangled signature is usually the first sign that a decorator in the stack is missing functools.wraps.

Once your decorator is solid, it's worth automating those checks - GitHub Actions for Python Projects walks you through running tests, linting, and type checks on every push with a single YAML file.

Conclusion

Parametrized decorators look intimidating at first glance, but the pattern is simpler than it appears: a factory captures your arguments, returns a decorator, which wraps your function in a closure that keeps everything in scope.

That three-layer structure is all there is to it - and once it clicks, you'll spot it everywhere. Django's @permission_required, FastAPI's @app.get("/"), Tenacity's @retry - they're all the same pattern dressed for different jobs.

Three things worth carrying forward.

First, always apply @functools.wraps - it costs one line and keeps __name__, __doc__, and __wrapped__ intact, which matters the moment you start debugging or writing tests.

Second, be deliberate about stacking order; the topmost decorator is entered first at runtime, so it sees everything below it.

Third, keep the decoration-time vs call-time distinction in mind - the factory and decorator layers run once when the module loads, wrapper runs on every call.

Mixing those two up is the most common source of subtle bugs when writing parameterized decorators.

If you find yourself copying @retry or @timer into every new project, that's a sign they've earned their own package - How to Build and Publish a Python Package to PyPI walks you through the whole process with a real project from start to finish.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Build a Terminal Disk Analyzer in Python with Textual

Developer Service — Mon, 30 Mar 2026 06:34:27 +0000

You've been there. CI is slow, the build server is running low on space, and df -h tells you the disk is 94% full - but not why.

You du -sh * your way through a few directories, mentally add up the numbers, and eventually find the culprit: six months of accumulated .venv folders, a Docker layer cache that got out of hand, or a node_modules directory that somehow outlived the project that created it.

It's not a disaster. It's just friction - the kind that adds up.

ncdu is a tool that makes this painless. It scans a directory tree, ranks everything by size, and lets you navigate and delete without ever leaving the terminal. If you're on a remote server or inside a container, it should be the first thing you reach for.

In this article, you'll build your own version in Python, called pydusk, using the textual TUI framework. It's a practical project - useful on its own - and a good way to get hands-on with Textual's core patterns: background workers, modal screens, reactive state, and a clean "state → render" loop.

What We're Building

pydusk is a keyboard-driven (with mouse support) terminal application that:

Recursively scans a directory and calculates sizes
Displays entries ranked largest-first with inline usage bars
Lets you navigate the tree, enter sub-directories, and go back up
Lets you delete files or directories with a confirmation prompt
Runs as a CLI tool: python pydusk.py ~/projects

Here's what it looks like in action:

Confirmation modal to delete a file/directory:

Notification after deleting a file/directory:

Architecture

pydusk is a single file, pydusk.py, with about 450 lines of Python.

It's split into three layers that are loosely coupled by design:

the scanner knows nothing about the UI
the UI knows nothing about the CLI
and Typer glues everything together at the entry point.

That separation also means the scanner can be pulled out and used in other contexts without touching the TUI code.

Scanner layer - pure stdlib (os, pathlib, dataclasses). The scan() function walks the filesystem using os.scandir, which is faster than os.walk because it retrieves file metadata in the same syscall as the directory listing. It builds a tree of DiskEntry dataclasses, sorted largest-first at every level. Symlinks are skipped to avoid loops and permission errors are swallowed silently so the tool works on any real system.

Background worker - the scanner runs inside a Textual @work(thread=True) worker, keeping the UI responsive during large scans. Once the scan finishes, call_from_thread() safely hands the result back to the main thread to update the display.

TUI layer - a textual.App subclass called DuskApp. It holds a navigation stack (a plain Python list of DiskEntry objects) and renders it into a DataTable widget on every state change. The delete flow uses a ModalScreen subclass that returns a bool via dismiss() - Textual's idiomatic pattern for confirmation dialogs. All keyboard shortcuts are declared as BINDINGS.

CLI layer - a typer command that accepts an optional path argument, validates that it exists and is a directory, and hands it to DuskApp. Nothing more.

The only dependencies needed, from the requirements.txt:

textual
typer

Step 1 - The Data Model and Scanner

Start by deciding what a "node" in your tree looks like. You want something that can represent a file or a directory, carry its computed size, and (for directories) hold children.

You represent every file and directory as a DiskEntry dataclass:

from dataclasses import dataclass, field
from pathlib import Path

@dataclass
class DiskEntry:
    path: Path
    size: int
    is_dir: bool
    children: list["DiskEntry"] = field(default_factory=list)

    @property
    def name(self) -> str:
        return self.path.name or str(self.path)

The children field uses field(default_factory=list) instead of a mutable default, if you're not sure why that distinction matters, I covered it in detail in Python Trick: Using dataclasses with field(default_factory=...).

The scanner uses os.scandir, which is faster than os.walk because it avoids extra stat calls, it gets file metadata in the same syscall as the directory listing.

The goal is to return a DiskEntry where children is a list of more DiskEntry objects, sorted largest-first.

You build the entire tree in one recursive function, skipping symlinks (to avoid loops) and swallowing permission errors (because real systems always have unreadable paths):

def scan(path: Path) -> DiskEntry:
    path = path.resolve()
    children: list[DiskEntry] = []
    try:
        with os.scandir(path) as it:
            for entry in it:
                try:
                    ep = Path(entry.path)
                    if entry.is_symlink():
                        size = entry.stat(follow_symlinks=False).st_size
                        children.append(DiskEntry(ep, size, False))
                    elif entry.is_dir(follow_symlinks=False):
                        children.append(scan(ep))
                    else:
                        size = entry.stat(follow_symlinks=False).st_size
                        children.append(DiskEntry(ep, size, False))
                except (PermissionError, OSError):
                    pass
    except (PermissionError, OSError):
        pass

    children.sort(key=lambda e: e.size, reverse=True)
    total = sum(c.size for c in children)
    return DiskEntry(path, total, True, children)

See the full implementation in pydusk.py.

Step 2 - Building the TUI with Textual

Now you need a UI that can render a list and react to input. Textual works well here because it gives you a layout system, widgets like DataTable, a built-in footer for keybindings, and a clean event/message model.

The layout is intentionally minimal: a header, a breadcrumb bar, a DataTable for the list, and a status line.

This keeps the "disk usage" part front-and-center while still giving you enough UI to navigate comfortably:

from textual.app import App, ComposeResult
from textual.widgets import DataTable, Footer, Header, Static

class DuskApp(App):
    TITLE = "pydusk"

    BINDINGS = [
        Binding("up", "move_up", "Up", show=True, priority=True),
        Binding("down", "move_down", "Down", show=True, priority=True),
        Binding("right,enter", "enter_dir", "Enter", show=True, priority=True),
        Binding("left,backspace", "go_up", "Up dir", show=True, priority=True),
        Binding("d", "delete",  "Delete", show=True),
        Binding("r", "rescan",  "Rescan", show=True),
        Binding("q", "quit",    "Quit",   show=True),
    ]

    def compose(self) -> ComposeResult:
        yield Header(show_clock=True)
        yield Static("", id="breadcrumb")
        yield DataTable(cursor_type="row", zebra_stripes=True)
        yield Static("Scanning…", id="status")
        yield Footer()

Navigation state is a plain Python list used as a stack.

Each time you enter a directory you append the DiskEntry to _stack; going up pops from it.

That makes breadcrumb rendering trivial (it's just the stack joined with /) and keeps your state model dead simple:

def action_enter_dir(self) -> None:
    row_key = self._selected_row_key()
    if row_key == "__parent__":
        self.action_go_up()
        return
    entry = self._entry_for_row_key(row_key)
    if entry and entry.is_dir:
        self._stack = self._stack + [entry]
        self._refresh_table()

def action_go_up(self) -> None:
    if len(self._stack) > 1:
        came_from = self._stack[-1]
        self._stack = self._stack[:-1]
        self._refresh_table()
        self._restore_cursor_to_row_key(str(came_from.path))

If you prefer, you can also make the list clickable.

DataTable posts a RowSelected message when you select a row (including via mouse click).

You can handle that message and reuse the same navigation actions:

from textual import on
from textual.widgets import DataTable

@on(DataTable.RowSelected)
def _on_table_row_selected(self, event: DataTable.RowSelected) -> None:
    row_key = event.row_key.value
    if row_key == "__parent__":
        self.action_go_up()
        return

    entry = self._entry_for_row_key(row_key)
    if entry and entry.is_dir:
        self._stack = self._stack + [entry]
        self._refresh_table()

This keeps mouse and keyboard behavior consistent: clicking a directory enters it, and clicking .. goes up one level.

Step 3 - Non-Blocking Scan with `@work`

Scanning a large directory can take several seconds, and on big filesystems it can be minutes.

If you do that work on the main thread, the UI can't repaint or process input, which feels like a crash.

Textual solves this cleanly with the @work decorator, which runs a function in a background thread and lets you safely push results back to the UI thread:

from textual import work

@work(thread=True)
def _do_scan(self, path: Path) -> None:
    self._scanning = True
    entry = scan(path)                          # runs in a background thread
    self.call_from_thread(self._push_entry, entry)  # safely updates the UI

@work(thread=True) runs the method in a thread pool.

call_from_thread() schedules the UI update back on the main thread, which is similar to root.after() in Tkinter or loop.call_soon_threadsafe() in asyncio, but with a much nicer API.

Textual's worker system is a clean abstraction over Python threads, but it's worth understanding what's happening underneath. If you want to go deeper on threading and multiprocessing in Python, when to use each and what the tradeoffs are, I wrote a full breakdown in Concurrency in Python with Threading and Multiprocessing.

Step 4 - Rendering the Table

Each row in the DataTable is built from four columns: size, a directory indicator, the entry name, and a proportional bar.

The bar is rendered using rich.text.Text objects, which Textual accepts natively:

from rich.text import Text

BAR_WIDTH = 20

def bar(ratio: float, width: int = BAR_WIDTH) -> Text:
    filled = round(ratio * width)
    empty  = width - filled
    t = Text("[", style="dim green")
    t.append("#" * filled, style="bold green")
    t.append("." * empty,  style="dim green")
    t.append("]", style="dim green")
    return t

In _refresh_table, you calculate the ratio for each entry against the largest sibling, so the biggest item always gets a full bar.

That gives you a quick "shape" of disk usage without needing absolute numbers:

max_size = current.children[0].size or 1   # children are sorted largest-first

for entry in current.children:
    ratio = entry.size / max_size
    table.add_row(
        Text(fmt_size(entry.size), justify="right", style="green"),
        Text("/", style="bold cyan") if entry.is_dir else Text(" "),
        Text(entry.name, style="bold cyan" if entry.is_dir else "default"),
        bar(ratio),
        key=str(entry.path),   # used later to look up the selected entry
    )

The key parameter is important: it lets you map from a selected table row back to the original DiskEntry without maintaining a separate index.

Step 5 - Delete with Confirmation

Deleting from a TUI needs a confirmation step.

Textual's ModalScreen is the right tool.

It overlays the current screen and returns a value via dismiss(), which is perfect for a "Yes/No" flow:

from textual.screen import ModalScreen

class ConfirmDelete(ModalScreen[bool]):
    BINDINGS = [
        Binding("y", "yes", "Yes"),
        Binding("n,escape", "no", "No"),
    ]

    def action_yes(self) -> None:
        self.dismiss(True)

    def action_no(self) -> None:
        self.dismiss(False)

In the main app, you push the modal and handle the result in a callback:

def action_delete(self) -> None:
    entry = self._selected_entry()
    if entry:
        self.push_screen(ConfirmDelete(entry), self._handle_delete_result)

def _handle_delete_result(self, confirmed: bool) -> None:
    if not confirmed:
        return
    entry = self._selected_entry()
    try:
        shutil.rmtree(entry.path) if entry.is_dir else entry.path.unlink()
    except OSError as exc:
        self.notify(f"Delete failed: {exc}", severity="error")
        return
    # Update the in-memory tree — no need for a full rescan
    current = self._current()
    current.children = [c for c in current.children if c.path != entry.path]
    current.size = sum(c.size for c in current.children)
    self._refresh_table()
    self.notify(f"Deleted {entry.name}", severity="warning")

The last part is worth highlighting: after a delete, you update the in-memory DiskEntry tree directly instead of triggering a full disk rescan.

This keeps the UI snappy and avoids re-reading a potentially large directory.

Step 6 - CLI Entrypoint with Typer

Wrapping the app in a Typer command gives you argument parsing and --help for free:

import typer

cli = typer.Typer(help="Terminal disk usage analyzer.")

@cli.command()
def main(
    path: Path = typer.Argument(
        Path("."),
        help="Directory to analyze.",
        exists=True,
        file_okay=False,
        dir_okay=True,
        resolve_path=True,
    ),
) -> None:
    DuskApp(path).run()

if __name__ == "__main__":
    cli()

Examples of usage:

python pydusk.py ~/projects
python pydusk.py /var/log
python pydusk.py          # defaults to current directory

Conclusion

In about 450 lines of Python, you built a fully functional disk usage analyzer with a keyboard-driven TUI, mouse support, background scanning, and safe deletion.

Not bad for a single file with two dependencies.

There's plenty of room to take this further. Sorting modes would be a natural next step: toggle between size and name order with a single keypress. pathspec would let you add .gitignore-aware scanning, so virtual environments and build artefacts can be excluded upfront rather than deleted manually.

A --json flag to export the tree would make pydusk useful in scripts and CI pipelines, not just interactively. And color-coding entries by file type - like executables, archives, media - would make the display more scannable at a glance.

And if you want to take it all the way, turn pydusk into a proper pip-installable tool with a pyproject.toml, a build step, and a real release on PyPI, I walked through the full process in How to Build and Publish a Python Package to PyPI.

The full source code for the project is at github.com/nunombispo/pydusk-article.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

GitHub Actions for Python Projects - Automate Your Workflow from Day One

Developer Service — Mon, 23 Mar 2026 07:38:22 +0000

You push your code. Your teammate pulls it. Nothing works. Sound familiar?

The "works on my machine" problem is one of the oldest frustrations in software development, and it doesn't go away on its own. The fix is automating your checks so every change gets validated the moment it hits your repository, not hours later when someone else is blocked.

That's CI/CD in a nutshell: run your tests, linting, and other quality checks automatically on every push or pull request, before anything breaks in production.

GitHub Actions is one of the best tools for this, especially if you're already hosting your code on GitHub. It's free for public repositories, requires zero external services, and lives right inside your project as a simple YAML file.

By the end of this tutorial, you'll have a working CI pipeline for a Python project, one that runs your tests automatically every time you push, catches issues early, and makes "it works on my machine" a thing of the past.

The full project is available on GitHub if you want to follow along.

Key Concepts

Before writing your first workflow, it helps to understand five terms that GitHub Actions uses everywhere. Once these click, the YAML syntax will make immediate sense.

Workflow: A YAML file that lives in .github/workflows/ in your repository. It defines what should happen, when it should happen, and how. You can have multiple workflows in a single project.

Event: The trigger that starts a workflow. Common examples are push (someone pushes a commit), pull_request (a PR is opened or updated), or schedule (a cron-based timer). You decide which events matter.

Job: A workflow is made up of one or more jobs. Each job runs independently on its own machine. By default, jobs run in parallel, though you can chain them if needed.

Step: The individual units inside a job. Each step is either a shell command you write (run: pytest) or a pre-built action you reference (uses: actions/checkout@v4). Steps run sequentially, top to bottom.

Runner: The virtual machine that executes a job. GitHub provides hosted runners for Linux, Windows, and macOS. For most Python projects, ubuntu-latest is the go-to choice.

Here's how they fit together:

Event (push)
  └── Workflow (ci.yml)
        └── Job (test)
              ├── Step: checkout code
              ├── Step: set up Python
              ├── Step: install dependencies
              └── Step: run pytest

Think of it as a chain reaction: an event fires, the workflow wakes up, jobs are assigned to runners, and steps execute one by one.

Your First Workflow: Running Tests on Every Push

Theory is useful, but nothing beats seeing it work. Let's build a minimal Python project and wire up a GitHub Actions workflow that runs your tests automatically on every push.

The Python Project

Start with a simple project structure:

my-project/
├── .github/
│   └── workflows/
│       └── ci.yml
├── calculator.py
├── test_calculator.py
└── requirements.txt

calculator.py contains a function to test:

def add(a, b):
    return a + b

test_calculator.py has the test:

from calculator import add

def test_add():
    assert add(2, 3) == 5

And requirements.txt lists your test dependency:

pytest

That's it. A real project is larger, but the workflow you're about to write scales to any size.

Creating the Workflow File

Create the file .github/workflows/ci.yml in your repository. The folder structure matters, GitHub only picks up workflows from that exact path.

Here's the complete workflow:

name: CI

on:
  push:
    branches:
      - main
  pull_request:
    branches:
      - main

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run tests
        run: pytest

Line-by-Line Walkthrough

name: CI - The display name for this workflow. It shows up in the GitHub UI under the Actions tab.

on: - Defines what triggers the workflow. In this case, it runs on every push or pull_request targeting the main branch. If your default branch is master, update this accordingly.

jobs: - Everything underneath this key defines the jobs to run. Here there's just one: test.

runs-on: ubuntu-latest - Tells GitHub to spin up a fresh Ubuntu virtual machine for this job. It's fast, free, and the most common choice for Python projects.

uses: actions/checkout@v4 - This is a pre-built action maintained by GitHub. It clones your repository onto the runner so the following steps have access to your code. Without this, the runner would have an empty machine.

uses: actions/setup-python@v5 - Another official action. It installs the specified Python version on the runner. The with: python-version: "3.12" block lets you pin the exact version you want.

run: pip install -r requirements.txt - A plain shell command. It installs your project's dependencies, in this case just pytest.

run: pytest - Runs your test suite. If any test fails, this step exits with a non-zero code, which marks the entire job as failed.

Pushing and Reading the Results

Commit the workflow file and push it to main:

git add .
git commit -m "Add CI workflow"
git push origin main

Head to your repository on GitHub and click the Actions tab. You'll see your workflow listed by name. Click into it to find the individual run, then drill into the test job to see each step expand with its output.

A green checkmark means everything passed. A red cross means something failed, click the failing step to read the exact error output, the same as you'd see in your local terminal.

From this point on, every push to main triggers the workflow automatically. You don't have to think about it, GitHub handles it for you.

You can find the complete project for this tutorial on GitHub: https://github.com/nunombispo/github-actions-article

Making It Smarter

A workflow that runs pytest is a great start. But with a few additions, you can catch more issues earlier and make your pipeline noticeably faster. Here's how to level up your CI without overcomplicating it.

Linting with `ruff`

Linting checks your code for style issues, unused imports, and common mistakes, the kind of things that slow down code review when left unchecked.

ruff is a fast Python linter written in Rust, and it's become the go-to choice for modern Python projects. Add it to your requirements.txt:

pytest
ruff

Then add a linting step to your workflow, right before pytest:

      - name: Lint with ruff
        run: ruff check .

That's all it takes.

If ruff finds any issues, the step fails and the rest of the job stops. Fix the warnings locally with ruff check --fix . before pushing, and your pipeline stays green.

Caching `pip` Dependencies

Every time your workflow runs, the runner starts from a clean machine and reinstalls your dependencies from scratch. For small projects this is fine, but as your requirements.txt grows, that pip install step gets slower.

The fix is caching. GitHub Actions lets you save the result of a step between runs and restore it the next time the same dependencies are needed. Add this step right before your install step:

      - name: Cache pip dependencies
        uses: actions/cache@v4
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
          restore-keys: |
            ${{ runner.os }}-pip-

Here's what each part does:

path - the folder to cache, which is where pip stores downloaded packages.
key - a unique identifier for this cache. It includes a hash of requirements.txt, so the cache is automatically invalidated whenever your dependencies change.
restore-keys - a fallback key used when an exact match isn't found. It restores the closest available cache, which is still faster than starting from zero.

On the first run, the cache is created.

On every subsequent run with the same dependencies, pip install skips downloading packages it already has. For larger projects, this can cut your install time significantly.

Testing Across Multiple Python Versions

Your users might not all be running the same Python version as you. A matrix strategy lets you run the same job against multiple versions in parallel, with almost no extra effort.

Replace the runs-on and python-version lines in your job with this:

  test:
    runs-on: ubuntu-latest

    strategy:
      matrix:
        python-version: ["3.10", "3.11", "3.12"]

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}

GitHub Actions expands the matrix and runs one job per version, all in parallel. The ${{ matrix.python-version }} syntax injects the current version into each job automatically.

The result: three jobs running simultaneously, each reporting its own pass or fail. If your code breaks on Python 3.10 but works on 3.12, you'll know immediately.

Type Checking with `mypy` (Optional)

If your project uses type annotations, mypy can verify them statically, catching bugs that tests might miss, like passing a string where an integer is expected.

Add it to requirements.txt:

pytest
ruff
mypy

And add a step after linting:

      - name: Type check with mypy
        run: mypy .

For projects without type annotations yet, skip this for now.

But if you're already annotating your functions, running mypy in CI is a low-effort way to keep your type hints honest as the codebase grows.

Common Pitfalls

GitHub Actions is beginner-friendly, but a few mistakes come up repeatedly. Here's what to watch out for.

YAML Indentation Errors

YAML is whitespace-sensitive, and a single misplaced space can break your entire workflow. GitHub will report a syntax error in the Actions tab, but the message isn't always obvious about where the problem is.

The safest habit is to use a YAML validator before pushing. The YAML Lint website works well for quick checks. If you're using VS Code, the YAML extension by Red Hat highlights indentation issues inline as you type. Always use spaces, never tabs - YAML doesn't allow tabs.

Wrong Branch Name or Event Trigger

If your workflow isn't running at all, the most common culprit is a mismatch between the branch name in your on: block and your actual default branch. Many older repositories use master while newer ones default to main. Double-check which one you're pushing to and update the workflow to match.

Forgetting `requirements.txt`

The runner starts with a clean Python installation. If a dependency isn't listed in requirements.txt, or if the file doesn't exist, the install step will either fail or silently skip packages your tests depend on. Make sure every dependency your tests need is listed, including pytest itself.

Hardcoding Credentials

Never put API keys, passwords, or tokens directly in your workflow file. The file lives in your repository, which means anyone with read access can see them.

Use GitHub Secrets instead. Go to your repository Settings → Secrets and variables → Actions, add your secret there, and reference it in your workflow like this:

      - name: Deploy
        env:
          API_KEY: ${{ secrets.API_KEY }}
        run: ./deploy.sh

The value is masked in logs and never exposed in plain text.

Conclusion

One YAML file is all it takes to stop testing manually and start catching issues automatically.

With GitHub Actions, every push to your Python project triggers your tests, linting, and type checks, no extra tools, no configuration overhead.

The best time to add CI to your project is before something breaks.

Push your first workflow today, your future self (and your teammates) will thank you.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Build an F1 Pit Wall Display with ESP32 CYD and OpenF1 API

Developer Service — Mon, 16 Mar 2026 07:42:00 +0000

The 2026 Australian GP is live. Russell and Leclerc are battling it out, and you want the data on your desk, not a browser tab, not your phone. Something physical that just sits there. You have an ESP32 CYD in the parts drawer: 320×240 touchscreen, built-in Wi-Fi, €10. OpenF1 is free and has real F1 timing data.

This sounds like a fun Sunday afternoon project. It wasn't.

The first problem hits immediately. OpenF1 doesn't return race state, it returns race history. Hit /position and you get thousands of chronological entries, one per position change, every driver, entire session. Hit /laps and you get the same for lap times. /stints for tyres. /intervals for gaps. Six endpoints, all flat arrays, none of them joined, none of them telling you what's happening right now.

The second problem is the ESP32 itself. 320KB of RAM. Fetching and aggregating six endpoints of session data on a microcontroller isn't a performance concern, it's a category error. Micro Python's urequests will run out of memory before it finishes parsing the response.

So the fun Sunday afternoon project becomes an exercise of a proper two-layer architecture: a Python aggregator on a local server that reconstructs race state from OpenF1's raw streams and exposes a single clean endpoint, and a Micro Python client that makes one HTTP request, gets back exactly what it needs, and renders it on a $10 screen sitting next to the keyboard.

Here's how it's built. Full source for the API and CYD display:

nunombispo / CYD-OpenF1

CYD F1 Pit Wall Display

MicroPython client for the ESP32-2432S028 (CYD) — 320×240 ILI9341 touchscreen, WiFi — that shows live F1 standings by polling a single race-status API. No OpenF1 calls from the device; one HTTP request, parse, render.

Hardware

Component	Details
Board	ESP32-2432S028 (CYD)
Display	ILI9341 320×240, RGB565
Touch	XPT2046 (handled by `cyd` helper)
Connectivity	WiFi (built-in)

Dependencies

MicroPython (ESP32 build with WiFi)
cyd — display and touch helper for this board (ILI9341 + XPT2046)
urequests, ujson — typically included in MicroPython

Configuration

Edit the top of main.py:

Variable	Description
`WIFI_SSID`	Your WiFi network name
`WIFI_PASSWORD`	WiFi password
`PITWALL_URL`	Full URL of the race-status API, e.g. `http://192.168.1.100:8000/api/race-status`
`MY_DRIVER`	3-letter driver code to highlight (e.g. `"NOR"`, `"VER"`)
`POLL_INTERVAL`	Seconds between API polls (default `30`)

The CYD must be on the same network as the machine running the API. Use the host’s LAN IP (not…

View on GitHub

One Endpoint, One Request

The first thing to get right is the boundary between server and device.

OpenF1 is the source: free, no API key, real timing data during sessions. Do note that access to live session data requires a paid API key, historical data access is free.

The catch is the shape of that data, the API is a log, not a state machine. It records everything that happened; figuring out what's happening now is your job. So we move that job off the device.

The fix is straightforward once you accept that the ESP32 is a display device, not a data processor.

You move the aggregation off the microcontroller entirely and onto anything with more headroom, like a Raspberry Pi, a spare machine, a Docker container, whatever is already running on your network. That server fetches all six OpenF1 endpoints, reconstructs current race state, and exposes a single endpoint: /api/race-status.

The ESP32 makes one HTTP request. It gets back one JSON response. That response contains exactly what the display needs, positions in order, driver codes, tyre compounds, tyre age in laps, gap to leader, fastest lap, and nothing else. No filtering, no aggregation, no memory pressure.

This is the core architectural decision of the whole build, and it applies well beyond OpenF1.

Any time you're hitting a complex external API from a microcontroller, like weather, sports scores, home automation, or any dashboard, the instinct is to do as much as possible on the device. Resist it.

Microcontrollers are good at talking to hardware; a Python server on your network is good at talking to APIs. Put each where it belongs and the problem gets simpler on both ends. The same aggregator can then serve other clients too: a second display, an e-ink board, a different form factor. One server, many endpoints, many devices.

The split also has a practical benefit: when OpenF1 changes an endpoint or returns unexpected data, you fix it in one file on the server. The display doesn't care where the data came from, only that it arrived in the right shape.

Part 1: The Aggregator

The server's job: turn six OpenF1 streams into one snapshot the display can consume without thinking. The aggregator lives in a race_status.py, in my case served as an API endpoint on a Raspberry Pi.

Positions

/position returns every position change for every driver, thousands of entries. You need the most recent entry per driver, not the last item in the array. Group by driver_number, take the entry with the highest date per group:

def _latest_position_per_driver(positions: list[dict]) -> dict[int, int]:
    """Map driver_number -> position using the most recent update per driver."""
    by_driver: dict[int, list[dict]] = {}
    for p in positions:
        if p.get("date") is None:
            continue
        dn = p["driver_number"]
        if dn not in by_driver:
            by_driver[dn] = []
        by_driver[dn].append(p)
    out = {}
    for dn, updates in by_driver.items():
        if not updates:
            continue
        latest = max(updates, key=lambda x: x["date"])
        out[dn] = latest["position"]
    return out

That pattern, group by entity, take latest, is the same for positions, intervals, and stints. The result is a {driver_number: position} mapping you can sort and iterate.

Tyres

/stints gives compound and lap_start, not tyre age. Sort each driver's stints by lap_start descending, find the stint that covers the current lap (lap_start ≤ now and no lap_end or lap_end ≥ now), then tyre_age = current_lap - lap_start + 1.

Gaps

First choice: /intervals and its gap_to_leader, latest per driver as above. If the endpoint is empty (e.g. early in the session), fall back to lap-time delta: leader's last lap time minus driver's. When a driver is lapped, OpenF1 returns "+1 LAP" as a string, check the type and pass through as-is instead of formatting as seconds.

Formatting and total laps

Lap times and gaps are formatted on the server (e.g. ≥60s → M:SS.mmm, else +N.NNNs) so the display never does number formatting. OpenF1 doesn't expose scheduled race distance; the aggregator leaves total_laps as None and the display shows LAP 24 without a denominator.

The Response

All of that feeds into a single get_race_status() that returns this shape:

{
  "session": {
    "circuit_name": "Melbourne",
    "country_iso3": "AUS"
  },
  "latest_lap_number": 58,
  "total_laps": null,
  "current_standings": [
    {
      "position": 1,
      "driver_number": 63,
      "name": "George RUSSELL",
      "name_acronym": "RUS",
      "team_colour": "00D7B6",
      "tyre": "HARD",
      "stint_duration_laps": 47,
      "lap_time_seconds": 83.351,
      "lap_time": "1:23.351",
      "lap_time_or_gap": "1:23.351"
    },
    {
      "position": 2,
      "driver_number": 12,
      "name": "Kimi ANTONELLI",
      "name_acronym": "ANT",
      "team_colour": "00D7B6",
      "tyre": "HARD",
      "stint_duration_laps": 47,
      "lap_time_seconds": 82.653,
      "lap_time": "1:22.653",
      "lap_time_or_gap": "+2.974s"
    },
    {
      "position": 3,
      "driver_number": 16,
      "name": "Charles LECLERC",
      "name_acronym": "LEC",
      "team_colour": "ED1131",
      "tyre": "HARD",
      "stint_duration_laps": 34,
      "lap_time_seconds": 83.317,
      "lap_time": "1:23.317",
      "lap_time_or_gap": "+15.519s"
    }
    ...
  ],
  "fastest_lap": {
    "driver_number": 3,
    "lap_number": 42,
    "duration_seconds": 82.091,
    "duration_formatted": "1:22.091",
    "driver_name": "Max VERSTAPPEN",
    "driver_name_acronym": "VER",
    "team_colour": "4781D7"
  }
}

Flat. Ordered by position. Every field the display needs, nothing it doesn't. That's the contract, the ESP32 never sees a raw OpenF1 response.

Part 2: The Display

The part you actually look at. The CYD runs Micro Python with an ILI9341: fill_rectangle, fill_circle, draw_text8x8.

The font is fixed 8×8, no sizing, no wrapping. Every element is placed by hand with explicit coordinates. At 320×240, every pixel is intentional. That's the design challenge.

Layout

Top bar (22px): circuit, lap, SC badge when active
Bottom bar (20px): fastest lap in purple, lap number
Rows (198px): 19px per row → 10 drivers on screen (top 10)

Column x-coordinates are constants; team colours from the API are hex, so convert once to RGB565 per driver and cache it, the ILI9341 needs 16-bit values.

# ── DISPLAY RENDERING ─────────────────────────────────────
# Uses ili9341: fill_rectangle, fill_circle, draw_text8x8 (8x8 built-in font)

ROW_H    = 19
TOP_BAR  = 22
BOT_BAR  = 20
MAX_ROWS = 10  # (240 - 22 - 20) / 19 = 10 rows, order by position

# Column x positions (more spacing between position, driver #, strip, name, tyre, stint, lap time, gap)
COL_POS = 5
COL_DRIVER_NUM = 30
COL_STRIP = 50
COL_CODE = 60
COL_TYRE = 105
COL_STINT = 125
COL_LAP_TIME = 160
COL_GAP = 250
TOP_BAR_LAP_X = 265

Every element is placed by these constants, like position, driver number, team strip, code, tyre circle, stint laps, lap time and gap. Change one and you touch every draw_text8x8 and fill_rectangle that uses it.

Rendering

Top bar: red strip, circuit left, lap right, yellow SC badge when the latest /race_control message says safety car.
Driver rows: alternating black/dark grey background; team colour on number and 3px strip; driver code; tyre as circle with compound initial (S/M/H/I); stint laps; lap time or gap (P1 gets their lap time, everyone else gets lap_time_or_gap from the aggregator, including "+1 LAP" for lapped drivers).
Bottom bar: FL driver and time in purple, lap number in grey.

The 8×8 font caps line length (e.g. 9 chars for the gap column); the aggregator already sends formatted strings so the client never does number formatting.

Mapping the response: parse_pitwall

The renderer expects four dicts keyed by driver number, positions, drivers, laps, stints, because each draw function takes a driver key and looks up what it needs.

The API returns a single list of standings rows. parse_pitwall() is the bridge: one pass over current_standings, fan out each row into the four dicts, add session metadata and fastest-lap fields for the bars.

If the aggregator changes its response shape, you update this one function and the render calls stay unchanged.

def parse_pitwall(data):
    """Map pitwall JSON to (session, positions, drivers, laps, stints, fastest_driver, fastest_time, current_lap, total_laps, updated).
    positions/drivers/laps/stints are dicts keyed by driver_number (str); order preserved via positions insertion order."""
    if not data or "current_standings" not in data:
        return None
    sess = data.get("session") or {}
    session = {
        "circuit_short_name": sess.get("circuit_name", "???"),
        "country_name": sess.get("country_iso3", "???"),
        "total_laps": data.get("total_laps"),
    }
    current_lap = data.get("latest_lap_number") or 0
    total_laps = data.get("total_laps")
    standings = data.get("current_standings") or []
    positions = {}
    drivers = {}
    laps = {}
    stints = {}
    for row in standings:
        drv = str(row.get("driver_number", ""))
        if not drv:
            continue
        positions[drv] = {
            "position": row.get("position", 99),
            "lap_time_or_gap": row.get("lap_time_or_gap"),
        }
        drivers[drv] = {
            "name_acronym": row.get("name_acronym") or "???",
            "team_name": "",
            "team_colour": row.get("team_colour"),
        }
        laps[drv] = {"lap_duration": row.get("lap_time_seconds")}  # seconds for format_laptime
        tyre = row.get("tyre")
        stints[drv] = {
            "compound": tyre if tyre else "",
            "stint_duration_laps": row.get("stint_duration_laps"),
        }
    fl = data.get("fastest_lap") or {}
    fastest_driver = fl.get("driver_name_acronym") or ""
    fastest_time = fl.get("duration_seconds")
    fastest_lap_number = fl.get("lap_number")
    updated = ""
    return (session, positions, drivers, laps, stints, fastest_driver, fastest_time, fastest_lap_number, current_lap, total_laps, updated)

Part 3: The Main Loop

The main loop is the smallest part of the system, on purpose.

Setup: display and touch via the cyd helper, tap the screen and it sets a flag for an immediate refresh, no waiting for the next poll.

Wi-Fi next, show "Connecting..." on screen, then clear to black. If connection fails, there's nothing to show.

The loop is poll, parse, redraw only when needed:

# ── MAIN LOOP ─────────────────────────────────────────────

def main():
    global touch_refresh
    # Display + touch via cyd (touch handler sets touch_refresh for manual refresh)
    display, backlight, touch = cyd.display_setup(320, 240, rotation=90, touch_handler=_on_touch)
    cyd.set_backlight_brightness(backlight, 90)

    cyd.display_loading(display, "Connecting...", RED)
    if not cyd.connect_wifi(WIFI_SSID, WIFI_PASSWORD):
        return

    display.clear(BLACK)
    time.sleep(1)
    gc.collect()

    last_lap = -1
    while True:
        data = get_pitwall()
        parsed = parse_pitwall(data) if data else None
        if parsed:
            (session, positions, drivers, laps, stints,
             fastest_driver, fastest_time, fastest_lap_number, current_lap, total_laps, updated) = parsed
            if session.get("total_laps") is None and total_laps is not None:
                session["total_laps"] = total_laps
            if current_lap != last_lap or touch_refresh:
                render_frame(display, session, positions, drivers, laps, stints, None,
                             current_lap=current_lap, fastest_driver=fastest_driver,
                             fastest_time=fastest_time, fastest_lap_number=fastest_lap_number)
                last_lap = current_lap
                touch_refresh = False
        time.sleep(POLL_INTERVAL)

Triggers: new lap or touch. Same lap, failed request, or parse error → no redraw. Full-screen redraw over SPI is costly; do it only when something changed.

Memory. MicroPython doesn't run the garbage collector in the background like CPython.

urequests allocates a response buffer on every HTTP call; if you don't release it explicitly, that memory stays allocated. Over a two-hour race, dozens of poll cycles, the heap eventually runs out and the device crashes.

The discipline in get_pitwall(): collect before the request to maximise free heap; parse the response; close the response and delete the reference so the buffer can be reclaimed; collect again to actually free it.

def get_pitwall():
    """GET pitwall endpoint, return parsed JSON or None."""
    print("Pitwall URL:", PITWALL_URL)
    try:
        gc.collect()
        r = urequests.request("GET", PITWALL_URL, timeout=10)
        data = ujson.loads(r.content)
        r.close()
        del r
        gc.collect()
        return data
    except OSError as e:
        print("Pitwall OSError:", e, type(e).__name__)
        return None
    except Exception as e:
        print("Pitwall error:", e, type(e).__name__)
        return None

Without that sequence, the device doesn't make it to the chequered flag.

Result

After the 2026 Australian GP, the display on the desk looks like this:

Ten drivers, tyre compounds, gaps, fastest lap in purple at the bottom. Norris's row is highlighted in gold because that's MY_DRIVER in the config. On a live session, a new lap crosses the line and within a few seconds the screen updates; tap the panel and it refreshes on demand.

What works well

Lap-change trigger: Redrawing only on a new lap means no flicker mid-corner; the 30-second poll is short enough that you never miss an update. Touch-to-refresh covers the "just powered on mid-race" case.
Team colours: OpenF1's colours are accurate, tyre circles and driver strips are readable at a glance; you can identify a row by colour before reading the name.
Wrapper API in practice: The aggregator handled all OpenF1 complexity invisibly; the ESP32 never dropped a request or ran out of memory across a full two-hour session.

What doesn't

total_laps: OpenF1 doesn't expose scheduled race distance. The top bar shows LAP 24 with no denominator. Workaround: hardcoded laps-per-circuit lookup by circuit_short_name, inelegant but functional.
Gap accuracy early on: When /intervals is slow to populate (first few laps), the fallback to lap-time delta is rough, similar lap times can show near-zero gaps that don't reflect track position. It should correct itself after a couple of laps.
8×8 font: Readable but unforgiving. Nine characters per column max; names are always three-letter codes. Fine for F1 (NOR, RUS, LEC); worth knowing if you adapt the layout for another sport.

What's Next

The display works. The important part: extending it doesn't mean rewiring the whole system, the aggregator-display boundary is the enabler. New behaviour is either a new endpoint on the server, a new view on the client, or both.

Three directions that fit that pattern:

Red flag and VSC detection — /race_control carries safety car, VSC, red flag, track clear. The code currently only checks for safety car. Add string checks for VSC (e.g. different badge) and red flag (full red overlay with RED FLAG in white; more useful than stale lap times when the race is paused).

Swipe between views — Touch currently triggers a refresh. Better: swipe left/right for standings vs championship points vs session schedule. Add /api/championship from OpenF1's /drivers and /team_results; the display is a second render function behind a view index the touch handler increments.

E-ink version — The CYD is right for live race (colour, refresh, lap-by-lap). Between weekends it stays off. An e-ink panel on another ESP32 on a small battery: always-on championship points, next race, countdown; power only when it updates (e.g. once a week). Same aggregator, different endpoint, simpler response shape.

Conclusion

It wasn't a fun Sunday afternoon, the build was harder than that. Not because the hardware was difficult, but because the data was.

OpenF1 is genuinely useful and leaves all the reconstruction work to the client. The aggregator-display split solved it: one Python server turns six raw endpoints into one clean API; the client does one request and renders.

That boundary is what makes expansion straightforward, new endpoints (championship, schedule), new clients (e-ink, second screen), or new views (swipe between standings and points) without tearing the system apart.

The CYD earned its place: cheap, capable, physically there next to the keyboard, updating every lap. Source on GitHub (link above).

The 2026 season is 24 races; plenty of time to add the extensions.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

How to Build and Publish a Python Package to PyPI (With a Real Project)

Developer Service — Mon, 09 Mar 2026 07:32:47 +0000

You've written a useful Python utility, a helper for parsing files, a small data tool, or a class you keep copying between projects. At some point you think: I wish I could just pip install this. You can, and it's simpler than it looks.

Packaging your code and publishing it to PyPI (the Python Package Index) forces you to treat your code as a product: clear interfaces, versioning, and documentation others can follow. A published package is also a credibility signal, you didn't just write code, you shipped it.

In this tutorial we'll build tinyfiledb, a lightweight file-based database that stores, retrieves, updates, and deletes records in a local JSON file. No server, no migrations, no dependencies. It's small by design so we can focus on the packaging workflow.

By the end you will have:

A properly structured Python package
A pyproject.toml configured and ready
A package built and published to PyPI
A working pip install tinyfiledb you can share

The Project: A Tiny File Database

tinyfiledb is a minimalist file-based database: no server, no ORM. It stores records as JSON and exposes a simple Python API:

insert(record) - add a record, get back an auto-generated ID
get(id) - retrieve one record by ID
all() - return every record
update(id, data) - merge data into an existing record
delete(id) - remove a record

Project Structure

This is the project structure you will use:

tinyfiledb/
├── tinyfiledb/
│   ├── __init__.py
│   └── db.py
├── pyproject.toml
└── README.md

`tinyfiledb/db.py`

The main project file that contains the tiny database implementation:

import json
import uuid
from pathlib import Path


class FileDB:
    def __init__(self, filepath: str = "db.json"):
        self.path = Path(filepath)
        if not self.path.exists():
            self._write({})

    def _read(self) -> dict:
        with open(self.path, "r") as f:
            return json.load(f)

    def _write(self, data: dict):
        with open(self.path, "w") as f:
            json.dump(data, f, indent=2)

    def insert(self, record: dict) -> str:
        data = self._read()
        record_id = str(uuid.uuid4())[:8]
        data[record_id] = record
        self._write(data)
        return record_id

    def get(self, record_id: str) -> dict | None:
        return self._read().get(record_id)

    def all(self) -> dict:
        return self._read()

    def update(self, record_id: str, new_data: dict) -> bool:
        data = self._read()
        if record_id not in data:
            return False
        data[record_id].update(new_data)
        self._write(data)
        return True

    def delete(self, record_id: str) -> bool:
        data = self._read()
        if record_id not in data:
            return False
        del data[record_id]
        self._write(data)
        return True

Supports full CRUD operations - insert() returns an auto-generated 8-character UUID key, get() and all() read from disk on every call, and update() merges new fields into an existing record rather than replacing it. The file is created automatically on initialization if it doesn't exist.

`tinyfiledb/init.py`

Expose only what users need:

from .db import FileDB

__all__ = ["FileDB"]

Quick Usage

You can test it with this quick usage example:

from tinyfiledb import FileDB

db = FileDB("mydata.json")
user_id = db.insert({"name": "Alice", "role": "admin"})
print(db.get(user_id))   # {'name': 'Alice', 'role': 'admin'}
db.update(user_id, {"role": "superadmin"})
print(db.all())
db.delete(user_id)

Should return:

{'name': 'Alice', 'role': 'admin'}
{'daf0f9fb': {'name': 'Alice', 'role': 'superadmin'}}

Roughly 50 lines of code - enough to be useful, small enough to keep the focus on packaging.

Structuring Your Package

Your project must be laid out so Python's tooling can find and install the package. There are two common layouts:

src/ layout - package lives under src/tinyfiledb/. Helps avoid importing the local uninstalled copy during tests.
Flat layout - package sits directly in the project root as tinyfiledb/. Simpler and what we'll use.

The project uses the flat layout:

tinyfiledb/
├── tinyfiledb/
│   ├── __init__.py
│   └── db.py
├── pyproject.toml
└── README.md

Your __init__.py is the package's public face.

Keep it minimal: re-export the main API and set __all__ so the public surface is explicit.

We already did that above.

Writing `pyproject.toml`

setup.py required running Python just to read metadata. pyproject.toml is a static config file, no code execution. Every modern build tool (Hatchling, Flit, setuptools) and PyPI expect it.

For new packages, it's the only config you need. Create pyproject.toml in the project root:

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "tinyfiledb"
version = "0.1.0"
description = "A lightweight file-based database for Python projects."
readme = "README.md"
license = { text = "MIT" }
authors = [
  { name = "Your Name", email = "you@example.com" }
]
requires-python = ">=3.10"
dependencies = []

What each part does:

[build-system] - Tells Python how to build the package. Hatchling is fast and needs no extra config for a simple project.
name - What users type in pip install. Must be unique on PyPI; use hyphens by convention.
version - Semantic versioning; start at 0.1.0.
description - One-line summary on PyPI; keep it short.
readme - PyPI uses this for the project description.
license - e.g. { text = "MIT" } or reference a LICENSE file.
authors - Name/email; shown on PyPI.
requires-python - We use dict | None in code, so >=3.10.
dependencies - Runtime dependencies; empty for tinyfiledb.

Building the Distribution

A distribution is a versioned snapshot of your package that can be uploaded and installed. Python uses two formats; you'll produce both.

Install the build tool

pip install build

Run the build

From the directory that contains pyproject.toml:

python -m build

You'll get:

* Creating isolated environment: venv+pip...
* Installing packages in isolated environment:
  - hatchling
* Getting build dependencies for sdist...
* Building sdist...
* Building wheel from sdist
* Creating isolated environment: venv+pip...
* Installing packages in isolated environment:
  - hatchling
* Getting build dependencies for wheel...
* Building wheel...
Successfully built tinyfiledb-0.1.0.tar.gz and tinyfiledb-0.1.0-py3-none-any.whl

A dist/ folder will appear:

.tar.gz - Source distribution (sdist): raw source + metadata. Pip can build from it when no wheel is available.
.whl - Wheel: pre-built; pip installs it directly with no build step.

Publishing to PyPI

Use TestPyPI first.

PyPI does not allow re-uploading the same version. If you push 0.1.0 with a mistake, you must release 0.1.1.

TestPyPI is a sandbox - upload and fix without burning versions.

Install twine (upload tool):

pip install twine

TestPyPI

First make sure your account and token are ready:

Account - test.pypi.org/account/register. Verify your email.
2FA - PyPI requires two-factor authentication for uploads. Enable it under Account Settings before creating a token.
API token - Account Settings → API Tokens → Add. Name it (e.g. tinyfiledb-upload), copy the token (starts with pypi-); you won't see it again.

Upload:

twine upload --repository testpypi dist/*

When prompted: username __token__, password = your token (literally __token__, not a placeholder).

Uploading distributions to https://test.pypi.org/legacy/
WARNING  This environment is not supported for trusted publishing
Enter your API token: 
Uploading tinyfiledb-0.1.0-py3-none-any.whl
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 4.8/4.8 kB • 00:01 • ?
Uploading tinyfiledb-0.1.0.tar.gz
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 4.4/4.4 kB • 00:00 • ?

View at:
https://test.pypi.org/project/tinyfiledb/0.1.0/

Test install in a fresh venv:

python -m venv test-env
# Windows: test-env\Scripts\activate
# macOS/Linux: source test-env/bin/activate

pip install --index-url https://test.pypi.org/simple/ tinyfiledb

Then:

python -c "from tinyfiledb import FileDB; db = FileDB('t.json'); print(db.insert({'x': 1}))"

If that works, you're ready for PyPI.

PyPI (production)

Make sure your account and token are ready:

Account - pypi.org/account/register. Verify email.
2FA - PyPI requires two-factor authentication for uploads. Enable it under Account Settings before creating a token.
API token - Account Settings → API Tokens → Add. Use a project-scoped token after the first upload if you prefer.

Upload:

twine upload dist/*

Username: __token__, password: your PyPI token. After a successful upload you'll get a project URL. Your package is live.

Uploading distributions to https://upload.pypi.org/legacy/
WARNING  This environment is not supported for trusted publishing
Enter your API token: 
Uploading tinyfiledb-0.1.0-py3-none-any.whl
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 4.8/4.8 kB • 00:00 • ?
Uploading tinyfiledb-0.1.0.tar.gz
100% ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 4.4/4.4 kB • 00:00 • ?

View at:
https://pypi.org/project/tinyfiledb/0.1.0/

Anyone can install it with:

pip install tinyfiledb

Installing and Using Your Package

Verify in a clean environment (no local copy of the code):

mkdir demo && cd demo
python -m venv env
# Windows: env\Scripts\activate
# macOS/Linux: source env/bin/activate

pip install tinyfiledb

Check metadata:

pip show tinyfiledb

Will show:

Name: tinyfiledb
Version: 0.1.0
Summary: A lightweight file-based database for Python projects.
Home-page:
Author:
Author-email: Your Name <you@example.com>
License: MIT
Location: D:\GitHub\tinyfiledb\demo\env\Lib\site-packages
Requires:
Required-by:

You can do a quick test:

from tinyfiledb import FileDB

db = FileDB("tasks.json")
id1 = db.insert({"title": "Write blog post", "done": False})
id2 = db.insert({"title": "Publish to PyPI", "done": True})
print(db.get(id1))
print(db.all())
db.update(id1, {"done": True})
db.delete(id2)

Which returns:

{'title': 'Write blog post', 'done': False}
{'4d62fb2a': {'title': 'Write blog post', 'done': False}, '0f43c5be': {'title': 'Publish to PyPI', 'done': True}}

After running, open tasks.json - you'll see human-readable JSON:

{
  "4d62fb2a": {
    "title": "Write blog post",
    "done": true
  }
}

That's the whole flow: from a folder on your machine to a package anyone can install with one command.

Full source code available at:

nunombispo / tinyfiledb

tinyfiledb

A lightweight file-based database for Python projects. No server, no migrations, no dependencies — just a local JSON file and a simple API.

If this project was useful to you, consider donating to support the Developer Service Blog: https://buy.stripe.com/bIYdTrggi5lZamkdQW

Install

pip install tinyfiledb

Usage

from tinyfiledb import FileDB

db = FileDB("mydata.json")

# Insert a record
user_id = db.insert({"name": "Alice", "role": "admin"})

# Get, update, list, delete
print(db.get(user_id))
db.update(user_id, {"role": "superadmin"})
print(db.all())
db.delete(user_id)

API

insert(record) — add a record, get back an auto-generated ID
get(id) — retrieve one record by ID (returns None if not found)
all() — return every record as a dict
update(id, data) — merge data into an existing record; returns…

View on GitHub

Conclusion

You built a real Python package: a small file-based database with a clear API, proper layout, and pyproject.toml.

You built distributions with python -m build, tested on TestPyPI with twine, then published to PyPI. The same workflow applies to any project you want to share - CLI tools, utilities, libraries.

The best first package is often something you've already written, a module you've copied between projects or a script you've shared. Add a pyproject.toml, a README, and a few minutes of setup, and it can be installable by anyone.

The ecosystem is built on packages that started that way. Yours doesn't need to be big; it just needs to exist.

Next steps worth exploring: semantic versioning, GitHub Actions for releases, pytest for tests, and project-scoped API tokens on PyPI.

Want to Go Deeper?

This article covers the full publish workflow, but there's a lot more ground to cover, structuring larger packages, writing a test suite, automating releases, handling versioning properly, and building a package others actually want to use.

If you want all of that in one place, pip install yours is the companion booklet to this tutorial. It picks up exactly where this article ends and walks you through everything it takes to build, maintain, and grow a Python package worth installing.

Now go ship something.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

How to Hash Passwords in Python and Encrypt Sensitive Data the Right Way

Developer Service — Mon, 02 Mar 2026 08:58:29 +0000

You’re building your first serious application, a chat app, a password manager, maybe even an e-commerce platform. Everything looks solid until someone asks:

“How are user passwords stored?”

That’s when it hits you: plain text.

It’s a mistake many developers make early on. But here’s the good news: Python makes secure password handling surprisingly straightforward.

In this guide, you’ll learn how to implement cryptography correctly and avoid one of the most common (and dangerous) beginner mistakes.

Why This Matters to You

Before we dive into the code, let’s talk about why cryptography matters to you as a developer.

Every day, your applications handle sensitive data, passwords, API keys, personal details, possible payment information. Without proper encryption, that data is essentially left unprotected.

And the stakes are high. A data breach isn’t just a PR nightmare; it can mean broken user trust, legal consequences, regulatory fines, and real harm to real people. In 2025, the average cost of a data breach reached $4.4 million. But beyond the numbers, there’s a more important reality: users are trusting you with their private information.

The good news? You don’t need a PhD in mathematics to handle cryptography correctly. You just need to know which tools to use and when to use them. Python’s cryptography ecosystem makes strong, modern security accessible to developers at any experience level.

Getting Started

The cryptography library is your go-to tool for secure cryptographic operations in Python.

It’s actively maintained by security professionals, widely trusted in production systems, and most importantly, designed to be difficult to misuse.

pip install cryptography

With it installed, let’s take a look at what you can actually build.

Hashing Passwords: Your First Line of Defense

Let's start with the most common use case: storing passwords. You should never store passwords in plain text.

Here's what you need to understand: hashing is a one-way function. You can turn a password into a hash, but you can't reverse it back. When a user logs in, you hash their input and compare it to the stored hash. If they match, the password is correct. This means even if someone steals your database, they can't get the actual passwords.

But not all hashing is created equal. You need PBKDF2, bcrypt, or Argon2, algorithms specifically designed for passwords.

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os

def hash_password(password: str) -> tuple[bytes, bytes]:
    salt = os.urandom(16)  # Random salt for uniqueness

    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=32,
        salt=salt,
        iterations=480000,  # OWASP recommended minimum
    )

    hashed = kdf.derive(password.encode())
    return salt, hashed

# Store both salt and hash with the user record
salt, hashed = hash_password("MySecurePassword123!")
print(f"Salt: {salt.hex()}")
print(f"Hashed: {hashed.hex()}")

Here's what makes this secure: First, the salt ensures that even if two users have identical passwords, their hashes will be completely different. Without salt, attackers could use rainbow tables, precomputed databases of password hashes, to crack passwords instantly. The salt makes every hash unique, forcing attackers to start from scratch for each password.

Second, PBKDF2 is intentionally slow with 480,000 iterations. This might seem counterintuitive, but it's brilliant. Each login takes a fraction of a second longer, unnoticeable to users. But for attackers trying billions of password combinations? It becomes computationally infeasible. They can't try millions of passwords per second anymore.

Finally, you store both the salt and the hash with each user record in your database. When a user logs in, you retrieve their salt, hash their input password with it, and compare the result. No plaintext passwords ever touch your database.

Running the example produces an output similar to this:

Salt: 88550cd4b4fac3e5153f9167733a82c0
Hashed: 6555686a7ce17e8fb81de3cb2f88809d81b410e572b039ece7526efd8e864b1c

Encrypting Data: Keeping Secrets Safe

Sometimes you need to encrypt data and decrypt it later, think API keys, sensitive user data, or confidential files. Unlike hashing, encryption is reversible.

You need to be able to decrypt and use that API key, or show users their saved credit card info. For this, use Fernet, a symmetric encryption system where the same key locks and unlocks your data.

I use Fernet for everything from encrypting database backups to protecting API tokens. It's become my go-to because it just works, and it's nearly impossible to mess up.

from cryptography.fernet import Fernet

key = Fernet.generate_key()  # Store this securely!
cipher = Fernet(key)

# Encrypt and decrypt
message = "Sensitive data here"
encrypted = cipher.encrypt(message.encode())
decrypted = cipher.decrypt(encrypted)  # Returns original message

print(encrypted.decode())
print(decrypted.decode())
print(key.decode())

Fernet is beautifully simple and handles the complexity for you, it uses AES encryption, includes authentication to prevent tampering, and adds timestamps to prevent replay attacks. It's what cryptographers call "authenticated encryption," meaning it both hides your data and proves it hasn't been modified.

But here's the critical part: that key is everything. If you lose it, your encrypted data is gone forever. There's no password recovery, no backdoor, no way to get it back. If someone steals it, they can decrypt everything you've ever encrypted with it.

This is why key management is crucial. Store your encryption keys in environment variables, use a secrets management service like AWS Secrets Manager, or at minimum, keep them in a secure configuration file that's never committed to version control. Never hardcode them in your source code.

Running the example produces an output similar to this:

Encrypted:  gAAAAABpnsxPJzrJfStXQvhlYDCBb6y42Qteqm7gcjtsM0KqyNZG5PG7KZxOz8OKXqItL4kGYxdnPq2IJXWS5ClrTWkcaFJm1hG10tnK5LqyqqRP3-NYY1A=
Decrypted:  Sensitive data here
Key:  WrhgfbdekrQcL1TpWo2Nh8lBR7LuqMKruj9Q5XaOc30=

Public-Private Key Encryption: The Magic of Two Keys

Asymmetric encryption uses two keys: a public key (that anyone can have) and a private key (that only you keep). It's like a mailbox, anyone can drop letters in (encrypt with your public key), but only you can open it (decrypt with your private key).

This solves a huge problem: how do two parties communicate securely without meeting first to exchange a secret key? With asymmetric encryption, you can publish your public key anywhere, on your website, in an email, even on a billboard, and anyone can use it to send you encrypted messages that only you can read.

from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes

# Generate keys
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key = private_key.public_key()

# Encrypt with public key, decrypt with private key
message = b"Secret message"
ciphertext = public_key.encrypt(
    message,
    padding.OAEP(mgf=padding.MGF1(algorithm=hashes.SHA256()),
                 algorithm=hashes.SHA256(), label=None)
)

# Decrypt with private key
plaintext = private_key.decrypt(
    ciphertext,
    padding.OAEP(mgf=padding.MGF1(algorithm=hashes.SHA256()),
                 algorithm=hashes.SHA256(), label=None)
)

print("Ciphertext: ", ciphertext)
print("Plaintext: ", plaintext.decode())

This is the foundation for secure communication, digital signatures, SSL/TLS certificates, and SSH keys. You'll use this when different parties need to communicate securely without sharing a secret key beforehand.

Running the example produces an output similar to this:

Ciphertext:  b'\x17\xdeS\xa7*\xa2p\xa6\xd23&{/64\x89\xb8/\x94\x15f\x9b\x80\x15\x86\xaf\xc6\x9ek\xc3\x90\xfb\xcdk^\xd8\xe4J\xb8g\r\n\x99\xacY2)\x92\xf7%\x94\xb9\x98]\x81\xa6{\x98\xcciA\x16\xe4L^B\xf9\x0fR:=\x90\x06\xa9\xf1\x9d__C\x0ca7\xbdK\xacB\xb6C\xc3H{9yq\x1f\xc3\x1fH\xdf\xbe\xbe\x9b\xcci%7\xe0\xa1\xea\xcf\xcf?]\x8c\x1b\xa6\xbf\xb9|\x11\x95\x9f\xf0\x15S=NXs6\xfe\x94\xf2\xf4\xaczt\x89\xcc\x07\xcf\xc7\x82\x9a\xecM~/C\xef\xbc\x12\xf7\xfbJ\xfc\xd2Am\xcd\xe9\xc2}1\xea\xd1\xfe\x07p\xfe\x9e\xec\x95l\xa6\x8a\xcc\xeaZQ\xc5\x81\xe1\x9f\xab_\x08\x9e\xf9\xebo\xb7\xda\x98\xc8\xc3*\xd6k\xe1\x8c=\xbcX \x9a\x12\x9f\x0c\xa5\x8f\xe5\xdb\xf5\x89?S\xa7*\x8e@\xe1\x10mq\x95l\xb9M6\xb7\xc8`\xd8\xe8\xf0\xea\xf5\x11\x93\xc5\xa7/c\x98\x9a\xd9\x8fe\xaf\x19\xc2\xad\xc2=p'
Plaintext:  Secret message

Mistakes I've Made (So You Don't Have To)

Let me save you some pain by sharing the errors I've made:

Don't roll your own crypto: Use established libraries like cryptography. The experts have spent decades finding and fixing vulnerabilities. Your clever modification will almost certainly make things less secure, not more.

Never hardcode keys: I've seen developers commit encryption keys to GitHub, hardcode them in mobile apps (where they're easily extracted), and even print them in error messages. Use environment variables: SECRET_KEY = os.environ.get('SECRET_KEY'). Better yet, use a secrets manager.

Use strong algorithms: MD5 and SHA1 are cryptographically broken. I still see them in legacy code, and it makes me cringe. Stick with SHA-256 or better, and use at least 480,000 iterations for PBKDF2. This number increases over time as computers get faster, check OWASP's current recommendations.

Always use a salt: Every password needs its own random salt, generated at account creation. It's that important.

Store keys separately from data: If someone gets your database dump, they shouldn't automatically get your encryption keys too. That's not encryption, that's theater. Use environment variables, key management services, or at minimum, a separate secured configuration system.

Building Secure APIs in Production?

If you're building real-world Python applications, security doesn't stop at encryption, your data also needs to be validated before it ever reaches your database.

That's where tools like Pydantic become essential for protecting your application from malformed or malicious input.

In my book Practical Pydantic, I walk through how to:

Validate incoming API requests
Enforce strict data contracts
Prevent injection-style attacks through schema validation
Build production-ready FastAPI applications safely

👉 Check it out here: https://leanpub.com/practical-pydantic/c/Dqfo5O4I7sb1

Where to Go From Here

You now have the foundation to protect your users' data, but cryptography is a vast field. As you grow more comfortable with these basics, here's what to explore next:

JWT (JSON Web Tokens) are everywhere in modern web applications. They're a standardized way to create tokens that can be verified without database lookups. Once you understand Fernet, JWTs will make perfect sense.

TLS/SSL certificates with Let's Encrypt let you secure your web traffic. HTTPS isn't optional anymore, browsers actively warn users about unencrypted sites. Fortunately, Let's Encrypt makes it free and automated.

OAuth 2.0 for third-party authentication lets users log in with Google, GitHub, or other providers. You leverage their security infrastructure while reducing your own attack surface.

Hardware Security Modules (HSMs) and cloud key management services like AWS KMS provide a secure place to store and use encryption keys at enterprise scale. When your side project becomes a company, this is where you graduate to.

Full source code of the examples on GitHub: https://github.com/nunombispo/cryptography-python-article

Start Building Secure Applications Today

Cryptography isn't just for security experts anymore.

You have both the power and the responsibility to protect your users' data, and Python makes it accessible. Start with proper password hashing, add encryption for sensitive data, and always use HTTPS. Each step makes your application more secure.

Remember: the best time to implement security was before you launched. The second best time is now.

Your users are counting on you to keep their data safe. With the tools and knowledge you have now, you're ready to build applications that earn their trust.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Automating Long-Term Backup - Hetzner Storage Box to AWS S3 Glacier Deep Archive

Developer Service — Mon, 23 Feb 2026 07:44:37 +0000

You’re paying €28.86/month for 10TB on a Hetzner Storage Box, it’s affordable, easy to access, and does a solid job keeping your backups safe.

But let’s talk about the part nobody likes to think about: what happens when things go sideways?

What if Hetzner has a prolonged outage? What if compliance rules suddenly require geographic redundancy? What if you actually need that 99.999999999% durability, the famous eleven nines enterprise systems are built around?

The move isn’t to ditch Hetzner’s great price-to-performance. It’s to add a second layer using Amazon S3 Glacier Deep Archive at roughly $1/TB/month, giving you ultra-durable, geo-replicated archival storage for the long haul.

The real challenge? Moving your data from point A (Hetzner) to point B (AWS) in a way that’s efficient, incremental, and fully automated… without turning the whole thing into a costly, complex mess.

That’s where automation starts paying for itself, in both euros and peace of mind.

The Motive: Why This Matters

Modern backup strategies run into a weird trade-off. Affordable storage, like ~€2.89/TB/month for 10TB on a Hetzner Storage Box, works great for active data, but it doesn’t give you the enterprise-grade durability or geographic redundancy you need for real disaster recovery.

On the flip side, that eleven-nines durability is available… but paying ~$23/TB/month for Amazon S3 Standard can quickly turn your backup bill into something bigger than your production costs.

The sweet spot is hybrid tiering: keep hot data on affordable storage, and push cold backups into ultra-cheap archival storage.

Why Not Manual Backups?

Manual backups tend to fall apart for three simple reasons:

Human error - eventually, you will forget.
Bandwidth waste - re-uploading unchanged files eats both time and money.
No state tracking - without knowing what’s already uploaded, you can’t resume failed transfers or skip duplicates.

The Streaming Advantage

Most backup tools take the scenic route: download files locally first, then upload them to the destination. That creates three avoidable problems:

Disk space - you need local storage as large as your biggest backup
Time - you’re doing two transfers instead of one
Cost - large-disk VMs aren’t cheap

Streaming directly from SFTP (in this case) to S3 removes all three: no local disk needed, a single transfer path, and the whole thing can run on even the smallest VM.

The Architecture

The Python script you will build sets up a direct pipeline from your Hetzner Storage Box straight into Amazon S3 Glacier Deep Archive, no staging, no local copies, no unnecessary moving parts.

System Flow

Requirements & Dependencies

The script runs on Python 3.7+ and relies on four core libraries:

boto3 - the AWS SDK for Python, handling authentication, streaming uploads, and automatic multipart transfers for large files
paramiko - a pure-Python SSH/SFTP implementation for secure access to your Storage Box
tqdm - provides real-time progress bars with transfer speeds and ETA
python-dotenv - loads environment variables from .env files to keep credentials out of your codebase

On top of that, it uses a few built-in modules:

sqlite3 for lightweight state tracking (no external DB required)
logging for console + file output
os, stat, time for file handling and system utilities

Access Setup

The pipeline needs two things configured outside the script:

Hetzner Storage Box - SFTP access:

The Storage Box must have SFTP access enabled and configured in the Hetzner Robot / Cloud Console.
You need the SFTP host (e.g. u123456.your-storagebox.de), port (usually 23), and a user/password that can read the directories you want to back up.
Without SFTP enabled, the script cannot connect to the Storage Box.

If you still need to get data onto the Storage Box in the first place (local → Hetzner), see Build Your Own Low-Cost Cloud Backup with Hetzner Storage Boxes.

AWS - access keys for S3:

Create an IAM user (or use an existing one) with permission to write to your target S3 bucket (e.g. s3:PutObject on that bucket).
Generate access keys for that user and put AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in your .env.
The script uses these to authenticate with S3; no keys means uploads will fail.

The Code

You can find the complete implementation on GitHub, but here are the core building blocks that make this pipeline from Hetzner to Amazon S3 Glacier Deep Archive actually work in practice.

System Components Overview

The script is structured around five key layers that work together:

Configuration Layer - pulls in credentials and runtime settings from a .env file
State Persistence Layer - uses SQLite to track which files have already been uploaded
SFTP Discovery Engine - recursively walks your Storage Box directory tree
Streaming Transfer Engine - streams files directly from SFTP to S3
Main Orchestration Loop - ties everything together and manages execution flow

Let’s break down each piece with code snippets and more importantly, why it’s built this way.

Configuration Loading

from dotenv import load_dotenv
load_dotenv()

HETZNER_HOST = os.getenv("HETZNER_HOST")
HETZNER_USER = os.getenv("HETZNER_USER")
HETZNER_PASSWORD = os.getenv("HETZNER_PASSWORD")
REMOTE_PATH = os.getenv("REMOTE_PATH")

S3_BUCKET = os.getenv("S3_BUCKET")
S3_PREFIX = os.getenv("S3_PREFIX")
AWS_REGION = os.getenv("AWS_REGION")
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")

STATE_DB = os.getenv("STATE_DB", "archive_state.db")
LOG_FILE = os.getenv("LOG_FILE", "archive.log")
# Comma-separated paths to exclude (that path and everything under it)
EXCLUDE_PATHS_RAW = os.getenv("EXCLUDE_PATHS", os.getenv("EXCLUDE_DIRS", ""))
EXCLUDE_PATHS = [p.strip().strip("/") for p in EXCLUDE_PATHS_RAW.split(",") if p.strip()]

All secrets and runtime settings are stored in a .env file and loaded at start-up using python-dotenv. This keeps credentials (for Hetzner and AWS), bucket details, remote paths, logging config, and optional exclusions outside of your actual codebase.

Keeping configuration separate from code is one of those boring best practices that quietly saves you from very real disasters. Loading environment variables at runtime lets you:

Avoid committing secrets to version control
Run the same code across dev, staging, and production with different configs
Rotate credentials without redeploying code
Open-source the project without leaking access keys

If you want to level up config and validation in Python (typed settings, env validation, APIs), check out Practical Pydantic.

State Database Initialization

def init_db():
    conn = sqlite3.connect(STATE_DB)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS files (
            path TEXT PRIMARY KEY,
            size INTEGER,
            mtime INTEGER,
            uploaded_at INTEGER
        )
    """)
    conn.commit()
    return conn

A local SQLite database keeps track of every file that’s already been archived using a simple fingerprint: (path, size, mtime). For each upload, it records the file path (as the primary key), its size in bytes, last modified timestamp, and when it was successfully transferred.

If the script stops midway, or you run it again later, it immediately knows what’s already been handled and what still needs attention.

SQLite gives you a zero-config database with no server to install or maintain. The schema is intentionally minimal:

path as the PRIMARY KEY guarantees uniqueness and fast lookups
size + mtime act as a reliable change detector
uploaded_at gives you an audit trail for debugging or reporting

Using CREATE TABLE IF NOT EXISTS also makes the process idempotent, meaning you can run it repeatedly without errors, duplicate uploads, or wasted bandwidth. Just clean, incremental backups every time.

Smart Duplicate Detection

def already_uploaded(conn, path, size, mtime):
    row = conn.execute(
        "SELECT size, mtime FROM files WHERE path=?",
        (path,)
    ).fetchone()
    return row == (size, mtime)

This function checks the state database to see whether a file with the same path, size, and last modification time has already been uploaded, allowing the script to intelligently skip anything that hasn’t changed.

That one-line tuple comparison, row == (size, mtime), quietly handles all the important cases:

If the path isn’t in the database → None == (size, mtime) → False → upload it
If the path exists but size or mtime changed → (old_size, old_mtime) == (new_size, new_mtime) → False → upload it
If the path exists with identical size and mtime → True → skip it

This makes incremental backups dramatically more efficient. After the initial full run, future executions may end up skipping the vast majority of files. Since the database is updated after each successful transfer, the process is also interruption-safe, if something stops halfway through, it simply resumes where it left off.

State Persistence After Upload

def mark_uploaded(conn, path, size, mtime):
    conn.execute(
        "REPLACE INTO files VALUES (?, ?, ?, ?)",
        (path, size, mtime, int(time.time()))
    )
    conn.commit()

After a file is successfully uploaded, this function stores its metadata in the state database, creating a persistent record that prevents the same unchanged file from being transferred again in future runs.

Using REPLACE instead of a plain INSERT neatly covers both scenarios:

If the file path already exists → update the existing record
If it doesn’t → insert a new one

The immediate commit() is what makes this reliable. If the script crashes right after this step, the upload is already recorded, so the next run won’t waste time or bandwidth reprocessing it.

This gives you interruption safety at the file level. Each commit is atomic: you either have a fully written record, or nothing at all. No half-finished states, no ambiguity.

SFTP Directory Walker with Path Exclusions

def _normalize_path(p):
    return p.replace("//", "/").strip("/") if p and p != "." else ""

def _is_path_excluded(full_path):
    norm = _normalize_path(full_path)
    if not norm:
        return False
    for exclude in EXCLUDE_PATHS:
        if norm == exclude or norm.startswith(exclude + "/"):
            return True
    return False

def walk_sftp(sftp, path):
    for entry in sftp.listdir_attr(path):
        # Skip hidden files and directories
        if entry.filename.startswith('.'):
            continue

        # Normalize path
        if path == ".":
            full = entry.filename
        else:
            full = f"{path}/{entry.filename}".replace("//", "/")

        if stat.S_ISDIR(entry.st_mode):
            if _is_path_excluded(full):
                continue
            yield from walk_sftp(sftp, full)
        else:
            yield full, entry.st_size, entry.st_mtime

This recursive generator walks the full directory tree on your Storage Box over SFTP. It lists directory contents, separates files from folders using mode attributes, skips hidden entries (anything starting with a dot), and ignores any directory that matches, or lives under, a path defined in EXCLUDE_PATHS. That makes it easy to leave out things like backup/old, log folders, or cache directories from your archive.

This behaves a lot like Python’s os.walk(), but for remote storage. A few design choices make it scale cleanly:

Generator-based (yield): files are produced one at a time, so memory usage stays flat even with huge directory trees
Hidden file filtering: avoids wasting time on system artifacts like .bash_history or .ssh/
Path-prefix exclusions: _is_path_excluded() treats entries in EXCLUDE_PATHS as prefixes, excluding backup/old skips that folder and everything inside it
Normalization: _normalize_path() strips extra slashes so path comparisons are consistent
Recursive delegation: yield from walk_sftp(...) passes control without building large intermediate lists

The result is a memory-efficient traversal that can handle anything from a few hundred files to millions, without needing to load the full tree into RAM first.

Streaming Upload with Progress

def upload_stream(fileobj, s3_key, size):
    progress = tqdm(
        total=size,
        unit="B",
        unit_scale=True,
        desc=s3_key,
        leave=False
    )

    def callback(bytes_amount):
        progress.update(bytes_amount)
        metrics["bytes_uploaded"] += bytes_amount

    s3.upload_fileobj(
        fileobj,
        Bucket=S3_BUCKET,
        Key=s3_key,
        ExtraArgs={"StorageClass": "DEEP_ARCHIVE"},
        Callback=callback
    )
    progress.close()

This function streams files directly from Hetzner SFTP to Amazon S3 Glacier Deep Archive without ever writing them to local disk. The SFTP file object is passed straight to boto3.upload_fileobj(), which handles chunked uploads internally. Real-time progress bars are updated via a callback after each chunk, giving visibility into transfer speed and completion.

This is the core of the streaming architecture. Key points:

upload_fileobj(): Streams any file-like object (including SFTP handles) to S3, no temporary disk storage required
StorageClass='DEEP_ARCHIVE': Saves cost by writing directly to the cheapest archival tier, avoiding Standard tier fees and lifecycle transitions
Callback=callback: Updates progress bar and metrics["bytes_uploaded"] after each chunk (~8MB)
leave=False: Keeps the console clean by removing progress bars when done

This method handles gigabyte-scale files on machines with only a few megabytes of RAM. The callback pattern provides live feedback without blocking the upload, and the metrics dictionary allows cumulative tracking for reporting purposes.

Main Orchestration

def main():
    conn = init_db()

    transport = paramiko.Transport((HETZNER_HOST, HETZNER_PORT))
    transport.connect(username=HETZNER_USER, password=HETZNER_PASSWORD)
    sftp = paramiko.SFTPClient.from_transport(transport)

    try:
        log.info(f"Starting backup from: {REMOTE_PATH}")
        for path, size, mtime in walk_sftp(sftp, REMOTE_PATH):
            if REMOTE_PATH == ".":
                rel = path
            else:
                rel = path.replace(REMOTE_PATH, "").lstrip("/")
            s3_key = f"{S3_PREFIX}{rel}"

            if already_uploaded(conn, path, size, mtime):
                log.info(f"SKIP  {path}")
                metrics["skipped"] += 1
                continue

            log.info(f"UPLOAD {path} -> s3://{S3_BUCKET}/{s3_key}")

            with sftp.open(path, "rb") as f:
                upload_stream(f, s3_key, size)

            mark_uploaded(conn, path, size, mtime)
            metrics["uploaded"] += 1
    finally:
        sftp.close()
        transport.close()
        conn.close()

The main() function orchestrates the entire pipeline: it connects to Hetzner via SFTP and AWS via boto3, initializes the SQLite state database, and iterates over every file discovered by the SFTP walker. The S3 key is built from S3_PREFIX plus the path relative to REMOTE_PATH (so backing up from a subdirectory doesn’t duplicate that prefix in the archive). For each file:

Checks if it’s already uploaded using the state database
Skips unchanged files (logging and incrementing the skip counter)
Streams new or modified files directly to Amazon S3 Glacier Deep Archive
Updates the database after successful upload
Tracks metrics like files uploaded, skipped, and bytes transferred

Why it’s built this way:

Connection management: Opens SFTP and DB connections at the start; try/finally ensures they’re closed even if something goes wrong
Generator-driven: walk_sftp() feeds one file at a time, keeping memory usage constant
Early state check: already_uploaded() prevents unnecessary file reads and saves bandwidth
Context managers: with sftp.open(...) safely closes remote file handles
Immediate commits: mark_uploaded() writes to the database after each file, making the system resilient to crashes or interruptions

The design favours simplicity over complexity: no threads, no async, no elaborate state machines.

This keeps debugging straightforward and failure modes predictable. At the end, a summary report logs total files uploaded, skipped, bytes transferred, and total runtime.

Progress Tracking with Metrics

metrics = {
    "uploaded": 0,
    "skipped": 0,
    "bytes_uploaded": 0,
    "start_time": time.time(),
}

# ... at the end:
duration = time.time() - metrics["start_time"]
log.info("==== SUMMARY ====")
log.info(f"Uploaded files : {metrics['uploaded']}")
log.info(f"Skipped files  : {metrics['skipped']}")
log.info(f"Bytes uploaded : {metrics['bytes_uploaded']:,}")
log.info(f"Duration (sec) : {int(duration)}")

A global metrics dictionary tracks operational statistics throughout execution: files uploaded, files skipped, total bytes transferred, and start time for duration calculation. At the end, a summary report is logged.

A simple dictionary tracks operational metrics. The callback in upload_stream() accumulates bytes, while the main loop increments counters. At the end, you get a summary:

==== SUMMARY ====
Uploaded files : 42
Skipped files  : 1,583
Bytes uploaded : 15,234,567,890
Duration (sec) : 340

This makes it easy to track incremental efficiency over time. After the first full backup, subsequent runs should show high skip counts and low upload counts, exactly what you want. These metrics help you verify the script is working correctly and monitor bandwidth usage.

Cost Analysis

Using a hybrid approach, Hetzner for active backups and Amazon S3 Glacier Deep Archive for cold storage, gives enterprise-grade durability at a fraction of standard cloud costs.

Monthly Storage Costs (10TB Example)

Service	Cost	Role
Hetzner Storage Box BX31 (10TB)	€28.86	Active, frequently accessed backups
AWS S3 Glacier Deep Archive (10TB)	$10.10	Cold, geo-redundant disaster recovery
Total	€39.96 (~$44)	Hybrid storage solution
Per-TB Cost	€4.00/TB (~$4.40/TB)	Unit economics

Comparison:

AWS S3 Standard: $230/month
Backblaze B2: $50/month (+ egress fees)
Google Cloud Archive: $12/month (higher retrieval costs)

Hybrid tiering delivers reliability at a fraction of S3 Standard pricing.

Transfer & API Costs

Hetzner → AWS: Free egress & ingress
S3 PUT Requests: ~$5 for 100,000 files initially; ~$0.25/month for incremental changes (~5%)
Impact: Negligible compared to storage savings

Retrieval Costs (Deep Archive Caveat)

Retrieval: $0.02/GB
Restoration: 12–48 hours
Temporary storage in S3 Standard: $0.023/GB/month

Example: Restoring 1TB costs ~$20 in retrieval fees plus ~$23/month if kept temporarily. Deep Archive is designed for “write-once, rarely-read” disaster recovery.

This hybrid setup gives the best of both worlds: affordable, accessible active storage and ultra-durable, geo-redundant archival backup, without breaking the bank.

Conclusion

Backing up from a Hetzner Storage Box to Amazon S3 Glacier Deep Archive gives you a powerful, cost-effective solution with:

Affordability: ~$5/month per TB using hybrid tiered storage
Durability: 11-nines (99.999999999%) with built-in geo-replication
Efficiency: Incremental, streaming transfers with zero local disk usage
Reliability: Interrupt-safe, state-persistent, fully logged operations
Automation: Fully hands-off, schedule via cron, Task Scheduler, or any job runner

Whether for personal data, compliance, or enterprise disaster recovery, this pipeline delivers enterprise-grade reliability at a fraction of typical cloud costs.

The code is production-ready, tested, and designed to run unattended. By combining Hetzner’s affordability with AWS Glacier Deep Archive’s durability, you get the best of both worlds.

Full code and documentation: https://github.com/nunombispo/hetzner-to-aws-s3-glacier-backups

More on Python, Hetzner and automation: Hetzner Storage Box backup, Django on Hetzner + Dokku) and Python One-Liners.

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

My Experience Building a URL Shortener Like TinyURL

Developer Service — Mon, 16 Feb 2026 07:53:57 +0000

You know that moment when you're sharing a link to your latest blog post, and the URL looks like https://developer-service.blog/build-your-own-low-cost-cloud-backup-with-hetzner-storage-boxes/?

Yeah, that's not going to fit nicely anywhere.

I faced this problem constantly while managing my blog, sharing free guides, and tracking GitHub repository statistics across different platforms.

Twitter's character limits, newsletter formatting issues, clean QR codes for printed materials, every platform seemed to hate my long, descriptive URLs.

Sure, I could use bit.ly or TinyURL, but then I'd be giving up control over my links, paying for premium features, and most importantly, losing valuable analytics data.

So I did what any self-respecting developer would do: I spent a weekend building my own URL shortener. And you know what? It was way more interesting than I expected.

Full source code link at the end of the article

What Exactly Is a URL Shortener?

At its core, a URL shortener is beautifully simple: it takes a long URL and maps it to a short, memorable code. When someone visits the short URL, they're redirected to the original destination.

Long URL:  https://developer-service.blog/build-your-own-low-cost-cloud-backup-with-hetzner-storage-boxes/
           ↓
Short URL: https://myurl.app/0001ab
           ↓
Redirect:  302 Found → Original URL

But the devil's in the details. A production-ready URL shortener needs to handle:

Unique short codes without collisions
Analytics tracking for every click
Custom slugs for branded links
Expiration dates for temporary campaigns
Security against abuse and malicious URLs

The Technical Challenge: URL Slug Collisions and Uniqueness

This was the most interesting problem to solve. How do you generate short, unique identifiers that won't collide as your database grows?

The Naive Approach (Don't Do This)

My first instinct was to use random strings:

import random
import string

def generate_slug():
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))

The problem? With a 6-character alphanumeric string (36^6 = 2.1 billion possibilities), you'd think collisions are rare. But thanks to the quirky birthday paradox, you'll hit your first collision around 55,000 URLs, which is a bit sooner than expected.

The Better Approach: Base62 Encoding with Auto-Increment IDs

Instead, I went with a deterministic approach using database auto-increment IDs:

import string
from typing import Optional

BASE62 = string.digits + string.ascii_lowercase + string.ascii_uppercase

def encode_id(num: int, min_length: int = 6) -> str:
    """
    Convert a number to a base62 string, padded to minimum length.
    With 6 characters and base62, you can encode up to 56.8 billion URLs.
    """
    if num == 0:
        return BASE62[0].zfill(min_length)

    result = []
    while num:
        result.append(BASE62[num % 62])
        num //= 62

    encoded = ''.join(reversed(result))

    # Pad with leading zeros to meet minimum length
    if len(encoded) < min_length:
        encoded = BASE62[0] * (min_length - len(encoded)) + encoded

    return encoded

def decode_slug(slug: str) -> Optional[int]:
    """Convert a base62 string back to a number."""
    try:
        num = 0
        for char in slug:
            num = num * 62 + BASE62.index(char)
        return num
    except ValueError:
        # Character not in BASE62 alphabet
        return None

This approach guarantees:

Zero collisions: Each ID maps to exactly one slug
Consistent length: All slugs are 6 characters (e.g., "000001", "000abc")
Predictable growth: You know exactly how many URLs each slug length supports
Reversibility: You can decode slugs back to IDs for quick lookup

The math is beautiful:

6 characters (default): 56.8 billion possible URLs
First URL: 000001 (ID: 1)
100th URL: 0001Cv (ID: 100)
Millionth URL: 004C92 (ID: 1,000,000)

By padding to 6 characters, all URLs have a consistent, professional look. When you eventually hit 6 characters of actual data, they're already quite short anyway!

Custom Slugs: The Best of Both Worlds

Of course, sometimes you want a branded slug like myurl.app/blog-launch instead of myurl.app/000x7k. I added support for custom slugs with validation and uniqueness checks:

RESERVED_SLUGS = {"stats", "shorten", "admin", "login", "logout"}

def validate_custom_slug(slug: str) -> bool:
    """Validate custom slug format."""
    if not slug or len(slug) < 3 or len(slug) > 50:
        return False

    allowed_chars = set(string.ascii_letters + string.digits + '-_')
    return all(char in allowed_chars for char in slug)

# In the /shorten route:
if custom_slug:
    custom_slug = custom_slug.strip().lower()  # Normalize to lowercase

    # Check if slug is reserved
    if custom_slug in RESERVED_SLUGS:
        raise HTTPException(status_code=400, detail="Slug is reserved")

    # Validate format
    if not validate_custom_slug(custom_slug):
        raise HTTPException(status_code=400, detail="Invalid slug format")

    # Check if already taken
    existing = db.query(URL).filter(URL.slug == custom_slug).first()
    if existing:
        raise HTTPException(status_code=400, detail="Slug already taken")

    new_url = URL(slug=custom_slug, long_url=long_url)
else:
    # Auto-generate from ID
    new_url = URL(long_url=long_url, slug="temp")
    db.add(new_url)
    db.flush()
    new_url.slug = encode_id(new_url.id, min_length=6)

Custom slugs are validated to ensure they only contain alphanumeric characters, hyphens, and underscores (3-50 characters). Reserved slugs like "stats" and "admin" are blocked to avoid conflicts with routes.

Analytics and Dashboard

This is where building your own URL shortener really shines. Every major platform (Twitter, LinkedIn, newsletters, GitHub README) has different traffic patterns, and I wanted to see them all in one place.

What I Track

For every click, I capture:

class Click(Base):
    __tablename__ = "clicks"

    id = Column(Integer, primary_key=True, index=True)
    url_id = Column(Integer, ForeignKey("urls.id"), nullable=False, index=True)
    timestamp = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)

    # Request metadata
    referrer = Column(String, nullable=True)
    user_agent = Column(String, nullable=True)
    ip_address = Column(String, nullable=True)

    # Geographic data (would require GeoIP library)
    country = Column(String, nullable=True)
    city = Column(String, nullable=True)

    # Device information (parsed from user agent)
    device_type = Column(String, nullable=True)  # mobile, desktop, tablet, bot
    browser = Column(String, nullable=True)
    os = Column(String, nullable=True)

    url = relationship("URL", back_populates="clicks")

The Dashboard

I built a simple dashboard using Bootstrap 5 that shows:

Total clicks with time-series line chart
Unique visitors tracked by IP address
Device type breakdown (mobile, desktop, tablet, bot) as a pie chart
Top referrers with percentage bars (Twitter, LinkedIn, Direct, etc.)
Average clicks per day since creation
Recent clicks table showing timestamp, referrer, device, browser, OS, and IP

The most surprising insight? My GitHub repository links get significantly more clicks from mobile devices than desktop. I never would have guessed that people browse code repos on their phones so much!

Note: Geographic distribution (countries/cities) could be added using a GeoIP library like geoip2, but I kept it simple to avoid external dependencies and API limits.

Building it with FastAPI, Bootstrap 5, and Jinja2

Let me walk you through the actual implementation. I chose this stack because:

FastAPI: Lightning-fast, automatic API docs, async support
Bootstrap 5: Quick, responsive UI without wrestling with CSS
Jinja2: Server-side rendering for SEO and simplicity
SQLite: Zero-config database perfect for side projects

The Database Schema

Simple SQLAlchemy models with a one-to-many relationship:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime, timezone

class URL(Base):
    __tablename__ = "urls"

    id = Column(Integer, primary_key=True, autoincrement=True, index=True)
    slug = Column(String, unique=True, index=True, nullable=False)
    long_url = Column(String, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    expires_at = Column(DateTime, nullable=True)

    clicks = relationship("Click", back_populates="url", cascade="all, delete-orphan")

class Click(Base):
    __tablename__ = "clicks"

    id = Column(Integer, primary_key=True, index=True)
    url_id = Column(Integer, ForeignKey("urls.id"), nullable=False, index=True)
    timestamp = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)

    # Request metadata
    referrer = Column(String, nullable=True)
    user_agent = Column(String, nullable=True)
    ip_address = Column(String, nullable=True)

    # Device information (parsed from user agent)
    device_type = Column(String, nullable=True)
    browser = Column(String, nullable=True)
    os = Column(String, nullable=True)

    url = relationship("URL", back_populates="clicks")

The URL model stores the shortened links, while Click tracks every redirect for analytics. Using datetime.now(timezone.utc) ensures timezone-aware timestamps. The cascade option means deleting a URL also removes all its clicks.

The FastAPI Application

Here are the key routes (see full implementation on GitHub):

from fastapi import FastAPI, HTTPException, Request, Depends, Form
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from typing import Optional

app = FastAPI(title="TinyURL Clone")

@app.post("/shorten")
async def create_short_url(
    request: Request,
    long_url: str = Form(...),
    custom_slug: Optional[str] = Form(None),
    db: Session = Depends(get_db)
):
    """Create a new short URL"""
    # Validate URL format
    if not validate_url(long_url):
        raise HTTPException(status_code=400, detail="Invalid URL format")

    if custom_slug:
        custom_slug = custom_slug.strip().lower()

        # Validate custom slug format
        if not validate_custom_slug(custom_slug):
            raise HTTPException(status_code=400, detail="Invalid slug format")

        # Check if already taken
        existing = db.query(URL).filter(URL.slug == custom_slug).first()
        if existing:
            raise HTTPException(status_code=400, detail="Slug already taken")

        new_url = URL(slug=custom_slug, long_url=long_url)
        db.add(new_url)
        db.commit()
    else:
        # Auto-generate from ID
        new_url = URL(long_url=long_url, slug="temp")
        db.add(new_url)
        db.flush()  # Get the auto-increment ID
        new_url.slug = encode_id(new_url.id)
        db.commit()

    base_url = str(request.base_url).rstrip('/')
    return {"short_url": f"{base_url}/{new_url.slug}"}

@app.get("/{slug}")
async def redirect_to_long_url(
    slug: str,
    request: Request,
    db: Session = Depends(get_db)
):
    """Redirect short URL to original URL and track analytics"""
    url = db.query(URL).filter(URL.slug == slug).first()

    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    # Check expiration
    if url.expires_at and datetime.now(timezone.utc) > url.expires_at:
        raise HTTPException(status_code=410, detail="URL has expired")

    # Parse user agent for device info
    device_info = parse_user_agent(request.headers.get("user-agent"))

    # Track the click
    click = Click(
        url_id=url.id,
        referrer=request.headers.get("referer"),
        user_agent=request.headers.get("user-agent"),
        ip_address=request.client.host if request.client else None,
        device_type=device_info["device_type"],
        browser=device_info["browser"],
        os=device_info["os"]
    )
    db.add(click)
    db.commit()

    return RedirectResponse(url=url.long_url, status_code=302)

The application also includes routes for viewing stats (/stats/{slug}) and the homepage dashboard. Note the use of Form(...) for proper form data handling and the parse_user_agent() utility for extracting device information.

The Bootstrap 5 Dashboard Template

I built a clean, responsive UI using Bootstrap 5. Here's the key form structure:

<!-- templates/index.html - Main form (simplified) -->
<form id="shortenForm">
  <div class="mb-3">
    <label for="longUrl" class="form-label">Long URL</label>
    <input
      type="url"
      class="form-control form-control-lg"
      id="longUrl"
      placeholder="https://example.com/very/long/url"
      required
    />
  </div>

  <div class="mb-3">
    <label for="customSlug" class="form-label"> Custom Slug (Optional) </label>
    <input
      type="text"
      class="form-control"
      id="customSlug"
      placeholder="my-custom-link"
    />
  </div>

  <button type="submit" class="btn btn-primary btn-lg w-100">
    Shorten URL
  </button>
</form>

<script>
  document
    .getElementById("shortenForm")
    .addEventListener("submit", async (e) => {
      e.preventDefault();

      const formData = new URLSearchParams();
      formData.append("long_url", document.getElementById("longUrl").value);
      formData.append(
        "custom_slug",
        document.getElementById("customSlug").value,
      );

      const response = await fetch("/shorten", {
        method: "POST",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: formData,
      });

      const data = await response.json();
      // Display the short URL
      document.getElementById("shortUrl").value = data.short_url;
    });
</script>

The stats dashboard (templates/stats.html) displays analytics cards showing total clicks, unique visitors, referrer breakdown, and device statistics using Chart.js for visualizations.

Full templates available in the GitHub repository.

SQLite: The Perfect Database for Side Projects

One of the best decisions I made was using SQLite. Here's why:

Zero Configuration

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Can be overridden with environment variable
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./urls.db")

# SQLite-specific optimizations
connect_args = {
    "check_same_thread": False,
    "timeout": 30
}

engine = create_engine(
    DATABASE_URL,
    connect_args=connect_args,
    pool_pre_ping=True  # Handle stale connections
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

The entire database is a single file (urls.db) that lives in your project directory. You can back it up, version control it, or deploy it anywhere. The pool_pre_ping option helps prevent stale connection errors.

Surprisingly Powerful

SQLite handles:

Transactions (ACID compliance)
Foreign keys
Indexes
Full-text search
JSON columns (SQLite 3.38+)

For a URL shortener with even millions of URLs, SQLite performs beautifully. It's only when you need high-concurrency writes (thousands per second) that you'd need PostgreSQL or MySQL.

Easy Backups

# Backup
sqlite3 urls.db ".backup urls_backup.db"

# Restore
cp urls_backup.db urls.db

When to Upgrade

You should consider moving to PostgreSQL when:

You're getting >100 writes/second
You need multiple servers accessing the same database
You want advanced features like full-text search or geospatial queries
You're running in a distributed environment

For a personal URL shortener tracking blog posts and GitHub stats, SQLite is more than sufficient.

Deployment and Production Considerations

Running in Production

I deployed mine using Docker on a small VPS:

FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application files
COPY . .

# Create directory for database
RUN mkdir -p /app/data

# Expose port
EXPOSE 8000

# Run the application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml (simplified)
services:
  app:
    build: .
    container_name: tinyurl-clone
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data
      - ./templates:/app/templates
    environment:
      - DATABASE_URL=sqlite:///./data/urls.db
    restart: unless-stopped

The volume mounts ensure your database persists across container restarts, and mounting templates allows hot-reload during development.

Security Considerations

The current implementation includes basic validation (URL format checking and slug sanitization), but for production deployment, you should add:

Rate Limiting: Use slowapi to limit requests (e.g., 10/minute per IP) to prevent abuse

   from slowapi import Limiter
   limiter = Limiter(key_func=get_remote_address)

   @app.post("/shorten")
   @limiter.limit("10/minute")
   async def create_short_url(...):
       # Your code here

Malicious URL Blocking: Maintain a blocklist of known malicious domains

   BLOCKED_DOMAINS = {"malware.com", "phishing.net"}

   def validate_url(url: str) -> bool:
       parsed = urlparse(url)
       if parsed.netloc in BLOCKED_DOMAINS:
           return False

HTTPS Only: Use Let's Encrypt with nginx reverse proxy in production
Database Backups: Automate regular SQLite backups to prevent data loss

For a production system handling sensitive data, consider adding authentication, API keys, and more comprehensive logging.

Try It Out: Running the Demo

Want to see it in action? Getting started takes less than 5 minutes:

# Clone the repository
git clone https://github.com/nunombispo/building-my-own-tinyurl-article
cd building-my-own-tinyurl-article

# Install dependencies
pip install -r requirements.txt

# Run the application
uvicorn main:app --reload

Then visit http://localhost:8000 in your browser. You'll see:

Homepage: Clean form to shorten URLs with optional custom slugs
Instant results: Copy button and links to visit or view stats
Analytics dashboard: Visit /stats/{slug} to see click tracking in action

Try creating a few short URLs, clicking them from different devices, and watching the analytics populate in real-time. The SQLite database will be created automatically on first run.

Example of the homepage:

Example of the analytics main information:

And referrers and recent clicks:

Full source code at: https://github.com/nunombispo/building-my-own-tinyurl-article

Was It Worth It?

Absolutely. The total time investment was about 8 hours, and I now have:

Complete ownership of my links
Detailed analytics tailored to my needs
No monthly subscription fees
A fun weekend project that taught me about base62 encoding, collision handling, and FastAPI

Plus, I can share this with friends, add it to my portfolio, and know exactly how my content performs across different platforms.

Conclusion

Building a URL shortener taught me that sometimes the best tools are the ones you build yourself. Not because they're necessarily better than existing solutions, but because they're perfectly tailored to your needs.

For tracking blog posts, free guides, and GitHub statistics, having complete control over the analytics pipeline is invaluable. You own your data, customize the features you want, and learn something new in the process.

And honestly? It was just really fun to build.

If you're thinking about building your own, I say go for it. The technical challenges are interesting but manageable, and the end result is something you'll actually use every day. Start simple with the core features, then add complexity as you need it.

The complete source code is available on GitHub, so clone it, try it out, and make it your own!

Follow me on Twitter: https://twitter.com/DevAsService

Follow me on Instagram: https://www.instagram.com/devasservice/

Follow me on TikTok: https://www.tiktok.com/@devasservice

Follow me on YouTube: https://www.youtube.com/@DevAsService

Server Monitoring Without Prometheus - A Python + SSH Approach

Developer Service — Mon, 09 Feb 2026 07:50:08 +0000

You don’t need a monitoring stack that requires its own monitoring. If you’re running a handful of servers, Prometheus and Grafana are often more work than the problems they’re meant to solve.

If you can ssh user@server, you already have everything you need.

In this tutorial, we’ll build a lightweight, no-agent server monitoring tool using Python and SSH, designed for indie hackers, solo founders, and small teams. By the end, you’ll have a single script that checks CPU, memory, disk usage, and critical services across multiple servers, renders a clean terminal dashboard, and sends alerts when something actually breaks.

What we’ll build:

Monitor multiple servers over SSH (no agents, no daemons)
A real-time terminal dashboard with colors and status indicators
Alerts via ntfy.sh
Remote server updates with push notifications
One Python script you can read, trust, and extend

Why Not Just Use Prometheus?

Before we dive in, let’s get this out of the way: Prometheus and Grafana are excellent tools. At scale, they’re hard to beat, and they’re industry standard for a reason.

But if you’re running a small setup, say 1–20 servers, a few side projects, or a lean SaaS, they often introduce more complexity than actual value.

What starts as “just add monitoring” quickly turns into:

Heavy infrastructure: Prometheus, Grafana, node exporters, persistent storage, backups
Ongoing maintenance: Upgrades, config drift, broken dashboards at the worst time
A steep learning curve: PromQL, recording rules, and yet another YAML ecosystem

All of that just to answer a few basic questions:

Is the server up?

Is disk space running out?

Did nginx or Docker crash?

Meanwhile, SSH is already (probably) open on your servers, and Python is already installed (or trivial to install). You don’t need a full metrics pipeline to check system health, you just need a reliable way to run commands and interpret the results.

If you can ssh user@server, you can monitor it.

That’s the premise of this article. We’ll keep the stack boring, the moving parts minimal, and the script small enough that you actually understand, and trust, it.

Project Setup

Let's get you started by creating a new project directory and set up the environment:

mkdir server-monitor
cd server-monitor

Then, create a requirements.txt file:

paramiko>=4.0.0
rich>=14.3.2
pyyaml>=6.0.3
requests>=2.32.5

And install the dependencies:

pip install -r requirements.txt

What each library does:

paramiko -> SSH connections and command execution
rich -> Beautiful terminal output with tables and colors
pyyaml -> Parse our server configuration file
requests -> Send alerts to webhooks (Slack)

Now create a servers.yml file to define which servers to monitor:

servers:
  - name: prd-dokku
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

  - name: prd-ghost
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

  - name: prd-n8n
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

thresholds:
  cpu_load_factor: 2.0 # Alert if load > cores × 2
  memory_percent: 85
  disk_percent: 25

Here you should replace the IP addresses and usernames with your actual servers.

Make sure you have SSH key authentication set up for each server. If your key has a passphrase, use the SSH_KEY_PASSPHRASE environment variable or add key_passphrase to a server entry; the script will also prompt interactively if needed.

SSH Connection Manager

Let's start building the monitor. Create a file called monitor.py and you'll build it piece by piece.

First, you'll need a function to connect to servers and run commands over SSH:

"""
Simple server monitoring over SSH
"""

import os
import sys
import threading
import getpass
import paramiko
import yaml
from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class ServerMetrics:
    """Container for server metrics"""
    name: str
    reachable: bool = True
    cpu_load: Optional[float] = None
    memory_percent: Optional[float] = None
    disk_percent: Optional[float] = None
    services: Dict[str, str] = field(default_factory=dict)
    error: Optional[str] = None


class SSHClient:
    """Manages SSH connections to servers"""

    def __init__(
        self,
        host: str,
        user: str,
        key_file: Optional[str] = None,
        key_passphrase: Optional[str] = None,
    ):
        self.host = host
        self.user = user
        self.key_file = key_file
        self.key_passphrase = key_passphrase
        self.client = None

    def _load_pkey(self, passphrase: Optional[str] = None) -> Optional[paramiko.PKey]:
        """Load private key, trying RSA then Ed25519. Returns None to fall back to key_filename."""
        if not self.key_file:
            return None
        password = passphrase or self.key_passphrase
        # Try RSA key first (PEM format)
        try:
            return paramiko.RSAKey.from_private_key_file(
                self.key_file, password=password
            )
        except paramiko.ssh_exception.PasswordRequiredException:
            if password is None:
                password = getpass.getpass(f"SSH key passphrase for {self.key_file}: ")
            return paramiko.RSAKey.from_private_key_file(
                self.key_file, password=password
            )
        except (paramiko.ssh_exception.SSHException, OSError):
            pass
        # Try Ed25519
        try:
            return paramiko.Ed25519Key.from_private_key_file(
                self.key_file, password=password
            )
        except paramiko.ssh_exception.PasswordRequiredException:
            if password is None:
                password = getpass.getpass(f"SSH key passphrase for {self.key_file}: ")
            return paramiko.Ed25519Key.from_private_key_file(
                self.key_file, password=password
            )
        except (paramiko.ssh_exception.SSHException, OSError):
            pass
        return None

    def connect(self) -> bool:
        """Establish SSH connection"""
        try:
            self.client = paramiko.SSHClient()
            self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            connect_kwargs = {
                'hostname': self.host,
                'username': self.user,
                'timeout': 10,
                'look_for_keys': True,
                'allow_agent': True,
            }

            # Load key explicitly (RSA / Ed25519) with optional passphrase
            if self.key_file:
                pkey = self._load_pkey()
                if pkey is not None:
                    connect_kwargs['pkey'] = pkey
                else:
                    connect_kwargs['key_filename'] = self.key_file
                    if self.key_passphrase:
                        connect_kwargs['passphrase'] = self.key_passphrase

            self.client.connect(**connect_kwargs)
            return True

        except Exception as e:
            print(f"❌ Failed to connect to {self.host}: {e}")
            return False

    def run_command(
        self, command: str, stream_output: bool = False
    ) -> tuple[str, str, int]:
        """
        Run a command over SSH.
        Returns: (stdout, stderr, exit_code).
        If stream_output=True, print stdout/stderr as it arrives (for long-running commands).
        """
        if not self.client:
            return "", "Not connected", 1

        try:
            stdin, stdout, stderr = self.client.exec_command(command)
            channel = stdout.channel

            if stream_output:
                out_lines = []
                err_lines = []

                def read_stdout():
                    for line in iter(stdout.readline, ""):
                        decoded = line
                        out_lines.append(decoded)
                        print(decoded, end="")

                def read_stderr():
                    for line in iter(stderr.readline, ""):
                        decoded = line
                        err_lines.append(decoded)
                        print(decoded, end="", file=sys.stderr)

                t1 = threading.Thread(target=read_stdout)
                t2 = threading.Thread(target=read_stderr)
                t1.daemon = True
                t2.daemon = True
                t1.start()
                t2.start()
                exit_code = channel.recv_exit_status()
                t1.join(timeout=1)
                t2.join(timeout=1)
                return ("".join(out_lines).strip(), "".join(err_lines).strip(), exit_code)

            exit_code = channel.recv_exit_status()
            return (
                stdout.read().decode("utf-8", errors="replace").strip(),
                stderr.read().decode("utf-8", errors="replace").strip(),
                exit_code,
            )
        except Exception as e:
            return "", str(e), 1

    def close(self):
        """Close SSH connection"""
        if self.client:
            self.client.close()


def load_config(config_file: str = 'servers.yml') -> dict:
    """Load server configuration from YAML"""
    with open(config_file, 'r') as f:
        return yaml.safe_load(f)

This gives you the foundation: a clean SSH client that can connect and run commands, plus a data structure to hold the metrics.

It defines:

A ServerMetrics dataclass to store health information per server (reachability, CPU, memory, disk, service status, and errors).
An SSHClient wrapper around Paramiko that:
- Handles SSH connections using RSA or Ed25519 keys (with optional passphrases)
- Falls back to SSH agents or default keys when possible
- Executes remote commands and captures output, errors, and exit codes (with optional streaming for long-running commands like apt upgrade)
- Manages connection lifecycle cleanly
A load_config helper to read server definitions from a YAML file.

For more Python fundamentals that help when reading or extending this script, for instance monitoring at the file-system level (e.g. watching config or log files), see Mastering File System Monitoring with Watchdog in Python.

Note on SSH key passphrase: If your private key has a passphrase you can (1) set the SSH_KEY_PASSPHRASE environment variable (recommended for a cron job), (2) add key_passphrase to a server entry in servers.yml, or (3) let the script prompt you interactively when needed.

Collecting Metrics

Now you'll add functions to gather actual metrics from each server. Add these functions to monitor.py:

def get_cpu_load(ssh: SSHClient) -> Optional[float]:
    """Get 1-minute load average from uptime command"""
    stdout, stderr, code = ssh.run_command('uptime')
    if code != 0:
        return None

    # Parse: "... load average: 0.52, 0.58, 0.59"
    try:
        load_str = stdout.split('load average:')[1].split(',')[0].strip()
        return float(load_str)
    except (IndexError, ValueError):
        return None


def get_memory_usage(ssh: SSHClient) -> Optional[float]:
    """Get memory usage percentage from free command"""
    stdout, stderr, code = ssh.run_command('free -m')
    if code != 0:
        return None

    # Parse output to get used/total
    try:
        lines = stdout.split('\n')
        mem_line = [l for l in lines if l.startswith('Mem:')][0]
        parts = mem_line.split()
        total = float(parts[1])
        used = float(parts[2])
        return (used / total) * 100
    except (IndexError, ValueError):
        return None


def get_disk_usage(ssh: SSHClient) -> Optional[float]:
    """Get root disk usage percentage from df command"""
    stdout, stderr, code = ssh.run_command('df -h /')
    if code != 0:
        return None

    # Parse: "Filesystem Size Used Avail Use% Mounted"
    try:
        lines = stdout.split('\n')
        data_line = lines[1]
        use_percent = data_line.split()[4].rstrip('%')
        return float(use_percent)
    except (IndexError, ValueError):
        return None


def get_service_status(ssh: SSHClient, service_name: str) -> str:
    """Check if a systemd service is active"""
    stdout, stderr, code = ssh.run_command(f'systemctl is-active {service_name}')

    if stdout == 'active':
        return 'active'
    elif code != 0:
        # Could be 'inactive', 'failed', or service doesn't exist
        return stdout if stdout else 'unknown'
    return 'unknown'


def collect_server_metrics(server_config: dict, thresholds: dict) -> ServerMetrics:
    """
    Connect to a server and collect all metrics
    """
    name = server_config['name']
    host = server_config['host']
    user = server_config['user']
    services = server_config.get('services', [])
    key_file = server_config.get('key_file')
    key_passphrase = server_config.get('key_passphrase') or os.environ.get('SSH_KEY_PASSPHRASE')

    metrics = ServerMetrics(name=name)
    ssh = SSHClient(host, user, key_file, key_passphrase)

    # Try to connect
    if not ssh.connect():
        metrics.reachable = False
        metrics.error = f"Could not connect to {host}"
        return metrics

    try:
        # Gather all metrics
        metrics.cpu_load = get_cpu_load(ssh)
        metrics.memory_percent = get_memory_usage(ssh)
        metrics.disk_percent = get_disk_usage(ssh)

        # Check services
        for service in services:
            metrics.services[service] = get_service_status(ssh, service)

    except Exception as e:
        metrics.error = str(e)

    finally:
        ssh.close()

    return metrics

This code collects basic server health metrics over SSH using standard Linux commands.

It:

Retrieves CPU load, memory usage, and disk usage by running uptime, free, and df remotely and parsing their output
Checks the status of systemd services via systemctl is-active
Wraps all results in a ServerMetrics object
Handles connection failures and parsing errors gracefully
Uses SSH keys (with optional passphrases) for secure, agentless access

If something fails, it returns None or a status string rather than crashing, which is important when monitoring multiple servers.

Terminal Dashboard with Rich

Now for the fun part, displaying our metrics in a beautiful terminal table. Add this to monitor.py:

from rich.console import Console
from rich.table import Table
from rich import box


def create_status_icon(value: Optional[float], threshold: float,
                       reverse: bool = False) -> str:
    """
    Create a colored status icon based on value vs threshold
    reverse=True means lower is better (e.g., disk usage)
    """
    if value is None:
        return "❓"

    if reverse:
        # Lower is better (disk, memory)
        if value < threshold * 0.7:
            return "✅"
        elif value < threshold:
            return "⚠️"
        else:
            return "❌"
    else:
        # Higher is better
        if value > threshold:
            return "❌"
        elif value > threshold * 0.7:
            return "⚠️"
        else:
            return "✅"


def format_metric_value(value: Optional[float], suffix: str = "",
                       decimals: int = 1) -> str:
    """Format a metric value for display"""
    if value is None:
        return "—"
    return f"{value:.{decimals}f}{suffix}"


def display_dashboard(metrics_list: List[ServerMetrics], thresholds: dict):
    """Display monitoring dashboard using Rich tables"""
    console = Console()

    # Create table
    table = Table(title="🖥️  Server Monitoring Dashboard",
                  box=box.ROUNDED,
                  show_header=True,
                  header_style="bold cyan")

    # Add columns
    table.add_column("Server", style="bold white", no_wrap=True)
    table.add_column("Status", justify="center")
    table.add_column("CPU Load", justify="right")
    table.add_column("Memory", justify="right")
    table.add_column("Disk", justify="right")
    table.add_column("Services", justify="left")

    # Add rows
    for metrics in metrics_list:
        # Overall status
        if not metrics.reachable:
            status = "🔴 Down"
        elif metrics.error:
            status = "⚠️ Error"
        else:
            status = "🟢 Up"

        # CPU with status icon
        cpu_display = format_metric_value(metrics.cpu_load)
        cpu_icon = create_status_icon(
            metrics.cpu_load,
            thresholds.get('cpu_load_factor', 2.0),
            reverse=False
        )
        cpu_cell = f"{cpu_icon} {cpu_display}"

        # Memory with status icon and percentage
        mem_display = format_metric_value(metrics.memory_percent, "%")
        mem_icon = create_status_icon(
            metrics.memory_percent,
            thresholds.get('memory_percent', 85),
            reverse=True
        )
        mem_cell = f"{mem_icon} {mem_display}"

        # Disk with status icon and percentage
        disk_display = format_metric_value(metrics.disk_percent, "%")
        disk_icon = create_status_icon(
            metrics.disk_percent,
            thresholds.get('disk_percent', 80),
            reverse=True
        )
        disk_cell = f"{disk_icon} {disk_display}"

        # Services status
        if metrics.services:
            service_items = []
            for svc, status_val in metrics.services.items():
                icon = "✅" if status_val == "active" else "❌"
                service_items.append(f"{icon} {svc}")
            services_cell = "\n".join(service_items)
        else:
            services_cell = "—"

        # Add row to table
        table.add_row(
            metrics.name,
            status,
            cpu_cell,
            mem_cell,
            disk_cell,
            services_cell
        )

    # Display
    console.print()
    console.print(table)
    console.print()

The rich library makes this surprisingly easy.

It:

Converts raw metric values into clear status icons (✅ ⚠️ ❌) based on configurable thresholds
Formats CPU, memory, and disk metrics for human-friendly display
Builds a styled Rich table showing per-server health, resource usage, and service status

Alert System

Monitoring without alerts is just anxiety. Let's add a simple alert system that can notify you via ntfy.sh. Add these functions:

def check_thresholds(metrics: ServerMetrics, thresholds: dict) -> List[str]:
    """
    Check if any metrics breach thresholds
    Returns list of alert messages
    """
    alerts = []

    if not metrics.reachable:
        alerts.append(f"🔴 {metrics.name}: Server unreachable")
        return alerts

    # CPU load
    if metrics.cpu_load and metrics.cpu_load > thresholds.get('cpu_load_factor', 2.0):
        alerts.append(
            f"⚠️ {metrics.name}: High CPU load ({metrics.cpu_load:.2f})"
        )

    # Memory
    if metrics.memory_percent and metrics.memory_percent > thresholds.get('memory_percent', 85):
        alerts.append(
            f"⚠️ {metrics.name}: High memory usage ({metrics.memory_percent:.1f}%)"
        )

    # Disk
    if metrics.disk_percent and metrics.disk_percent > thresholds.get('disk_percent', 80):
        alerts.append(
            f"⚠️ {metrics.name}: High disk usage ({metrics.disk_percent:.1f}%)"
        )

    # Failed services
    for service, status in metrics.services.items():
        if status != 'active':
            alerts.append(
                f"❌ {metrics.name}: Service '{service}' is {status}"
            )

    return alerts


def send_alert_ntfy(message: str, topic: str):
    """Send alert to ntfy.sh"""
    try:
        import requests
        url = f"https://ntfy.sh/{topic}"
        requests.post(url, data=message.encode('utf-8'))
    except Exception as e:
        print(f"Failed to send ntfy alert: {e}")


def send_alerts(alerts: List[str], config: dict):
    """Send alerts through configured channels"""
    if not alerts:
        return

    alert_config = config.get('alerts', {})
    message = "\n".join(alerts)

    # Send to ntfy
    if ntfy_topic := alert_config.get('ntfy_topic'):
        send_alert_ntfy(message, ntfy_topic)

    # Print to console
    console = Console()
    console.print(f"\n[bold red]🚨 Alerts:[/bold red]")
    for alert in alerts:
        console.print(f"  {alert}")
    console.print()

The alert logic is straightforward:

Evaluates collected server metrics against configurable thresholds
Detects unreachable servers, high CPU/memory/disk usage, and failed services
Generates human-readable alert messages
Sends alerts to ntfy.sh

You can easily add more channels (email, Slack, Telegram, PagerDuty) by adding similar send_alert_* functions.

To configure alerts, add this to your servers.yml:

alerts:
  ntfy_topic: "myapp-monitoring" # Optional

Putting It All Together

Now let's add the main function that ties everything together:

def main():
    """Main monitoring loop"""
    console = Console()

    try:
        # Load configuration
        config = load_config('servers.yml')
        servers = config.get('servers', [])
        thresholds = config.get('thresholds', {})

        if not servers:
            console.print("[red]No servers configured in servers.yml[/red]")
            return

        console.print(f"[cyan]Monitoring {len(servers)} server(s)...[/cyan]\n")

        # Collect metrics from all servers
        all_metrics = []
        all_alerts = []

        for server in servers:
            metrics = collect_server_metrics(server, thresholds)
            all_metrics.append(metrics)

            # Check for alerts
            alerts = check_thresholds(metrics, thresholds)
            all_alerts.extend(alerts)

        # Display dashboard
        display_dashboard(all_metrics, thresholds)

        # Send alerts if any
        if all_alerts:
            send_alerts(all_alerts, config)
        else:
            console.print("[green]✅ All systems healthy[/green]\n")

    except FileNotFoundError:
        console.print("[red]Error: servers.yml not found[/red]")
    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")


if __name__ == '__main__':
    main()

This the main entry point of the monitoring script:

Loads server and threshold configuration from servers.yml
Connects to each server and collects health metrics
Evaluates metrics against alert thresholds
Displays a Rich-based monitoring dashboard
Sends alerts if any issues are detected, otherwise reports a healthy state

That's it! You now have a complete monitoring script.

Let's test it:

python monitor.py

You should see a beautiful table showing the status of all your servers:

Monitoring 3 server(s)...

SSH key passphrase for id_rsa: 
SSH key passphrase for id_rsa: 
SSH key passphrase for id_rsa: 

                  🖥️  Server Monitoring Dashboard                  
╭───────────┬────────┬──────────┬──────────┬──────────┬───────────╮
│ Server    │ Status │ CPU Load │   Memory │     Disk │ Services  │
├───────────┼────────┼──────────┼──────────┼──────────┼───────────┤
│ prd-dokku │ 🟢 Up  │   ✅ 0.0 │ ✅ 17.8% │ ✅ 15.0% │ ✅ docker │
│ prd-ghost │ 🟢 Up  │   ✅ 0.5 │ ✅ 46.6% │ ❌ 32.0% │ ✅ docker │
│ prd-n8n   │ 🟢 Up  │   ✅ 0.0 │ ✅ 26.7% │ ⚠️ 19.0% │ ✅ docker │
╰───────────┴────────┴──────────┴──────────┴──────────┴───────────╯


🚨 Alerts:
  ⚠️ prd-ghost: High disk usage (32.0%)

And in this case, you should also receive a notification:

Remote Server Updates

Let's extend your monitor to handle server updates. Add these functions to monitor.py:

def run_server_update(ssh: SSHClient, stream_output: bool = True) -> dict:
    """
    Run apt update and upgrade on a server.
    stream_output=True prints apt progress so it doesn't look stuck (can take 5–15 min).
    Returns dict with results.
    """
    result = {
        'success': False,
        'output': '',
        'reboot_required': False,
        'error': None
    }

    # Run update and upgrade (stream output so user sees progress)
    # DEBIAN_FRONTEND=noninteractive avoids any prompts that could hang
    cmd = 'sudo DEBIAN_FRONTEND=noninteractive apt-get update && sudo DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1'
    stdout, stderr, code = ssh.run_command(cmd, stream_output=stream_output)

    result['output'] = stdout
    result['success'] = (code == 0)

    if code != 0:
        result['error'] = stderr or 'Update failed'
        return result

    # Check if reboot is required
    reboot_check_cmd = '[ -f /var/run/reboot-required ] && echo "yes" || echo "no"'
    reboot_stdout, _, _ = ssh.run_command(reboot_check_cmd)
    result['reboot_required'] = (reboot_stdout.strip() == 'yes')

    return result


def update_server(server_config: dict, notify_topic: Optional[str] = None) -> bool:
    """Update a single server and optionally send notification"""
    name = server_config['name']
    host = server_config['host']
    user = server_config['user']
    key_file = server_config.get('key_file')
    key_passphrase = server_config.get('key_passphrase') or os.environ.get('SSH_KEY_PASSPHRASE')

    console = Console()
    console.print(f"[cyan]Updating {name}...[/cyan]")
    console.print("[dim]Running apt update & upgrade (may take 5–15 min). Output streams below.[/dim]\n")

    ssh = SSHClient(host, user, key_file, key_passphrase)

    if not ssh.connect():
        console.print(f"[red]❌ Could not connect to {name}[/red]")
        return False

    try:
        result = run_server_update(ssh)

        if result['success']:
            reboot_msg = "⚠️ Reboot required" if result['reboot_required'] else "No reboot needed"
            console.print(f"[green]✅ {name} updated successfully[/green]")
            console.print(f"   {reboot_msg}")

            # Send notification
            if notify_topic:
                message = f"✅ {name} updated successfully\n{reboot_msg}"
                send_alert_ntfy(message, notify_topic)

            return True
        else:
            console.print(f"[red]❌ {name} update failed: {result['error']}[/red]")

            if notify_topic:
                message = f"❌ {name} update failed\n{result['error']}"
                send_alert_ntfy(message, notify_topic)

            return False

    finally:
        ssh.close()


def update_all_servers():
    """Update all configured servers"""
    config = load_config('servers.yml')
    servers = config.get('servers', [])
    notify_topic = config.get('alerts', {}).get('ntfy_topic')

    console = Console()
    console.print(f"\n[bold cyan]Updating {len(servers)} server(s)...[/bold cyan]\n")

    for server in servers:
        update_server(server, notify_topic)
        console.print()

This code adds remote server update automation to the monitoring tool:

Runs apt update and apt upgrade over SSH with streaming output so progress is visible (avoids looking stuck; updates can take 5–15 minutes)
Uses DEBIAN_FRONTEND=noninteractive so apt never waits for prompts
Detects whether a reboot is required after updates
Reports success or failure with clear terminal output
Sends update notifications via ntfy.sh

Now update the main() function to support an --update flag:

import sys

def main():
    """Main monitoring script"""
    console = Console()

    # Check for update mode
    if len(sys.argv) > 1 and sys.argv[1] == '--update':
        update_all_servers()
        return

    # ... rest of the monitoring code stays the same ...

You can update all your servers with:

python monitor.py --update

The script will connect to each server, run apt-get update && apt-get upgrade -y (with output streamed so you see progress), check if a reboot is needed, and send you a notification via ntfy:

Updating 3 server(s)...

Updating prd-dokku...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:
Hit:1 https://mirror.hetzner.com/ubuntu/packages noble InRelease
Hit:2 https://download.docker.com/linux/ubuntu noble InRelease
Hit:3 https://mirror.hetzner.com/ubuntu/packages noble-updates InRelease
Hit:4 https://mirror.hetzner.com/ubuntu/packages noble-backports InRelease
Hit:5 https://mirror.hetzner.com/ubuntu/security noble-security InRelease
Hit:6 https://packagecloud.io/dokku/dokku/ubuntu noble InRelease
Reading package lists...
Reading package lists...
Building dependency tree...
Reading state information...
Calculating upgrade...
The following upgrades have been deferred due to phasing:
  libldap-common libldap2 python-apt-common python3-apt
The following packages have been kept back:
  linux-image-virtual
0 upgraded, 0 newly installed, 0 to remove and 5 not upgraded.
✅ prd-dokku updated successfully
   ⚠️ Reboot required

Updating prd-ghost...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:

[...]

Running kernel seems to be up-to-date.

Restarting services...
 systemctl restart qemu-guest-agent.service

Service restarts being deferred:
 /etc/needrestart/restart.d/dbus.service
 systemctl restart getty@tty1.service
 systemctl restart serial-getty@ttyS0.service
 systemctl restart systemd-logind.service
 systemctl restart unattended-upgrades.service

No containers need to be restarted.

No user sessions are running outdated binaries.

No VM guests are running outdated hypervisor (qemu) binaries on this host.
✅ prd-ghost updated successfully
   ⚠️ Reboot required

Updating prd-n8n...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:
[...]

Running kernel seems to be up-to-date.

Restarting services...
 systemctl restart qemu-guest-agent.service

Service restarts being deferred:
 systemctl restart systemd-logind.service
 systemctl restart unattended-upgrades.service

No containers need to be restarted.

No user sessions are running outdated binaries.

No VM guests are running outdated hypervisor (qemu) binaries on this host.
✅ prd-n8n updated successfully
   No reboot needed

You will also get the notifications:

Wrapping Up

You don't have to choose between "no monitoring" and "full Prometheus/Grafana stack."

This middle ground, one Python script, SSH, and a few shell commands, gives you real visibility with minimal complexity.

The script you just built is production-ready for small setups. It's simple enough that you can debug it at 2 AM when something breaks, yet powerful enough to keep your servers healthy and your phone quiet. If you want to avoid scope-related bugs when extending it, Master Python Variable Scope is a concise, practical guide (LEGB, local vs global).

When you outgrow this approach (dozens of servers, complex dependencies, multiple teams), you'll know exactly what you need from a heavier system. Until then, keep it boring.

Download the complete script: https://github.com/nunombispo/server-monitoring-python-article

DEV Community: Developer Service

Your RAG Pipeline Is Lying to You

Why RAG Is Harder Than It Looks

The Concrete Example: RAG Over an Annual Report

Setup

Step 1: Ingest the PDF

Step 2: Query the Pipeline

Step 3: Run It

What Happens Without Evaluation

The Testing Layer

Stage 1: Chunk Quality

Stage 2: Retrieval Evaluation

Stage 3: End-to-End Answer Quality

The Monitoring Layer

The Organizational Gap

What Good Looks Like: A Minimal Viable Test Stack

Conclusion: The Trust Problem

Your Python Tool Needs Persistence - It Doesn't Need a Database Server

What TinyDB Actually Is

The Real Cost of Over-Engineering Internal Tools

A Concrete Example: CLI Task Manager

When to Stop Using TinyDB

Conclusion

Serving a Live Battery Dashboard from ESP32-C3 with Microdot and MicroPython

What You Will Build

Why Microdot?

System Overview

Hardware

Safety

Wiring Summary

Software: Project Structure

sensor.py - Hardware Reads

web.py - Routes and Inline UI

main.py - Boot and Server Start

What to Expect During a Charge Session

Conclusion

Python Decorators - The Three-Layer Pattern

Decorators Under the Hood

The identity problem

The Three Layers of a Parametrized Decorator

Why the extra layer exists

Closure capture

Building Your First Parameterized Decorator

Decorators You'll Actually Ship

@retry - handling flaky operations

@timer - measuring execution time

@require_role - protecting endpoints

Stacking Decorators

Conclusion

Build a Terminal Disk Analyzer in Python with Textual

What We're Building

Architecture

Step 1 - The Data Model and Scanner

Step 2 - Building the TUI with Textual

Step 3 - Non-Blocking Scan with @work

Step 4 - Rendering the Table

Step 5 - Delete with Confirmation

Step 6 - CLI Entrypoint with Typer

Conclusion

GitHub Actions for Python Projects - Automate Your Workflow from Day One

Key Concepts

Your First Workflow: Running Tests on Every Push

The Python Project

Creating the Workflow File

Line-by-Line Walkthrough

Pushing and Reading the Results

Making It Smarter

Linting with ruff

Caching pip Dependencies

Testing Across Multiple Python Versions

Type Checking with mypy (Optional)

Common Pitfalls

YAML Indentation Errors

Wrong Branch Name or Event Trigger

Forgetting requirements.txt

Hardcoding Credentials

Conclusion

Build an F1 Pit Wall Display with ESP32 CYD and OpenF1 API

nunombispo / CYD-OpenF1

CYD F1 Pit Wall Display

`sensor.py` - Hardware Reads

`web.py` - Routes and Inline UI

`main.py` - Boot and Server Start

`@retry` - handling flaky operations

`@timer` - measuring execution time

`@require_role` - protecting endpoints

Step 3 - Non-Blocking Scan with `@work`

Linting with `ruff`

Caching `pip` Dependencies

Type Checking with `mypy` (Optional)

Forgetting `requirements.txt`

`tinyfiledb/db.py`

`tinyfiledb/init.py`

Writing `pyproject.toml`