DEV Community: Developer Service

uv for Faster Teams, Fewer Environment Fires

Developer Service — Mon, 01 Jun 2026 06:42:54 +0000

Python dependency management is not a developer problem. It is a team productivity problem that shows up as slow CI, painful on-boarding, and a different tool in every repository.

If you run a Python shop with five to thirty engineers, a measurable slice of your payroll quietly funds work that never ships: waiting on dependency installs, re-explaining local setup to new hires, and arbitrating which package manager each team decided to use this quarter. None of it shows up on a road-map. All of it compounds.

uv is a Python package and project manager that replaces pip, pip-tools, and Poetry with a single, faster workflow.

This article is not a tutorial. It is a business case: what standardizing on uv costs, what it returns, and how to run a one-week pilot without stopping feature work. By the end, you will have a go/no-go framework and a minimal team policy you can paste into your engineering handbook today.

The Hidden Tax

Most of this work is invisible. There is no ticket for "waited on a slow install". There is no incident report for "new hire lost two days to a broken environment". It just disappears into the sprint - and the next sprint, and the one after that.

Those costs are not developer preferences. They are line items that nobody is measuring.

Slow CI: On many teams, the dependency install step is the longest part of a pull-request build. Every push pays that cost again. If installs take 10 minutes and you merge 20 PRs a week, you are buying a part-time engineer whose job is watching progress bars.

Onboarding drag: Picture a new hire on day one. Clone the repo, follow the README, hit a version mismatch, ask in Slack, get three different answers, open a doc that was true six months ago. By Thursday they might have a green test run. You have already spent senior time you could not spend on the road-map.

One of the simplest ways to make on-boarding real is to make the “first green test run” path as simple as a single copy-paste:

git clone git@github.com:your-org/your-repo.git
cd your-repo
uv sync
pytest

Tool sprawl: One service pins with pip-tools. Another uses Poetry because a contractor set it up. A third has a Makefile that wraps both. Your CTO job is not to pick the prettiest CLI; it is to know whether your organization can reproduce a production-like environment on demand. When every repo invents its own workflow, you are paying coordination tax on top of everything else.

None of this requires a dramatic failure. It is the steady leak that makes "we are too busy to change tools" sound reasonable, even while the leak itself burns the capacity you are protecting.

The fix is not a weekend rewrite of every repository. It is a policy decision: one default way to create environments, install dependencies, and lock what CI runs. That is where uv enters the conversation - not as hype about Rust, but as a way to stop re-litigating the same afternoon in every channel.

The next section is about what actually changes when pip is no longer the unnamed default on your team.

What Changes When Pip Isn't the Default

Your engineers will talk about uv in terms of speed. You should hear something else: one reproducible workflow across repositories.

At a high level, uv is a single tool-chain for creating virtual environments, resolving dependencies, and installing locked versions. Faster resolves matter because they shorten CI and local setup. Reproducible environments matter because they reduce "works on my machine" and make audits easier. One tool-chain matters because you stop paying the meeting tax every time a repo picks a different package manager.

That is the executive translation. You do not need to understand Rust or memorize flags to approve the direction. You need to know that the team can answer three questions the same way everywhere: how do we create an environment, how do we install what this commit expects, and how does CI prove we did?

You will still hear enthusiasm about performance. Treat it as supporting evidence, not the decision criterion. A startup CTO standardizes on uv when the organization needs a default, not when a benchmark chart wins an argument.

What stays the same is larger than what changes. Your application code is still Python. Your runtime version policy is still yours. Most CI pipelines keep their shape: checkout, install dependencies, run tests. You are usually swapping the install step and standardizing on a lock-file in git, not re-platforming the product.

What changes is the argument surface. Teams stop debating pip versus Poetry versus pip-tools on every new service. Code review stops treating "how we install" as a local choice. On-boarding docs can say one thing. When production misbehaves, you have fewer variables.

The practical policy looks boring on purpose: uv is the default for new Python work; existing repos adopt it when touched or during a time-boxed pilot; lock-files are required; CI uses a frozen install so what merged is what ran. For commands and platform-specific setup, the official uv documentation is the reference. Your handbook holds the rules.

The next section turns to evidence: which metrics to capture before and after so you can tell whether the pilot earned its half-day of attention.

The Productivity Case

Treat uv like any other productivity change: measure before, run a small pilot, then decide with data. If you cannot point to a number, you will default to opinion. And in tooling debates, opinions are loud.

Start with three metrics that map cleanly to cost:

CI dependency time: how long the install step takes on a cold cache (and how often caches miss). If installs are ten minutes and you run fifty builds a week, you are spending roughly eight engineer-hours a week just waiting for dependencies.
Cold clone-to-test time: on a clean machine, how long it takes to go from git clone to a green test run. This is onboarding time, but it also shows up every time a laptop is replaced, a new service is pulled down, or someone needs to reproduce a bug quickly.
Environment support load: count tickets or Slack threads tagged “env”, “pip”, “poetry”, “dependency”, “won’t install”. You do not need perfect accounting. A rough baseline is enough to see direction.

Now set expectations like a CTO, not a benchmark blog post. In a 5–30 engineer shop, you are usually chasing a combination of:

Faster installs (minutes back per build)
Fewer broken local environments (fewer interruptions)
Less variance between repos (less cognitive load)

Set a go/no-go before you start:

Decision: roll out to new repos if the pilot clears either threshold below
CI threshold: saves ≥ 10 minutes per PR in the dependency step
Time threshold: recovers ≥ 10 engineer-hours/month in avoided environment work

If you get even 1–2 minutes back per build on the dependency step and cut a couple of “why won’t this install” interruptions per week, the ROI can be real. The pilot cost is also real, but bounded: a half-day to wire one repo end-to-end and document one workflow. Do the math in hours, not gut feel.

Be honest about when this will not matter. If your Python footprint is small, your CI time is dominated by integration tests, or your dependency set rarely changes, switching tools will not move the needle. In those cases, keep your current setup and spend the time where it buys you more.

If the numbers suggest this is worth it, the next question is execution: how to adopt without turning it into a migration project.

Adopt Without a Migration Project

The easiest way to fail a tooling change is to turn it into a program. The fastest way to succeed is to treat it like an experiment with a tight box around it.

If your first reaction is “we don’t have bandwidth,” take that seriously. But also notice what it implies: you are already paying the bandwidth tax, just in a less visible form. The goal of a uv rollout is not purity. It is a measurable reduction in time spent on installs, environment drift, and repo-by-repo debate.

Run a one-repo, one-week pilot. Do not start with the cleanest service. Pick the repo that creates the most noise: the CI job everyone waits on, or the codebase that makes onboarding a Slack scavenger hunt. If uv can help there, it will help anywhere.

Keep the steps boring:

Add or standardize pyproject.toml and commit the lock-file you want CI to honor.
Swap the dependency install step in CI to use uv and make it “frozen” so CI fails if the lock-file is stale.
Write a short local workflow in the README: install uv, sync dependencies, run tests. One page. No tribal knowledge.

In GitHub Actions, the “swap the install step” part can be as small as this:

- name: Install dependencies (locked)
  run: uv sync --frozen

Define “done” before you start. Good definitions of done are observable:

CI is green and the dependency step is shorter (or at least more stable) on cold caches.
One teammate who did not touch the pilot can clone the repo and run tests on a clean machine in under X minutes.
You can point to a single default command for “get me to a working environment.”

Also define the rollback in one sentence. This lowers the political risk of trying something new. For example: “If the pilot burns more than half a day of engineer time or destabilizes CI, we revert the CI install step and keep the lockfile as-is.” You are buying information, not committing the company.

If the pilot works, resist the urge to immediately migrate every repo. The value of the pilot is not the migration - it is the evidence and the template. Capture both, then write a short policy that makes the choice automatic for every new repo from here on.

A Minimal uv Team Policy

If you want the benefits of uv without turning it into a prolonged debate, write down a minimal policy and enforce it in CI. The policy is not “everyone must love this tool.” The policy is “we have one default workflow and CI proves it.”

Here is a paste-ready version you can drop into your engineering handbook or an ADR.

Policy (minimal)

Default package workflow: For Python repos, uv is the default tool for creating environments and installing dependencies.
Lock-files are required: Dependency changes must update the lock-file and commit it to git.
CI is authoritative: CI installs dependencies from the lock-file in a frozen mode. If the lock-file is stale, CI fails.
Local workflow matches CI: The same “sync from lock-file” approach is the expected developer path.

If you want a single line to standardize on in CI, use:

uv sync --frozen

(Link to the official uv documentation for the exact installation steps and any platform-specific details. Your handbook should own the rules, not the flags.)

Onboarding checklist (one page)

Install uv
Clone the repo
Sync dependencies from the lock-file
Run tests

That is the whole point: fewer hidden steps and fewer Slack rituals.

Governance (light-touch)

Exceptions: Any repo can opt out temporarily, but the exception needs a named owner and an expiration date. Otherwise “temporary” becomes permanent.
New repos: Provide a template repo (or a copy-paste snippet for CI) that already follows the policy, so teams do not reinvent it.
Decision log: If your team uses ADRs, write a one-page “Why uv?” note that captures the motivation (cycle time, reproducibility) and the rollback trigger.

If you want to make that decision durable, use a tiny template:

## ADR: Standardize Python dependency workflow on uv

- **Status**: Proposed / Accepted / Rejected
- **Why now**: CI minutes + onboarding time + tool sprawl
- **Decision**: uv is the default; lockfiles required; CI uses `uv sync --frozen`
- **Pilot**: repo = `your-repo`; start = 2026-05-27; success metric = “save 10 minutes per PR in CI dependency step”
- **Rollback**: if CI becomes flaky or the pilot consumes > 0.5 day of engineer time, revert the CI install step and stop the rollout

This is what standardization buys you: fewer repo-by-repo decisions, fewer environment surprises, and a workflow you can explain in five minutes to a new hire. uv is not the only way to get there, but right now it is one of the lowest-friction ways for Python teams to make the default path fast and consistent.

Run the one-week pilot. Measure one metric you care about. If the numbers agree, expand. If they do not, stop and move on.

Conclusion

Python tooling debates are a proxy for a deeper question: does your team have a default, or does every service get to decide? uv does not answer that question for you. It lowers the cost of picking an answer.

The pattern is simple. You are already paying for slow installs, broken local environments, and tool sprawl - you are just paying in ways that do not show up on a dashboard. A half-day pilot on one noisy repo will tell you whether the numbers justify expanding. A short policy will tell your team what the default is from here on.

That is all this is: one default, enforced in CI, documented in your handbook. The benefit is not speed for its own sake. It is fewer conversations about tooling and more conversations about the work.

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

The 3am Pager - A Scrappy LLM Cost Monitor with Python and ntfy.sh

Developer Service — Tue, 26 May 2026 06:54:16 +0000

You shipped an LLM feature last quarter. The demo worked, the stakeholders were happy, and the model output was good enough to put in front of users. Three months in, your bill is climbing, and your board is asking what your gross margin looks like by feature.

You open the provider dashboard. It shows you a monthly total. That's it.

But here is the uncomfortable question: do you actually know which of your features is profitable? Not "are we under budget this month", but which features, which users, and which prompt shapes are eating your margin - and would you know if one of them was looping at 3am, burning through your runway one token at a time?

If you can't answer that with a query, you don't have a monitoring system. You have a credit card statement and a hope.

This article is for the ones building an LLM product and want to know what it actually costs you - per feature, per user, per request - without paying for Datadog or wiring up OpenTelemetry. We'll walk through a Python wrapper (Anthropic in the example, the same pattern for any provider), a SQLite cost store, and a small watcher script that pages you via ntfy.sh before a bad night becomes a bad month.

All code in this article is available in the companion repository: github.com/nunombispo/llm-costs.

The 3am Story

Friday afternoon. You merge a small prompt change for your busiest feature - one extra sentence in the system prompt, asking the model to show its reasoning before answering. CI is green. Your eval suite passes. You ship it and close your laptop for the weekend.

The change does exactly what you asked. Every response now includes a reasoning section. Output tokens jump by 10×. Your bill, which usually accumulates at a few hundred dollars a night, starts accumulating at a few thousand. Saturday at 3am, your cost-per-request has tripled. No pager fires. No alert exists.

Monday morning. You open the provider Dashboard to check usage and notice the provider invoice line: $4,200 - roughly 12% of your monthly runway, spent while everyone slept. It shows you a single number: $4,200. It does not show you which feature spent it, which users triggered it, or whether one bad request shape was responsible for half the total. You post in #engineering: "did anything change this weekend?". Three replies, all variations of "I don't think so". You grep your logs. There are 380,000 log lines. None of them include token counts. Three coffees in, you have a number and no story.

This did not have to be a Monday-morning discovery. The change went out Friday afternoon. By Saturday at 3am, three log lines and one SQL query would have known. Not the bill - the system. A pager at 3am is bad. A pager at 3am means someone is looking. The actual horror is the silence: no pager, no email, no signal, just a bill that arrived after the damage was already done.

A bill is not an alert. It is a receipt.

Why Total Spend Lies to You

Your provider dashboard gives you one number: total spend. That number answers one question - "did we spend money?" - and no others. It cannot tell you which features are profitable, which users are burning through your compute, or how much your average request cost has grown since the last prompt change. You need all three to run a margin-positive LLM product.

The dashboard gives you none of them:

Per-feature cost. Two features at $3k/month each look identical on the bill - but one serves 50,000 paying users at $0.06 per use, and the other serves 200 free-tier users at $15 per use. One is your business. The other is a leak. Total spend hides which is which.
Per-user cost. In almost every LLM product I've seen, roughly 0.5% of users account for 40% of spend. Without per-user attribution, you cannot decide whether to rate-limit them, upsell them, or remove them. You just watch the number climb.
Per-request cost shape. Your average request has quietly grown from 1,200 tokens to 8,400 tokens since the last prompt change - the new system prompt is three pages long and gets prepended to every call. The bill does not show this until the end of the month, when the damage is already done.

Total spend is not a monitoring signal. It is a lagging confirmation that something already went wrong.

Here is what to monitor instead.

Three Alerts That Cover 95% of Disasters

You don't need ten alerts. You need three. Three signals cover the failure modes that will actually cost you money before anyone notices - not the long tail of edge cases, but the ones that show up in your bill at the end of the month and in your Slack on Monday morning.

Get these three working before you build anything else:

Per-user spend rate. A user finds a way to loop your agent - a recursive prompt, an autoplay feature nobody rate-limited, a script someone wrote to use your app as a cheap API. Without this alert, you find out at the end of the month. With it, you find out at the end of the hour. Fires when any single user spends more than $5 in 60 minutes. Does not catch slow spend distributed across many users - that is what alert 2 is for.
Per-feature daily spend. You shipped a prompt change on Friday. By Sunday it had tripled the token count on your busiest feature - a longer reasoning block, an extra sentence prepended to every call 50,000 times, a context window that quietly doubled. This is the alert that would have fired before your Monday morning Slack message. Fires when a feature's daily spend exceeds its rolling 7-day p95 by 50%. The p95 threshold needs at least 30 days of baseline data to be meaningful. Fall back to a $50/day hard ceiling for the first month. Does not catch gradual creep over weeks - that is what the SQL query in the next point is for.
Single-request cost ceiling. One request. $4. You found out from the bill. The cause is one of three things: a prompt-injection attempt that asked your model to write a 30,000-token essay, a tool-calling loop that did not terminate, or a context that grew unbounded across turns. Any of them trips this. Fires when any single request costs more than $0.50. Does not catch many cheap requests adding up - see alert 2.

Start with these thresholds and move them when you have a reason to:

The point of an alert is not to be quiet. It is to be the first thing that knows.

The Wrapper

The wrapper is a single function that sits between your code and the Anthropic API. Your existing call to client.messages.create() becomes tracked_create(client, feature="...", user_id="...", ...). Every other argument passes through unchanged. The response object is identical. The only difference is a SQLite row written before the function returns.

The only dependencies are the Anthropic SDK and requests:

pip install anthropic requests

And the script:

# costs.py
import sqlite3, time, uuid
import anthropic

DB_PATH = "costs.db"

# Prices in USD per 1,000 tokens — see "The Pricing Table" for the full dict
# Verify current prices at: https://www.anthropic.com/pricing
PRICING = {
    ("anthropic", "claude-opus-4-7"):   {"input": 0.005,   "output": 0.025},
    ("anthropic", "claude-sonnet-4-6"): {"input": 0.003,   "output": 0.015},
    ("anthropic", "claude-haiku-4-5"):  {"input": 0.001, "output": 0.005},
}

def _init_db() -> None:
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS cost_events (
                id TEXT PRIMARY KEY, ts TEXT NOT NULL,
                provider TEXT NOT NULL, model TEXT NOT NULL,
                feature TEXT NOT NULL, user_id TEXT NOT NULL,
                prompt_tokens INTEGER NOT NULL, completion_tokens INTEGER NOT NULL,
                cost_usd REAL NOT NULL, latency_ms REAL NOT NULL
            )
        """)

_init_db()  # runs once on import — safe for SQLite, no migration script needed

def _cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    rates = PRICING.get(("anthropic", model))
    if rates is None:
        raise ValueError(
            f"No pricing entry for anthropic/{model}. Add it to PRICING before deploying."
        )
    return (prompt_tokens / 1000) * rates["input"] + (completion_tokens / 1000) * rates["output"]

def tracked_create(client: anthropic.Anthropic, *, feature: str, user_id: str, **kwargs):
    """Drop-in for client.messages.create. Logs cost and latency to SQLite."""
    t0 = time.perf_counter()
    response = client.messages.create(**kwargs)
    latency_ms = round((time.perf_counter() - t0) * 1000, 1)

    model = kwargs["model"]
    cost = _cost(model, response.usage.input_tokens, response.usage.output_tokens)

    with sqlite3.connect(DB_PATH) as conn:
        conn.execute(
            "INSERT INTO cost_events VALUES (?,?,?,?,?,?,?,?,?,?)",
            (str(uuid.uuid4()), time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
             "anthropic", model, feature, user_id,
             response.usage.input_tokens, response.usage.output_tokens, cost, latency_ms),
        )
    return response

This is the entire instrumentation layer. Three functions, one file - Anthropic SDK and SQLite3 (stdlib). The watcher adds requests for ntfy. The rows it writes are what the watcher, the SQL queries, and the alerts all read from.

Switching providers means changing three lines: the import, response.usage.input_tokens (and output tokens) to whatever your provider's SDK calls input tokens, and the PRICING dict keys. The pattern is identical.

The Pricing Table

Every LLM provider publishes a pricing page. None of them expose a pricing API. The source of truth is HTML, and it changes without notice when a model is updated, deprecated, or re-priced for a new tier. There is no authoritative feed to subscribe to, no webhook to catch the change. The moment you deploy, your table starts drifting from reality.

The pragmatic answer is not to fight this. It is to own it explicitly: one dict, one file, one dated comment, and a function that throws an error the moment you call a model that isn't in it.

# costs.py — PRICING dict
# Prices in USD per 1,000 tokens. Last verified: 2026-05-19
# Source: https://www.anthropic.com/pricing — check before each deploy
PRICING = {
    ("anthropic", "claude-opus-4-7"):   {"input": 0.005,   "output": 0.025},
    ("anthropic", "claude-sonnet-4-6"): {"input": 0.003,   "output": 0.015},
    ("anthropic", "claude-haiku-4-5"):  {"input": 0.001, "output": 0.005},
}

Adding a different provider is four lines - same shape, different key prefix:

# Add inside your PRICING dict (source: https://openai.com/api/pricing):
PRICING.update({
    ("openai", "gpt-5.5"):     {"input": 0.005,  "output": 0.030},
    ("openai", "gpt-5.4-mini"):{"input": 0.00075, "output": 0.0045},
})

The fail-loud check - already in _cost() in costs.py - raises a ValueError the moment your code tries to log a cost for a model not in the table. You find out in CI or on first run, not at the end of the month when you notice the cost column is suspiciously round. A silent zero is worse than a missing alert.

One caveat: this table covers the standard pay-per-token case. It does not cover prompt caching (Anthropic's cache read tokens cost roughly 10% of normal input price), fine-tuned model rates, or batch API discounts. Those all have separate rates and separate line items on your bill. Add them when they apply to your stack. Build the table you can maintain, not the one that is complete.

A pricing table you maintain is more useful than a pricing API you trust.

The Watcher

The watcher needs somewhere to send alerts. ntfy.sh is an open-source pub/sub notification service. You publish a message to a URL. Anyone subscribed to that same URL gets a push notification on their phone. No accounts. No API keys. No dashboard to configure. One HTTP POST per alert, and the free hosted version at ntfy.sh handles the rest.

To set it up: pick any string as your topic name - it becomes the URL path (https://ntfy.sh/your-topic), install the ntfy app on your phone, and subscribe to that same name. Three minutes. PagerDuty does not.

Your topic name is effectively a bearer token. Anyone who knows it can subscribe. Keep it out of your repo; NTFY_TOPIC in watcher.py should come from an environment variable in production.

# watcher.py
import sqlite3, time, os, requests
from costs import DB_PATH

NTFY_TOPIC = os.getenv("NTFY_TOPIC", "https://ntfy.sh/your-secret-topic-name")  # set NTFY_TOPIC env var in production
POLL_INTERVAL = 60  # seconds

# Adding a new alert: add one dict to this list. Nothing else changes.
ALERTS = [
    {
        "name":    "per_user_hourly",
        "query":   """
            SELECT user_id AS entity, ROUND(SUM(cost_usd), 4) AS value
            FROM cost_events WHERE ts >= datetime('now', '-1 hour')
            GROUP BY user_id HAVING value > 5.0
        """,
        "message": lambda r: f"User {r['entity']} burned ${r['value']:.2f} in the last hour",
    },
    {
        "name":    "per_feature_daily",
        "query":   """
            SELECT feature AS entity, ROUND(SUM(cost_usd), 4) AS value
            FROM cost_events WHERE ts >= datetime('now', '-1 day')
            GROUP BY feature HAVING value > 50.0
        """,
        "message": lambda r: f"Feature {r['entity']} spent ${r['value']:.2f} today - check for prompt regressions",
    },
    {
        "name":    "per_request_ceiling",
        "query":   """
            SELECT id AS entity, ROUND(cost_usd, 4) AS value, feature
            FROM cost_events WHERE cost_usd > 0.50 AND ts >= datetime('now', '-1 hour')
        """,
        "message": lambda r: f"Single request ${r['value']:.4f} on {r['feature']}",
    },
]

def _init_db() -> None:
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS alerts_sent (
                alert_type TEXT NOT NULL, entity_id TEXT NOT NULL, sent_at TEXT NOT NULL
            )
        """)

_init_db()

def _already_sent(conn, alert_type: str) -> bool:
    return conn.execute(
        "SELECT 1 FROM alerts_sent WHERE alert_type=? AND entity_id=? "
        "AND sent_at >= datetime('now', '-1 hour')",
        (alert_type),
    ).fetchone() is not None

def _notify(title: str, message: str) -> None:
    requests.post(NTFY_TOPIC, data=message.encode(), headers={"Title": title}, timeout=5)

def run_alerts() -> None:
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row
        for alert in ALERTS:
            for row in conn.execute(alert["query"]).fetchall():
                entity_id = str(row["entity"])
                if _already_sent(conn, alert["name"]):
                    continue
                _notify(alert["name"], alert["message"](row))
                conn.execute(
                    "INSERT INTO alerts_sent VALUES (?,datetime('now'))",
                    (alert["name"]),
                )

if __name__ == "__main__":
    print(f"Watching {DB_PATH}. Polling every {POLL_INTERVAL}s.")
    while True:
        run_alerts()
        time.sleep(POLL_INTERVAL)

The ALERTS list is the key design decision. Each alert is a name, a SQL query, and a message template. Adding a fourth alert is five lines. The _already_sent check deduplicates on a one-hour window per alert.

The per-feature query uses the $50/day hard-ceiling fallback from the Three Alerts section, not the full p95 version. Once you have the history, the companion repo has the full version.

If the watcher dies, systemd with Restart=always or a Docker container with restart: always brings it back. That is enough.

Example of the alerts:

Two scripts. One database. Your phone rings before your bill does.

The One SQL Query You Actually Need

Your cost_events table already knows which feature is bleeding margin. You don't need a dashboard to ask it.

SELECT
  feature,
  COUNT(*)                                                           AS calls,
  ROUND(SUM(cost_usd), 2)                                           AS total_usd,
  ROUND(AVG(cost_usd), 4)                                           AS avg_call_usd,
  ROUND(SUM(prompt_tokens + completion_tokens) * 1.0 / COUNT(*), 0) AS avg_tokens
FROM cost_events
WHERE ts >= datetime('now', '-7 days')
GROUP BY feature
ORDER BY total_usd DESC;

Run this in your terminal with sqlite3 costs.db. It tells you which feature is spending the most, how many calls it made, what each call costs on average, and how many tokens each call consumes on average. Five columns. One screen. The thing your board is actually asking about.

Example output:

============================================================
Per-feature (7 days)
============================================================
feature | calls | total_usd | avg_call_usd | avg_tokens
------------------------------------------------------------
agent_loop | 1 | 0.82 | 0.82 | 128000.0

What it does not tell you: per-user spend, how cost trended day-by-day, or cost-per-revenue ratio. Per-user spend and day-by-day trending are one-line variants of the same query. The third requires a join to a revenue table you don't have yet - and when you do, the structure is the same.

Per-user spend this week:

SELECT user_id, COUNT(*) AS calls, ROUND(SUM(cost_usd), 2) AS total_usd
FROM cost_events
WHERE ts >= datetime('now', '-7 days')
GROUP BY user_id ORDER BY total_usd DESC LIMIT 20;

Example output:

============================================================
Top users (7 days)
============================================================
user_id | calls | total_usd
------------------------------------------------------------
user_alice | 1 | 0.82

Daily cost by feature - the trend that shows you when a regression happened:

SELECT date(ts) AS day, feature, ROUND(SUM(cost_usd), 2) AS daily_usd
FROM cost_events
WHERE ts >= datetime('now', '-30 days')
GROUP BY date(ts), feature
ORDER BY day DESC, daily_usd DESC;

Example output:

============================================================
Daily by feature (30 days)
============================================================
day | feature | daily_usd
------------------------------------------------------------
2026-05-20 | agent_loop | 0.82

A dashboard is a SQL query someone else got paid to render.

When This Stops Being Enough

This article is not an argument against Datadog. It is an argument against Datadog on day one, before you have a problem Datadog solves. The watcher in this article is what you run until one of the four conditions below becomes true. Most teams never cross any of them.

This setup has four limits. When you hit one, you will know it - because the problem will be specific, not vague.

You sustain more than ~100 LLM calls/second. This is the easiest limit to hit and the easiest to fix. SQLite concurrent writes become the bottleneck first. Move cost_events to Postgres. The wrapper, the watcher, and the SQL queries are unchanged - only the connection string swaps.
You need cross-region or cross-cloud aggregation. A single SQLite file on a single box does not survive multi-region deployments. At this point you want OpenTelemetry semantic conventions and a real metrics backend - Prometheus + Grafana, Datadog, or whatever your platform team already runs.
You are a multi-tenant SaaS at compliance scale. Per-tenant cost attribution for billing is now a finance system, not a monitoring system. It has different reliability requirements, different audit trails, and a different team owning it. This article's setup is not the right foundation for that problem.
You need to alert on token-distribution shifts, not just dollar thresholds. When you start caring about p99 latency by model, prompt-injection signals in token shapes, or model degradation patterns, you are past what a 60-second polling watcher can detect reliably.

Graduate to the heavier stack when the heavier stack is solving a problem you actually have. Not before.

Conclusion: Before the 3am Page

Your LLM bill tells you one thing: how much you spent. It does not tell you which feature spent it, which users drove it, or whether a prompt change you shipped on Friday quietly tripled your cost-per-request over the weekend. The signal you actually need - per-feature, per-user, per-request - is already in your calls. You just have not been writing it down.

Two scripts and a SQLite file change that. tracked_create() writes a row every time your code calls the Anthropic API. The watcher reads those rows every 60 seconds and sends a push notification when any of the three thresholds break. The SQL query tells you which feature is bleeding margin. The whole thing runs on the same box as your application and costs nothing to operate.

Without it, you have a credit card statement and a hope.

Wire it up before the 3am page.

All code in this article is available in the companion GitHub repository: github.com/nunombispo/llm-costs. The repository includes the wrapper, the watcher, the pricing table, and the SQL queries discussed above.

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 Kevin Schneider from Pixabay

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

DEV Community: Developer Service

uv for Faster Teams, Fewer Environment Fires

The Hidden Tax

What Changes When Pip Isn't the Default

The Productivity Case

Adopt Without a Migration Project

A Minimal uv Team Policy

Policy (minimal)

Onboarding checklist (one page)

Governance (light-touch)

Conclusion

The 3am Pager - A Scrappy LLM Cost Monitor with Python and ntfy.sh

The 3am Story

Why Total Spend Lies to You

Three Alerts That Cover 95% of Disasters

The Wrapper

The Pricing Table

The Watcher

The One SQL Query You Actually Need

When This Stops Being Enough

Conclusion: Before the 3am Page

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

`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`