Forem: Developer Service

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

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

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

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

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

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

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

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

What You Will Build

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

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

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

This is the running dashboard while charging a 18650 battery:

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

Why Microdot?

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

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

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

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

from microdot import Microdot

app = Microdot()

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

app.run(port=80)

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

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

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

Nothing too exotic. That's the point.

System Overview

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

Data flow:

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

Power and charge flow:

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

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

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

Hardware

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

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

Safety

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

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

Wiring Summary

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

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

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

For reference, the full wiring diagram:

This is how it looks like on a breadboard:

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

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

Software: Project Structure

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

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

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

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

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

`sensor.py` - Hardware Reads

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

_ina = None

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

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

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

A few deliberate choices here:

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

`web.py` - Routes and Inline UI

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

from microdot import Microdot, Response
from sensor import read_metrics

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

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

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

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

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

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

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

`main.py` - Boot and Server Start

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

import time
import network
from web import app


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

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

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

main()

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

What to Expect During a Charge Session

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

Constant Current phase (early):

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

Constant Voltage phase (late):

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

Let' see the CC phase in action:

And if it gets disconnect for some reason:

Conclusion

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

In this build you now have:

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

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

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

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

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

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

Python Decorators - The Three-Layer Pattern

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

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

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

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

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

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

Decorators Under the Hood

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

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

Here's the simplest possible decorator:

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

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

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

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

greet = shout(greet)

That's the whole trick.

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

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

Visually, the transformation looks like this:

The identity problem

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

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

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

The fix is functools.wraps:

import functools

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

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

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

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

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

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

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

The Three Layers of a Parametrized Decorator

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

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

That gives you three layers total.

Here's the structure, stripped to its bones:

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

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

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

Which collapses to the familiar one-liner:

greet = outer(param)(greet)

The diagram maps this exactly:

Why the extra layer exists

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

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

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

Closure capture

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

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

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

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

A concrete way to see this:

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

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

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

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

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

Building Your First Parameterized Decorator

Time to put the three-layer pattern to work.

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

Here's the goal:

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

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

And here's the full implementation:

import functools

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

Let's walk through each layer.

Layer 1 - repeat(times)

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

Layer 2 - decorator(fn)

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

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

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

A note on return value

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

Checking it works

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

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

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

Decorators You'll Actually Ship

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

Here are three parameterized decorators that solve real problems.

`@retry` - handling flaky operations

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

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

import time
import functools

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

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

A few design decisions worth noting.

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

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

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

`@timer` - measuring execution time

Useful during development and in production monitoring.

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

import time
import functools

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

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

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

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

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

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

`@require_role` - protecting endpoints

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

import functools
from functools import wraps

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

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

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

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

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

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

Stacking Decorators

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

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

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

But the order matters, and it trips people up.

Application order is bottom-up

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

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

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

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

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

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

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

Swap the order and it measures only a single attempt:

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

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

Debugging stacked decorators

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

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

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

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

import inspect

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

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

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

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

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

Conclusion

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

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

Three things worth carrying forward.

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

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

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

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

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

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

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

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

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

Build a Terminal Disk Analyzer in Python with Textual

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

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

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

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

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

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

What We're Building

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

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

Here's what it looks like in action:

Confirmation modal to delete a file/directory:

Notification after deleting a file/directory:

Architecture

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

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

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

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

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

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

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

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

The only dependencies needed, from the requirements.txt:

textual
typer

Step 1 - The Data Model and Scanner

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

You represent every file and directory as a DiskEntry dataclass:

from dataclasses import dataclass, field
from pathlib import Path

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

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

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

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

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

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

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

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

See the full implementation in pydusk.py.

Step 2 - Building the TUI with Textual

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

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

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

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

class DuskApp(App):
    TITLE = "pydusk"

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

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

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

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

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

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

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

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

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

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

from textual import on
from textual.widgets import DataTable

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

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

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

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

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

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

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

from textual import work

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

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

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

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

Step 4 - Rendering the Table

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

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

from rich.text import Text

BAR_WIDTH = 20

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

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

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

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

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

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

Step 5 - Delete with Confirmation

Deleting from a TUI needs a confirmation step.

Textual's ModalScreen is the right tool.

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

from textual.screen import ModalScreen

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

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

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

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

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

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

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

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

Step 6 - CLI Entrypoint with Typer

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

import typer

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

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

if __name__ == "__main__":
    cli()

Examples of usage:

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

Conclusion

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

Not bad for a single file with two dependencies.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Key Concepts

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

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

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

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

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

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

Here's how they fit together:

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

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

Your First Workflow: Running Tests on Every Push

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

The Python Project

Start with a simple project structure:

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

calculator.py contains a function to test:

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

test_calculator.py has the test:

from calculator import add

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

And requirements.txt lists your test dependency:

pytest

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

Creating the Workflow File

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

Here's the complete workflow:

name: CI

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

jobs:
  test:
    runs-on: ubuntu-latest

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

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

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

      - name: Run tests
        run: pytest

Line-by-Line Walkthrough

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

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

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

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

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

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

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

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

Pushing and Reading the Results

Commit the workflow file and push it to main:

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

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

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

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

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

Making It Smarter

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

Linting with `ruff`

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

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

pytest
ruff

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

      - name: Lint with ruff
        run: ruff check .

That's all it takes.

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

Caching `pip` Dependencies

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

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

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

Here's what each part does:

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

On the first run, the cache is created.

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

Testing Across Multiple Python Versions

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

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

  test:
    runs-on: ubuntu-latest

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

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

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

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

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

Type Checking with `mypy` (Optional)

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

Add it to requirements.txt:

pytest
ruff
mypy

And add a step after linting:

      - name: Type check with mypy
        run: mypy .

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

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

Common Pitfalls

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

YAML Indentation Errors

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

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

Wrong Branch Name or Event Trigger

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

Forgetting `requirements.txt`

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

Hardcoding Credentials

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

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

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

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

Conclusion

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

nunombispo / CYD-OpenF1

CYD F1 Pit Wall Display

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

Hardware

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

Dependencies

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

Configuration

Edit the top of main.py:

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

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

View on GitHub

One Endpoint, One Request

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

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

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

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

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

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

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

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

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

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

Part 1: The Aggregator

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

Positions

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

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

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

Tyres

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

Gaps

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

Formatting and total laps

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

The Response

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

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

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

Part 2: The Display

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

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

Layout

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

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

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

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

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

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

Rendering

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

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

Mapping the response: parse_pitwall

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

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

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

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

Part 3: The Main Loop

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Result

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

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

What works well

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

What doesn't

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

What's Next

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

Three directions that fit that pattern:

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

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

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

Conclusion

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

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

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

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

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

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

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

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

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

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

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

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

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

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

By the end you will have:

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

The Project: A Tiny File Database

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

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

Project Structure

This is the project structure you will use:

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

`tinyfiledb/db.py`

The main project file that contains the tiny database implementation:

import json
import uuid
from pathlib import Path


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

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

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

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

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

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

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

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

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

`tinyfiledb/init.py`

Expose only what users need:

from .db import FileDB

__all__ = ["FileDB"]

Quick Usage

You can test it with this quick usage example:

from tinyfiledb import FileDB

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

Should return:

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

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

Structuring Your Package

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

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

The project uses the flat layout:

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

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

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

We already did that above.

Writing `pyproject.toml`

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

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

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

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

What each part does:

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

Building the Distribution

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

Install the build tool

pip install build

Run the build

From the directory that contains pyproject.toml:

python -m build

You'll get:

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

A dist/ folder will appear:

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

Publishing to PyPI

Use TestPyPI first.

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

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

Install twine (upload tool):

pip install twine

TestPyPI

First make sure your account and token are ready:

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

Upload:

twine upload --repository testpypi dist/*

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

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

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

Test install in a fresh venv:

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

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

Then:

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

If that works, you're ready for PyPI.

PyPI (production)

Make sure your account and token are ready:

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

Upload:

twine upload dist/*

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

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

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

Anyone can install it with:

pip install tinyfiledb

Installing and Using Your Package

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

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

pip install tinyfiledb

Check metadata:

pip show tinyfiledb

Will show:

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

You can do a quick test:

from tinyfiledb import FileDB

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

Which returns:

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

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

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

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

Full source code available at:

nunombispo / tinyfiledb

tinyfiledb

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

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

Install

pip install tinyfiledb

Usage

from tinyfiledb import FileDB

db = FileDB("mydata.json")

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

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

API

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

View on GitHub

Conclusion

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

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

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

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

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

Want to Go Deeper?

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

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

Now go ship something.

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

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

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

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

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

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

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

“How are user passwords stored?”

That’s when it hits you: plain text.

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

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

Why This Matters to You

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

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

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

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

Getting Started

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

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

pip install cryptography

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

Hashing Passwords: Your First Line of Defense

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

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

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

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

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

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

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

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

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

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

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

Running the example produces an output similar to this:

Salt: 88550cd4b4fac3e5153f9167733a82c0
Hashed: 6555686a7ce17e8fb81de3cb2f88809d81b410e572b039ece7526efd8e864b1c

Encrypting Data: Keeping Secrets Safe

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

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

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

from cryptography.fernet import Fernet

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

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

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

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

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

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

Running the example produces an output similar to this:

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

Public-Private Key Encryption: The Magic of Two Keys

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

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

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

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

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

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

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

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

Running the example produces an output similar to this:

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

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

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

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

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

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

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

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

Building Secure APIs in Production?

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

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

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

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

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

Where to Go From Here

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

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

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

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

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

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

Start Building Secure Applications Today

Cryptography isn't just for security experts anymore.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

The Motive: Why This Matters

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

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

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

Why Not Manual Backups?

Manual backups tend to fall apart for three simple reasons:

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

The Streaming Advantage

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

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

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

The Architecture

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

System Flow

Requirements & Dependencies

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

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

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

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

Access Setup

The pipeline needs two things configured outside the script:

Hetzner Storage Box - SFTP access:

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

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

AWS - access keys for S3:

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

The Code

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

System Components Overview

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

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

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

Configuration Loading

from dotenv import load_dotenv
load_dotenv()

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

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

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

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

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

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

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

State Database Initialization

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

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

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

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

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

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

Smart Duplicate Detection

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

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

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

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

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

State Persistence After Upload

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

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

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

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

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

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

SFTP Directory Walker with Path Exclusions

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

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

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

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

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

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

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

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

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

Streaming Upload with Progress

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

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

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

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

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

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

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

Main Orchestration

def main():
    conn = init_db()

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

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

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

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

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

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

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

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

Why it’s built this way:

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

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

This keeps debugging straightforward and failure modes predictable. At the end, a summary report logs total files uploaded, skipped, bytes transferred, and total runtime.

Progress Tracking with Metrics

metrics = {
    "uploaded": 0,
    "skipped": 0,
    "bytes_uploaded": 0,
    "start_time": time.time(),
}

# ... at the end:
duration = time.time() - metrics["start_time"]
log.info("==== SUMMARY ====")
log.info(f"Uploaded files : {metrics['uploaded']}")
log.info(f"Skipped files  : {metrics['skipped']}")
log.info(f"Bytes uploaded : {metrics['bytes_uploaded']:,}")
log.info(f"Duration (sec) : {int(duration)}")

A global metrics dictionary tracks operational statistics throughout execution: files uploaded, files skipped, total bytes transferred, and start time for duration calculation. At the end, a summary report is logged.

A simple dictionary tracks operational metrics. The callback in upload_stream() accumulates bytes, while the main loop increments counters. At the end, you get a summary:

==== SUMMARY ====
Uploaded files : 42
Skipped files  : 1,583
Bytes uploaded : 15,234,567,890
Duration (sec) : 340

This makes it easy to track incremental efficiency over time. After the first full backup, subsequent runs should show high skip counts and low upload counts, exactly what you want. These metrics help you verify the script is working correctly and monitor bandwidth usage.

Cost Analysis

Using a hybrid approach, Hetzner for active backups and Amazon S3 Glacier Deep Archive for cold storage, gives enterprise-grade durability at a fraction of standard cloud costs.

Monthly Storage Costs (10TB Example)

Service	Cost	Role
Hetzner Storage Box BX31 (10TB)	€28.86	Active, frequently accessed backups
AWS S3 Glacier Deep Archive (10TB)	$10.10	Cold, geo-redundant disaster recovery
Total	€39.96 (~$44)	Hybrid storage solution
Per-TB Cost	€4.00/TB (~$4.40/TB)	Unit economics

Comparison:

AWS S3 Standard: $230/month
Backblaze B2: $50/month (+ egress fees)
Google Cloud Archive: $12/month (higher retrieval costs)

Hybrid tiering delivers reliability at a fraction of S3 Standard pricing.

Transfer & API Costs

Hetzner → AWS: Free egress & ingress
S3 PUT Requests: ~$5 for 100,000 files initially; ~$0.25/month for incremental changes (~5%)
Impact: Negligible compared to storage savings

Retrieval Costs (Deep Archive Caveat)

Retrieval: $0.02/GB
Restoration: 12–48 hours
Temporary storage in S3 Standard: $0.023/GB/month

Example: Restoring 1TB costs ~$20 in retrieval fees plus ~$23/month if kept temporarily. Deep Archive is designed for “write-once, rarely-read” disaster recovery.

This hybrid setup gives the best of both worlds: affordable, accessible active storage and ultra-durable, geo-redundant archival backup, without breaking the bank.

Conclusion

Backing up from a Hetzner Storage Box to Amazon S3 Glacier Deep Archive gives you a powerful, cost-effective solution with:

Affordability: ~$5/month per TB using hybrid tiered storage
Durability: 11-nines (99.999999999%) with built-in geo-replication
Efficiency: Incremental, streaming transfers with zero local disk usage
Reliability: Interrupt-safe, state-persistent, fully logged operations
Automation: Fully hands-off, schedule via cron, Task Scheduler, or any job runner

Whether for personal data, compliance, or enterprise disaster recovery, this pipeline delivers enterprise-grade reliability at a fraction of typical cloud costs.

The code is production-ready, tested, and designed to run unattended. By combining Hetzner’s affordability with AWS Glacier Deep Archive’s durability, you get the best of both worlds.

Full code and documentation: https://github.com/nunombispo/hetzner-to-aws-s3-glacier-backups

More on Python, Hetzner and automation: Hetzner Storage Box backup, Django on Hetzner + Dokku) and Python One-Liners.

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

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

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

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

My Experience Building a URL Shortener Like TinyURL

Developer Service — Mon, 16 Feb 2026 07:53:57 +0000

You know that moment when you're sharing a link to your latest blog post, and the URL looks like https://developer-service.blog/build-your-own-low-cost-cloud-backup-with-hetzner-storage-boxes/?

Yeah, that's not going to fit nicely anywhere.

I faced this problem constantly while managing my blog, sharing free guides, and tracking GitHub repository statistics across different platforms.

Twitter's character limits, newsletter formatting issues, clean QR codes for printed materials, every platform seemed to hate my long, descriptive URLs.

Sure, I could use bit.ly or TinyURL, but then I'd be giving up control over my links, paying for premium features, and most importantly, losing valuable analytics data.

So I did what any self-respecting developer would do: I spent a weekend building my own URL shortener. And you know what? It was way more interesting than I expected.

Full source code link at the end of the article

What Exactly Is a URL Shortener?

At its core, a URL shortener is beautifully simple: it takes a long URL and maps it to a short, memorable code. When someone visits the short URL, they're redirected to the original destination.

Long URL:  https://developer-service.blog/build-your-own-low-cost-cloud-backup-with-hetzner-storage-boxes/
           ↓
Short URL: https://myurl.app/0001ab
           ↓
Redirect:  302 Found → Original URL

But the devil's in the details. A production-ready URL shortener needs to handle:

Unique short codes without collisions
Analytics tracking for every click
Custom slugs for branded links
Expiration dates for temporary campaigns
Security against abuse and malicious URLs

The Technical Challenge: URL Slug Collisions and Uniqueness

This was the most interesting problem to solve. How do you generate short, unique identifiers that won't collide as your database grows?

The Naive Approach (Don't Do This)

My first instinct was to use random strings:

import random
import string

def generate_slug():
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))

The problem? With a 6-character alphanumeric string (36^6 = 2.1 billion possibilities), you'd think collisions are rare. But thanks to the quirky birthday paradox, you'll hit your first collision around 55,000 URLs, which is a bit sooner than expected.

The Better Approach: Base62 Encoding with Auto-Increment IDs

Instead, I went with a deterministic approach using database auto-increment IDs:

import string
from typing import Optional

BASE62 = string.digits + string.ascii_lowercase + string.ascii_uppercase

def encode_id(num: int, min_length: int = 6) -> str:
    """
    Convert a number to a base62 string, padded to minimum length.
    With 6 characters and base62, you can encode up to 56.8 billion URLs.
    """
    if num == 0:
        return BASE62[0].zfill(min_length)

    result = []
    while num:
        result.append(BASE62[num % 62])
        num //= 62

    encoded = ''.join(reversed(result))

    # Pad with leading zeros to meet minimum length
    if len(encoded) < min_length:
        encoded = BASE62[0] * (min_length - len(encoded)) + encoded

    return encoded

def decode_slug(slug: str) -> Optional[int]:
    """Convert a base62 string back to a number."""
    try:
        num = 0
        for char in slug:
            num = num * 62 + BASE62.index(char)
        return num
    except ValueError:
        # Character not in BASE62 alphabet
        return None

This approach guarantees:

Zero collisions: Each ID maps to exactly one slug
Consistent length: All slugs are 6 characters (e.g., "000001", "000abc")
Predictable growth: You know exactly how many URLs each slug length supports
Reversibility: You can decode slugs back to IDs for quick lookup

The math is beautiful:

6 characters (default): 56.8 billion possible URLs
First URL: 000001 (ID: 1)
100th URL: 0001Cv (ID: 100)
Millionth URL: 004C92 (ID: 1,000,000)

By padding to 6 characters, all URLs have a consistent, professional look. When you eventually hit 6 characters of actual data, they're already quite short anyway!

Custom Slugs: The Best of Both Worlds

Of course, sometimes you want a branded slug like myurl.app/blog-launch instead of myurl.app/000x7k. I added support for custom slugs with validation and uniqueness checks:

RESERVED_SLUGS = {"stats", "shorten", "admin", "login", "logout"}

def validate_custom_slug(slug: str) -> bool:
    """Validate custom slug format."""
    if not slug or len(slug) < 3 or len(slug) > 50:
        return False

    allowed_chars = set(string.ascii_letters + string.digits + '-_')
    return all(char in allowed_chars for char in slug)

# In the /shorten route:
if custom_slug:
    custom_slug = custom_slug.strip().lower()  # Normalize to lowercase

    # Check if slug is reserved
    if custom_slug in RESERVED_SLUGS:
        raise HTTPException(status_code=400, detail="Slug is reserved")

    # Validate format
    if not validate_custom_slug(custom_slug):
        raise HTTPException(status_code=400, detail="Invalid slug format")

    # Check if already taken
    existing = db.query(URL).filter(URL.slug == custom_slug).first()
    if existing:
        raise HTTPException(status_code=400, detail="Slug already taken")

    new_url = URL(slug=custom_slug, long_url=long_url)
else:
    # Auto-generate from ID
    new_url = URL(long_url=long_url, slug="temp")
    db.add(new_url)
    db.flush()
    new_url.slug = encode_id(new_url.id, min_length=6)

Custom slugs are validated to ensure they only contain alphanumeric characters, hyphens, and underscores (3-50 characters). Reserved slugs like "stats" and "admin" are blocked to avoid conflicts with routes.

Analytics and Dashboard

This is where building your own URL shortener really shines. Every major platform (Twitter, LinkedIn, newsletters, GitHub README) has different traffic patterns, and I wanted to see them all in one place.

What I Track

For every click, I capture:

class Click(Base):
    __tablename__ = "clicks"

    id = Column(Integer, primary_key=True, index=True)
    url_id = Column(Integer, ForeignKey("urls.id"), nullable=False, index=True)
    timestamp = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)

    # Request metadata
    referrer = Column(String, nullable=True)
    user_agent = Column(String, nullable=True)
    ip_address = Column(String, nullable=True)

    # Geographic data (would require GeoIP library)
    country = Column(String, nullable=True)
    city = Column(String, nullable=True)

    # Device information (parsed from user agent)
    device_type = Column(String, nullable=True)  # mobile, desktop, tablet, bot
    browser = Column(String, nullable=True)
    os = Column(String, nullable=True)

    url = relationship("URL", back_populates="clicks")

The Dashboard

I built a simple dashboard using Bootstrap 5 that shows:

Total clicks with time-series line chart
Unique visitors tracked by IP address
Device type breakdown (mobile, desktop, tablet, bot) as a pie chart
Top referrers with percentage bars (Twitter, LinkedIn, Direct, etc.)
Average clicks per day since creation
Recent clicks table showing timestamp, referrer, device, browser, OS, and IP

The most surprising insight? My GitHub repository links get significantly more clicks from mobile devices than desktop. I never would have guessed that people browse code repos on their phones so much!

Note: Geographic distribution (countries/cities) could be added using a GeoIP library like geoip2, but I kept it simple to avoid external dependencies and API limits.

Building it with FastAPI, Bootstrap 5, and Jinja2

Let me walk you through the actual implementation. I chose this stack because:

FastAPI: Lightning-fast, automatic API docs, async support
Bootstrap 5: Quick, responsive UI without wrestling with CSS
Jinja2: Server-side rendering for SEO and simplicity
SQLite: Zero-config database perfect for side projects

The Database Schema

Simple SQLAlchemy models with a one-to-many relationship:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime, timezone

class URL(Base):
    __tablename__ = "urls"

    id = Column(Integer, primary_key=True, autoincrement=True, index=True)
    slug = Column(String, unique=True, index=True, nullable=False)
    long_url = Column(String, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
    expires_at = Column(DateTime, nullable=True)

    clicks = relationship("Click", back_populates="url", cascade="all, delete-orphan")

class Click(Base):
    __tablename__ = "clicks"

    id = Column(Integer, primary_key=True, index=True)
    url_id = Column(Integer, ForeignKey("urls.id"), nullable=False, index=True)
    timestamp = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)

    # Request metadata
    referrer = Column(String, nullable=True)
    user_agent = Column(String, nullable=True)
    ip_address = Column(String, nullable=True)

    # Device information (parsed from user agent)
    device_type = Column(String, nullable=True)
    browser = Column(String, nullable=True)
    os = Column(String, nullable=True)

    url = relationship("URL", back_populates="clicks")

The URL model stores the shortened links, while Click tracks every redirect for analytics. Using datetime.now(timezone.utc) ensures timezone-aware timestamps. The cascade option means deleting a URL also removes all its clicks.

The FastAPI Application

Here are the key routes (see full implementation on GitHub):

from fastapi import FastAPI, HTTPException, Request, Depends, Form
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from typing import Optional

app = FastAPI(title="TinyURL Clone")

@app.post("/shorten")
async def create_short_url(
    request: Request,
    long_url: str = Form(...),
    custom_slug: Optional[str] = Form(None),
    db: Session = Depends(get_db)
):
    """Create a new short URL"""
    # Validate URL format
    if not validate_url(long_url):
        raise HTTPException(status_code=400, detail="Invalid URL format")

    if custom_slug:
        custom_slug = custom_slug.strip().lower()

        # Validate custom slug format
        if not validate_custom_slug(custom_slug):
            raise HTTPException(status_code=400, detail="Invalid slug format")

        # Check if already taken
        existing = db.query(URL).filter(URL.slug == custom_slug).first()
        if existing:
            raise HTTPException(status_code=400, detail="Slug already taken")

        new_url = URL(slug=custom_slug, long_url=long_url)
        db.add(new_url)
        db.commit()
    else:
        # Auto-generate from ID
        new_url = URL(long_url=long_url, slug="temp")
        db.add(new_url)
        db.flush()  # Get the auto-increment ID
        new_url.slug = encode_id(new_url.id)
        db.commit()

    base_url = str(request.base_url).rstrip('/')
    return {"short_url": f"{base_url}/{new_url.slug}"}

@app.get("/{slug}")
async def redirect_to_long_url(
    slug: str,
    request: Request,
    db: Session = Depends(get_db)
):
    """Redirect short URL to original URL and track analytics"""
    url = db.query(URL).filter(URL.slug == slug).first()

    if not url:
        raise HTTPException(status_code=404, detail="URL not found")

    # Check expiration
    if url.expires_at and datetime.now(timezone.utc) > url.expires_at:
        raise HTTPException(status_code=410, detail="URL has expired")

    # Parse user agent for device info
    device_info = parse_user_agent(request.headers.get("user-agent"))

    # Track the click
    click = Click(
        url_id=url.id,
        referrer=request.headers.get("referer"),
        user_agent=request.headers.get("user-agent"),
        ip_address=request.client.host if request.client else None,
        device_type=device_info["device_type"],
        browser=device_info["browser"],
        os=device_info["os"]
    )
    db.add(click)
    db.commit()

    return RedirectResponse(url=url.long_url, status_code=302)

The application also includes routes for viewing stats (/stats/{slug}) and the homepage dashboard. Note the use of Form(...) for proper form data handling and the parse_user_agent() utility for extracting device information.

The Bootstrap 5 Dashboard Template

I built a clean, responsive UI using Bootstrap 5. Here's the key form structure:

<!-- templates/index.html - Main form (simplified) -->
<form id="shortenForm">
  <div class="mb-3">
    <label for="longUrl" class="form-label">Long URL</label>
    <input
      type="url"
      class="form-control form-control-lg"
      id="longUrl"
      placeholder="https://example.com/very/long/url"
      required
    />
  </div>

  <div class="mb-3">
    <label for="customSlug" class="form-label"> Custom Slug (Optional) </label>
    <input
      type="text"
      class="form-control"
      id="customSlug"
      placeholder="my-custom-link"
    />
  </div>

  <button type="submit" class="btn btn-primary btn-lg w-100">
    Shorten URL
  </button>
</form>

<script>
  document
    .getElementById("shortenForm")
    .addEventListener("submit", async (e) => {
      e.preventDefault();

      const formData = new URLSearchParams();
      formData.append("long_url", document.getElementById("longUrl").value);
      formData.append(
        "custom_slug",
        document.getElementById("customSlug").value,
      );

      const response = await fetch("/shorten", {
        method: "POST",
        headers: { "Content-Type": "application/x-www-form-urlencoded" },
        body: formData,
      });

      const data = await response.json();
      // Display the short URL
      document.getElementById("shortUrl").value = data.short_url;
    });
</script>

The stats dashboard (templates/stats.html) displays analytics cards showing total clicks, unique visitors, referrer breakdown, and device statistics using Chart.js for visualizations.

Full templates available in the GitHub repository.

SQLite: The Perfect Database for Side Projects

One of the best decisions I made was using SQLite. Here's why:

Zero Configuration

import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Can be overridden with environment variable
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./urls.db")

# SQLite-specific optimizations
connect_args = {
    "check_same_thread": False,
    "timeout": 30
}

engine = create_engine(
    DATABASE_URL,
    connect_args=connect_args,
    pool_pre_ping=True  # Handle stale connections
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

The entire database is a single file (urls.db) that lives in your project directory. You can back it up, version control it, or deploy it anywhere. The pool_pre_ping option helps prevent stale connection errors.

Surprisingly Powerful

SQLite handles:

Transactions (ACID compliance)
Foreign keys
Indexes
Full-text search
JSON columns (SQLite 3.38+)

For a URL shortener with even millions of URLs, SQLite performs beautifully. It's only when you need high-concurrency writes (thousands per second) that you'd need PostgreSQL or MySQL.

Easy Backups

# Backup
sqlite3 urls.db ".backup urls_backup.db"

# Restore
cp urls_backup.db urls.db

When to Upgrade

You should consider moving to PostgreSQL when:

You're getting >100 writes/second
You need multiple servers accessing the same database
You want advanced features like full-text search or geospatial queries
You're running in a distributed environment

For a personal URL shortener tracking blog posts and GitHub stats, SQLite is more than sufficient.

Deployment and Production Considerations

Running in Production

I deployed mine using Docker on a small VPS:

FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application files
COPY . .

# Create directory for database
RUN mkdir -p /app/data

# Expose port
EXPOSE 8000

# Run the application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml (simplified)
services:
  app:
    build: .
    container_name: tinyurl-clone
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data
      - ./templates:/app/templates
    environment:
      - DATABASE_URL=sqlite:///./data/urls.db
    restart: unless-stopped

The volume mounts ensure your database persists across container restarts, and mounting templates allows hot-reload during development.

Security Considerations

The current implementation includes basic validation (URL format checking and slug sanitization), but for production deployment, you should add:

Rate Limiting: Use slowapi to limit requests (e.g., 10/minute per IP) to prevent abuse

   from slowapi import Limiter
   limiter = Limiter(key_func=get_remote_address)

   @app.post("/shorten")
   @limiter.limit("10/minute")
   async def create_short_url(...):
       # Your code here

Malicious URL Blocking: Maintain a blocklist of known malicious domains

   BLOCKED_DOMAINS = {"malware.com", "phishing.net"}

   def validate_url(url: str) -> bool:
       parsed = urlparse(url)
       if parsed.netloc in BLOCKED_DOMAINS:
           return False

HTTPS Only: Use Let's Encrypt with nginx reverse proxy in production
Database Backups: Automate regular SQLite backups to prevent data loss

For a production system handling sensitive data, consider adding authentication, API keys, and more comprehensive logging.

Try It Out: Running the Demo

Want to see it in action? Getting started takes less than 5 minutes:

# Clone the repository
git clone https://github.com/nunombispo/building-my-own-tinyurl-article
cd building-my-own-tinyurl-article

# Install dependencies
pip install -r requirements.txt

# Run the application
uvicorn main:app --reload

Then visit http://localhost:8000 in your browser. You'll see:

Homepage: Clean form to shorten URLs with optional custom slugs
Instant results: Copy button and links to visit or view stats
Analytics dashboard: Visit /stats/{slug} to see click tracking in action

Try creating a few short URLs, clicking them from different devices, and watching the analytics populate in real-time. The SQLite database will be created automatically on first run.

Example of the homepage:

Example of the analytics main information:

And referrers and recent clicks:

Full source code at: https://github.com/nunombispo/building-my-own-tinyurl-article

Was It Worth It?

Absolutely. The total time investment was about 8 hours, and I now have:

Complete ownership of my links
Detailed analytics tailored to my needs
No monthly subscription fees
A fun weekend project that taught me about base62 encoding, collision handling, and FastAPI

Plus, I can share this with friends, add it to my portfolio, and know exactly how my content performs across different platforms.

Conclusion

Building a URL shortener taught me that sometimes the best tools are the ones you build yourself. Not because they're necessarily better than existing solutions, but because they're perfectly tailored to your needs.

For tracking blog posts, free guides, and GitHub statistics, having complete control over the analytics pipeline is invaluable. You own your data, customize the features you want, and learn something new in the process.

And honestly? It was just really fun to build.

If you're thinking about building your own, I say go for it. The technical challenges are interesting but manageable, and the end result is something you'll actually use every day. Start simple with the core features, then add complexity as you need it.

The complete source code is available on GitHub, so clone it, try it out, and make it your own!

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

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

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

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

Server Monitoring Without Prometheus - A Python + SSH Approach

Developer Service — Mon, 09 Feb 2026 07:50:08 +0000

You don’t need a monitoring stack that requires its own monitoring. If you’re running a handful of servers, Prometheus and Grafana are often more work than the problems they’re meant to solve.

If you can ssh user@server, you already have everything you need.

In this tutorial, we’ll build a lightweight, no-agent server monitoring tool using Python and SSH, designed for indie hackers, solo founders, and small teams. By the end, you’ll have a single script that checks CPU, memory, disk usage, and critical services across multiple servers, renders a clean terminal dashboard, and sends alerts when something actually breaks.

What we’ll build:

Monitor multiple servers over SSH (no agents, no daemons)
A real-time terminal dashboard with colors and status indicators
Alerts via ntfy.sh
Remote server updates with push notifications
One Python script you can read, trust, and extend

Why Not Just Use Prometheus?

Before we dive in, let’s get this out of the way: Prometheus and Grafana are excellent tools. At scale, they’re hard to beat, and they’re industry standard for a reason.

But if you’re running a small setup, say 1–20 servers, a few side projects, or a lean SaaS, they often introduce more complexity than actual value.

What starts as “just add monitoring” quickly turns into:

Heavy infrastructure: Prometheus, Grafana, node exporters, persistent storage, backups
Ongoing maintenance: Upgrades, config drift, broken dashboards at the worst time
A steep learning curve: PromQL, recording rules, and yet another YAML ecosystem

All of that just to answer a few basic questions:

Is the server up?

Is disk space running out?

Did nginx or Docker crash?

Meanwhile, SSH is already (probably) open on your servers, and Python is already installed (or trivial to install). You don’t need a full metrics pipeline to check system health, you just need a reliable way to run commands and interpret the results.

If you can ssh user@server, you can monitor it.

That’s the premise of this article. We’ll keep the stack boring, the moving parts minimal, and the script small enough that you actually understand, and trust, it.

Project Setup

Let's get you started by creating a new project directory and set up the environment:

mkdir server-monitor
cd server-monitor

Then, create a requirements.txt file:

paramiko>=4.0.0
rich>=14.3.2
pyyaml>=6.0.3
requests>=2.32.5

And install the dependencies:

pip install -r requirements.txt

What each library does:

paramiko -> SSH connections and command execution
rich -> Beautiful terminal output with tables and colors
pyyaml -> Parse our server configuration file
requests -> Send alerts to webhooks (Slack)

Now create a servers.yml file to define which servers to monitor:

servers:
  - name: prd-dokku
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

  - name: prd-ghost
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

  - name: prd-n8n
    host: XX.XX.XX.XX
    user: root
    key_file: id_rsa
    services:
      - docker

thresholds:
  cpu_load_factor: 2.0 # Alert if load > cores × 2
  memory_percent: 85
  disk_percent: 25

Here you should replace the IP addresses and usernames with your actual servers.

Make sure you have SSH key authentication set up for each server. If your key has a passphrase, use the SSH_KEY_PASSPHRASE environment variable or add key_passphrase to a server entry; the script will also prompt interactively if needed.

SSH Connection Manager

Let's start building the monitor. Create a file called monitor.py and you'll build it piece by piece.

First, you'll need a function to connect to servers and run commands over SSH:

"""
Simple server monitoring over SSH
"""

import os
import sys
import threading
import getpass
import paramiko
import yaml
from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class ServerMetrics:
    """Container for server metrics"""
    name: str
    reachable: bool = True
    cpu_load: Optional[float] = None
    memory_percent: Optional[float] = None
    disk_percent: Optional[float] = None
    services: Dict[str, str] = field(default_factory=dict)
    error: Optional[str] = None


class SSHClient:
    """Manages SSH connections to servers"""

    def __init__(
        self,
        host: str,
        user: str,
        key_file: Optional[str] = None,
        key_passphrase: Optional[str] = None,
    ):
        self.host = host
        self.user = user
        self.key_file = key_file
        self.key_passphrase = key_passphrase
        self.client = None

    def _load_pkey(self, passphrase: Optional[str] = None) -> Optional[paramiko.PKey]:
        """Load private key, trying RSA then Ed25519. Returns None to fall back to key_filename."""
        if not self.key_file:
            return None
        password = passphrase or self.key_passphrase
        # Try RSA key first (PEM format)
        try:
            return paramiko.RSAKey.from_private_key_file(
                self.key_file, password=password
            )
        except paramiko.ssh_exception.PasswordRequiredException:
            if password is None:
                password = getpass.getpass(f"SSH key passphrase for {self.key_file}: ")
            return paramiko.RSAKey.from_private_key_file(
                self.key_file, password=password
            )
        except (paramiko.ssh_exception.SSHException, OSError):
            pass
        # Try Ed25519
        try:
            return paramiko.Ed25519Key.from_private_key_file(
                self.key_file, password=password
            )
        except paramiko.ssh_exception.PasswordRequiredException:
            if password is None:
                password = getpass.getpass(f"SSH key passphrase for {self.key_file}: ")
            return paramiko.Ed25519Key.from_private_key_file(
                self.key_file, password=password
            )
        except (paramiko.ssh_exception.SSHException, OSError):
            pass
        return None

    def connect(self) -> bool:
        """Establish SSH connection"""
        try:
            self.client = paramiko.SSHClient()
            self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            connect_kwargs = {
                'hostname': self.host,
                'username': self.user,
                'timeout': 10,
                'look_for_keys': True,
                'allow_agent': True,
            }

            # Load key explicitly (RSA / Ed25519) with optional passphrase
            if self.key_file:
                pkey = self._load_pkey()
                if pkey is not None:
                    connect_kwargs['pkey'] = pkey
                else:
                    connect_kwargs['key_filename'] = self.key_file
                    if self.key_passphrase:
                        connect_kwargs['passphrase'] = self.key_passphrase

            self.client.connect(**connect_kwargs)
            return True

        except Exception as e:
            print(f"❌ Failed to connect to {self.host}: {e}")
            return False

    def run_command(
        self, command: str, stream_output: bool = False
    ) -> tuple[str, str, int]:
        """
        Run a command over SSH.
        Returns: (stdout, stderr, exit_code).
        If stream_output=True, print stdout/stderr as it arrives (for long-running commands).
        """
        if not self.client:
            return "", "Not connected", 1

        try:
            stdin, stdout, stderr = self.client.exec_command(command)
            channel = stdout.channel

            if stream_output:
                out_lines = []
                err_lines = []

                def read_stdout():
                    for line in iter(stdout.readline, ""):
                        decoded = line
                        out_lines.append(decoded)
                        print(decoded, end="")

                def read_stderr():
                    for line in iter(stderr.readline, ""):
                        decoded = line
                        err_lines.append(decoded)
                        print(decoded, end="", file=sys.stderr)

                t1 = threading.Thread(target=read_stdout)
                t2 = threading.Thread(target=read_stderr)
                t1.daemon = True
                t2.daemon = True
                t1.start()
                t2.start()
                exit_code = channel.recv_exit_status()
                t1.join(timeout=1)
                t2.join(timeout=1)
                return ("".join(out_lines).strip(), "".join(err_lines).strip(), exit_code)

            exit_code = channel.recv_exit_status()
            return (
                stdout.read().decode("utf-8", errors="replace").strip(),
                stderr.read().decode("utf-8", errors="replace").strip(),
                exit_code,
            )
        except Exception as e:
            return "", str(e), 1

    def close(self):
        """Close SSH connection"""
        if self.client:
            self.client.close()


def load_config(config_file: str = 'servers.yml') -> dict:
    """Load server configuration from YAML"""
    with open(config_file, 'r') as f:
        return yaml.safe_load(f)

This gives you the foundation: a clean SSH client that can connect and run commands, plus a data structure to hold the metrics.

It defines:

A ServerMetrics dataclass to store health information per server (reachability, CPU, memory, disk, service status, and errors).
An SSHClient wrapper around Paramiko that:
- Handles SSH connections using RSA or Ed25519 keys (with optional passphrases)
- Falls back to SSH agents or default keys when possible
- Executes remote commands and captures output, errors, and exit codes (with optional streaming for long-running commands like apt upgrade)
- Manages connection lifecycle cleanly
A load_config helper to read server definitions from a YAML file.

For more Python fundamentals that help when reading or extending this script, for instance monitoring at the file-system level (e.g. watching config or log files), see Mastering File System Monitoring with Watchdog in Python.

Note on SSH key passphrase: If your private key has a passphrase you can (1) set the SSH_KEY_PASSPHRASE environment variable (recommended for a cron job), (2) add key_passphrase to a server entry in servers.yml, or (3) let the script prompt you interactively when needed.

Collecting Metrics

Now you'll add functions to gather actual metrics from each server. Add these functions to monitor.py:

def get_cpu_load(ssh: SSHClient) -> Optional[float]:
    """Get 1-minute load average from uptime command"""
    stdout, stderr, code = ssh.run_command('uptime')
    if code != 0:
        return None

    # Parse: "... load average: 0.52, 0.58, 0.59"
    try:
        load_str = stdout.split('load average:')[1].split(',')[0].strip()
        return float(load_str)
    except (IndexError, ValueError):
        return None


def get_memory_usage(ssh: SSHClient) -> Optional[float]:
    """Get memory usage percentage from free command"""
    stdout, stderr, code = ssh.run_command('free -m')
    if code != 0:
        return None

    # Parse output to get used/total
    try:
        lines = stdout.split('\n')
        mem_line = [l for l in lines if l.startswith('Mem:')][0]
        parts = mem_line.split()
        total = float(parts[1])
        used = float(parts[2])
        return (used / total) * 100
    except (IndexError, ValueError):
        return None


def get_disk_usage(ssh: SSHClient) -> Optional[float]:
    """Get root disk usage percentage from df command"""
    stdout, stderr, code = ssh.run_command('df -h /')
    if code != 0:
        return None

    # Parse: "Filesystem Size Used Avail Use% Mounted"
    try:
        lines = stdout.split('\n')
        data_line = lines[1]
        use_percent = data_line.split()[4].rstrip('%')
        return float(use_percent)
    except (IndexError, ValueError):
        return None


def get_service_status(ssh: SSHClient, service_name: str) -> str:
    """Check if a systemd service is active"""
    stdout, stderr, code = ssh.run_command(f'systemctl is-active {service_name}')

    if stdout == 'active':
        return 'active'
    elif code != 0:
        # Could be 'inactive', 'failed', or service doesn't exist
        return stdout if stdout else 'unknown'
    return 'unknown'


def collect_server_metrics(server_config: dict, thresholds: dict) -> ServerMetrics:
    """
    Connect to a server and collect all metrics
    """
    name = server_config['name']
    host = server_config['host']
    user = server_config['user']
    services = server_config.get('services', [])
    key_file = server_config.get('key_file')
    key_passphrase = server_config.get('key_passphrase') or os.environ.get('SSH_KEY_PASSPHRASE')

    metrics = ServerMetrics(name=name)
    ssh = SSHClient(host, user, key_file, key_passphrase)

    # Try to connect
    if not ssh.connect():
        metrics.reachable = False
        metrics.error = f"Could not connect to {host}"
        return metrics

    try:
        # Gather all metrics
        metrics.cpu_load = get_cpu_load(ssh)
        metrics.memory_percent = get_memory_usage(ssh)
        metrics.disk_percent = get_disk_usage(ssh)

        # Check services
        for service in services:
            metrics.services[service] = get_service_status(ssh, service)

    except Exception as e:
        metrics.error = str(e)

    finally:
        ssh.close()

    return metrics

This code collects basic server health metrics over SSH using standard Linux commands.

It:

Retrieves CPU load, memory usage, and disk usage by running uptime, free, and df remotely and parsing their output
Checks the status of systemd services via systemctl is-active
Wraps all results in a ServerMetrics object
Handles connection failures and parsing errors gracefully
Uses SSH keys (with optional passphrases) for secure, agentless access

If something fails, it returns None or a status string rather than crashing, which is important when monitoring multiple servers.

Terminal Dashboard with Rich

Now for the fun part, displaying our metrics in a beautiful terminal table. Add this to monitor.py:

from rich.console import Console
from rich.table import Table
from rich import box


def create_status_icon(value: Optional[float], threshold: float,
                       reverse: bool = False) -> str:
    """
    Create a colored status icon based on value vs threshold
    reverse=True means lower is better (e.g., disk usage)
    """
    if value is None:
        return "❓"

    if reverse:
        # Lower is better (disk, memory)
        if value < threshold * 0.7:
            return "✅"
        elif value < threshold:
            return "⚠️"
        else:
            return "❌"
    else:
        # Higher is better
        if value > threshold:
            return "❌"
        elif value > threshold * 0.7:
            return "⚠️"
        else:
            return "✅"


def format_metric_value(value: Optional[float], suffix: str = "",
                       decimals: int = 1) -> str:
    """Format a metric value for display"""
    if value is None:
        return "—"
    return f"{value:.{decimals}f}{suffix}"


def display_dashboard(metrics_list: List[ServerMetrics], thresholds: dict):
    """Display monitoring dashboard using Rich tables"""
    console = Console()

    # Create table
    table = Table(title="🖥️  Server Monitoring Dashboard",
                  box=box.ROUNDED,
                  show_header=True,
                  header_style="bold cyan")

    # Add columns
    table.add_column("Server", style="bold white", no_wrap=True)
    table.add_column("Status", justify="center")
    table.add_column("CPU Load", justify="right")
    table.add_column("Memory", justify="right")
    table.add_column("Disk", justify="right")
    table.add_column("Services", justify="left")

    # Add rows
    for metrics in metrics_list:
        # Overall status
        if not metrics.reachable:
            status = "🔴 Down"
        elif metrics.error:
            status = "⚠️ Error"
        else:
            status = "🟢 Up"

        # CPU with status icon
        cpu_display = format_metric_value(metrics.cpu_load)
        cpu_icon = create_status_icon(
            metrics.cpu_load,
            thresholds.get('cpu_load_factor', 2.0),
            reverse=False
        )
        cpu_cell = f"{cpu_icon} {cpu_display}"

        # Memory with status icon and percentage
        mem_display = format_metric_value(metrics.memory_percent, "%")
        mem_icon = create_status_icon(
            metrics.memory_percent,
            thresholds.get('memory_percent', 85),
            reverse=True
        )
        mem_cell = f"{mem_icon} {mem_display}"

        # Disk with status icon and percentage
        disk_display = format_metric_value(metrics.disk_percent, "%")
        disk_icon = create_status_icon(
            metrics.disk_percent,
            thresholds.get('disk_percent', 80),
            reverse=True
        )
        disk_cell = f"{disk_icon} {disk_display}"

        # Services status
        if metrics.services:
            service_items = []
            for svc, status_val in metrics.services.items():
                icon = "✅" if status_val == "active" else "❌"
                service_items.append(f"{icon} {svc}")
            services_cell = "\n".join(service_items)
        else:
            services_cell = "—"

        # Add row to table
        table.add_row(
            metrics.name,
            status,
            cpu_cell,
            mem_cell,
            disk_cell,
            services_cell
        )

    # Display
    console.print()
    console.print(table)
    console.print()

The rich library makes this surprisingly easy.

It:

Converts raw metric values into clear status icons (✅ ⚠️ ❌) based on configurable thresholds
Formats CPU, memory, and disk metrics for human-friendly display
Builds a styled Rich table showing per-server health, resource usage, and service status

Alert System

Monitoring without alerts is just anxiety. Let's add a simple alert system that can notify you via ntfy.sh. Add these functions:

def check_thresholds(metrics: ServerMetrics, thresholds: dict) -> List[str]:
    """
    Check if any metrics breach thresholds
    Returns list of alert messages
    """
    alerts = []

    if not metrics.reachable:
        alerts.append(f"🔴 {metrics.name}: Server unreachable")
        return alerts

    # CPU load
    if metrics.cpu_load and metrics.cpu_load > thresholds.get('cpu_load_factor', 2.0):
        alerts.append(
            f"⚠️ {metrics.name}: High CPU load ({metrics.cpu_load:.2f})"
        )

    # Memory
    if metrics.memory_percent and metrics.memory_percent > thresholds.get('memory_percent', 85):
        alerts.append(
            f"⚠️ {metrics.name}: High memory usage ({metrics.memory_percent:.1f}%)"
        )

    # Disk
    if metrics.disk_percent and metrics.disk_percent > thresholds.get('disk_percent', 80):
        alerts.append(
            f"⚠️ {metrics.name}: High disk usage ({metrics.disk_percent:.1f}%)"
        )

    # Failed services
    for service, status in metrics.services.items():
        if status != 'active':
            alerts.append(
                f"❌ {metrics.name}: Service '{service}' is {status}"
            )

    return alerts


def send_alert_ntfy(message: str, topic: str):
    """Send alert to ntfy.sh"""
    try:
        import requests
        url = f"https://ntfy.sh/{topic}"
        requests.post(url, data=message.encode('utf-8'))
    except Exception as e:
        print(f"Failed to send ntfy alert: {e}")


def send_alerts(alerts: List[str], config: dict):
    """Send alerts through configured channels"""
    if not alerts:
        return

    alert_config = config.get('alerts', {})
    message = "\n".join(alerts)

    # Send to ntfy
    if ntfy_topic := alert_config.get('ntfy_topic'):
        send_alert_ntfy(message, ntfy_topic)

    # Print to console
    console = Console()
    console.print(f"\n[bold red]🚨 Alerts:[/bold red]")
    for alert in alerts:
        console.print(f"  {alert}")
    console.print()

The alert logic is straightforward:

Evaluates collected server metrics against configurable thresholds
Detects unreachable servers, high CPU/memory/disk usage, and failed services
Generates human-readable alert messages
Sends alerts to ntfy.sh

You can easily add more channels (email, Slack, Telegram, PagerDuty) by adding similar send_alert_* functions.

To configure alerts, add this to your servers.yml:

alerts:
  ntfy_topic: "myapp-monitoring" # Optional

Putting It All Together

Now let's add the main function that ties everything together:

def main():
    """Main monitoring loop"""
    console = Console()

    try:
        # Load configuration
        config = load_config('servers.yml')
        servers = config.get('servers', [])
        thresholds = config.get('thresholds', {})

        if not servers:
            console.print("[red]No servers configured in servers.yml[/red]")
            return

        console.print(f"[cyan]Monitoring {len(servers)} server(s)...[/cyan]\n")

        # Collect metrics from all servers
        all_metrics = []
        all_alerts = []

        for server in servers:
            metrics = collect_server_metrics(server, thresholds)
            all_metrics.append(metrics)

            # Check for alerts
            alerts = check_thresholds(metrics, thresholds)
            all_alerts.extend(alerts)

        # Display dashboard
        display_dashboard(all_metrics, thresholds)

        # Send alerts if any
        if all_alerts:
            send_alerts(all_alerts, config)
        else:
            console.print("[green]✅ All systems healthy[/green]\n")

    except FileNotFoundError:
        console.print("[red]Error: servers.yml not found[/red]")
    except Exception as e:
        console.print(f"[red]Error: {e}[/red]")


if __name__ == '__main__':
    main()

This the main entry point of the monitoring script:

Loads server and threshold configuration from servers.yml
Connects to each server and collects health metrics
Evaluates metrics against alert thresholds
Displays a Rich-based monitoring dashboard
Sends alerts if any issues are detected, otherwise reports a healthy state

That's it! You now have a complete monitoring script.

Let's test it:

python monitor.py

You should see a beautiful table showing the status of all your servers:

Monitoring 3 server(s)...

SSH key passphrase for id_rsa: 
SSH key passphrase for id_rsa: 
SSH key passphrase for id_rsa: 

                  🖥️  Server Monitoring Dashboard                  
╭───────────┬────────┬──────────┬──────────┬──────────┬───────────╮
│ Server    │ Status │ CPU Load │   Memory │     Disk │ Services  │
├───────────┼────────┼──────────┼──────────┼──────────┼───────────┤
│ prd-dokku │ 🟢 Up  │   ✅ 0.0 │ ✅ 17.8% │ ✅ 15.0% │ ✅ docker │
│ prd-ghost │ 🟢 Up  │   ✅ 0.5 │ ✅ 46.6% │ ❌ 32.0% │ ✅ docker │
│ prd-n8n   │ 🟢 Up  │   ✅ 0.0 │ ✅ 26.7% │ ⚠️ 19.0% │ ✅ docker │
╰───────────┴────────┴──────────┴──────────┴──────────┴───────────╯


🚨 Alerts:
  ⚠️ prd-ghost: High disk usage (32.0%)

And in this case, you should also receive a notification:

Remote Server Updates

Let's extend your monitor to handle server updates. Add these functions to monitor.py:

def run_server_update(ssh: SSHClient, stream_output: bool = True) -> dict:
    """
    Run apt update and upgrade on a server.
    stream_output=True prints apt progress so it doesn't look stuck (can take 5–15 min).
    Returns dict with results.
    """
    result = {
        'success': False,
        'output': '',
        'reboot_required': False,
        'error': None
    }

    # Run update and upgrade (stream output so user sees progress)
    # DEBIAN_FRONTEND=noninteractive avoids any prompts that could hang
    cmd = 'sudo DEBIAN_FRONTEND=noninteractive apt-get update && sudo DEBIAN_FRONTEND=noninteractive apt-get upgrade -y 2>&1'
    stdout, stderr, code = ssh.run_command(cmd, stream_output=stream_output)

    result['output'] = stdout
    result['success'] = (code == 0)

    if code != 0:
        result['error'] = stderr or 'Update failed'
        return result

    # Check if reboot is required
    reboot_check_cmd = '[ -f /var/run/reboot-required ] && echo "yes" || echo "no"'
    reboot_stdout, _, _ = ssh.run_command(reboot_check_cmd)
    result['reboot_required'] = (reboot_stdout.strip() == 'yes')

    return result


def update_server(server_config: dict, notify_topic: Optional[str] = None) -> bool:
    """Update a single server and optionally send notification"""
    name = server_config['name']
    host = server_config['host']
    user = server_config['user']
    key_file = server_config.get('key_file')
    key_passphrase = server_config.get('key_passphrase') or os.environ.get('SSH_KEY_PASSPHRASE')

    console = Console()
    console.print(f"[cyan]Updating {name}...[/cyan]")
    console.print("[dim]Running apt update & upgrade (may take 5–15 min). Output streams below.[/dim]\n")

    ssh = SSHClient(host, user, key_file, key_passphrase)

    if not ssh.connect():
        console.print(f"[red]❌ Could not connect to {name}[/red]")
        return False

    try:
        result = run_server_update(ssh)

        if result['success']:
            reboot_msg = "⚠️ Reboot required" if result['reboot_required'] else "No reboot needed"
            console.print(f"[green]✅ {name} updated successfully[/green]")
            console.print(f"   {reboot_msg}")

            # Send notification
            if notify_topic:
                message = f"✅ {name} updated successfully\n{reboot_msg}"
                send_alert_ntfy(message, notify_topic)

            return True
        else:
            console.print(f"[red]❌ {name} update failed: {result['error']}[/red]")

            if notify_topic:
                message = f"❌ {name} update failed\n{result['error']}"
                send_alert_ntfy(message, notify_topic)

            return False

    finally:
        ssh.close()


def update_all_servers():
    """Update all configured servers"""
    config = load_config('servers.yml')
    servers = config.get('servers', [])
    notify_topic = config.get('alerts', {}).get('ntfy_topic')

    console = Console()
    console.print(f"\n[bold cyan]Updating {len(servers)} server(s)...[/bold cyan]\n")

    for server in servers:
        update_server(server, notify_topic)
        console.print()

This code adds remote server update automation to the monitoring tool:

Runs apt update and apt upgrade over SSH with streaming output so progress is visible (avoids looking stuck; updates can take 5–15 minutes)
Uses DEBIAN_FRONTEND=noninteractive so apt never waits for prompts
Detects whether a reboot is required after updates
Reports success or failure with clear terminal output
Sends update notifications via ntfy.sh

Now update the main() function to support an --update flag:

import sys

def main():
    """Main monitoring script"""
    console = Console()

    # Check for update mode
    if len(sys.argv) > 1 and sys.argv[1] == '--update':
        update_all_servers()
        return

    # ... rest of the monitoring code stays the same ...

You can update all your servers with:

python monitor.py --update

The script will connect to each server, run apt-get update && apt-get upgrade -y (with output streamed so you see progress), check if a reboot is needed, and send you a notification via ntfy:

Updating 3 server(s)...

Updating prd-dokku...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:
Hit:1 https://mirror.hetzner.com/ubuntu/packages noble InRelease
Hit:2 https://download.docker.com/linux/ubuntu noble InRelease
Hit:3 https://mirror.hetzner.com/ubuntu/packages noble-updates InRelease
Hit:4 https://mirror.hetzner.com/ubuntu/packages noble-backports InRelease
Hit:5 https://mirror.hetzner.com/ubuntu/security noble-security InRelease
Hit:6 https://packagecloud.io/dokku/dokku/ubuntu noble InRelease
Reading package lists...
Reading package lists...
Building dependency tree...
Reading state information...
Calculating upgrade...
The following upgrades have been deferred due to phasing:
  libldap-common libldap2 python-apt-common python3-apt
The following packages have been kept back:
  linux-image-virtual
0 upgraded, 0 newly installed, 0 to remove and 5 not upgraded.
✅ prd-dokku updated successfully
   ⚠️ Reboot required

Updating prd-ghost...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:

[...]

Running kernel seems to be up-to-date.

Restarting services...
 systemctl restart qemu-guest-agent.service

Service restarts being deferred:
 /etc/needrestart/restart.d/dbus.service
 systemctl restart getty@tty1.service
 systemctl restart serial-getty@ttyS0.service
 systemctl restart systemd-logind.service
 systemctl restart unattended-upgrades.service

No containers need to be restarted.

No user sessions are running outdated binaries.

No VM guests are running outdated hypervisor (qemu) binaries on this host.
✅ prd-ghost updated successfully
   ⚠️ Reboot required

Updating prd-n8n...
Running apt update & upgrade (may take 5–15 min). Output streams below.

SSH key passphrase for id_rsa:
[...]

Running kernel seems to be up-to-date.

Restarting services...
 systemctl restart qemu-guest-agent.service

Service restarts being deferred:
 systemctl restart systemd-logind.service
 systemctl restart unattended-upgrades.service

No containers need to be restarted.

No user sessions are running outdated binaries.

No VM guests are running outdated hypervisor (qemu) binaries on this host.
✅ prd-n8n updated successfully
   No reboot needed

You will also get the notifications:

Wrapping Up

You don't have to choose between "no monitoring" and "full Prometheus/Grafana stack."

This middle ground, one Python script, SSH, and a few shell commands, gives you real visibility with minimal complexity.

The script you just built is production-ready for small setups. It's simple enough that you can debug it at 2 AM when something breaks, yet powerful enough to keep your servers healthy and your phone quiet. If you want to avoid scope-related bugs when extending it, Master Python Variable Scope is a concise, practical guide (LEGB, local vs global).

When you outgrow this approach (dozens of servers, complex dependencies, multiple teams), you'll know exactly what you need from a heavier system. Until then, keep it boring.

Download the complete script: https://github.com/nunombispo/server-monitoring-python-article

Happy monitoring! 🚀

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

ESP32-C3 SuperMini DHT11 Tutorial - Temperature & Humidity Sensor with MicroPython Web Server

Developer Service — Mon, 02 Feb 2026 07:54:01 +0000

Have you ever looked at a smart home dashboard and thought, “I could build this myself”? That itch to tinker, to understand what’s really going on behind the scenes, is what pulled me into building my own home monitoring setup from scratch.

I wanted a simple way to collect real environmental data using my own sensors, my own hardware, and my own code; no cloud dependency, no black boxes. So I started small. Using an ESP32-C3 SuperMini and a humble DHT11 temperature and humidity sensor, I built a tiny Wi-Fi-enabled node that reports real-time readings over serial and serves them through a minimal web interface.

This article is a follow-up to Getting Started with ESP32-C3 SuperMini and MicroPython. We’ll extend that foundation by adding a DHT11 sensor, reading data via GPIO, and exposing those readings both in the serial console and on a lightweight web server running directly on the ESP32.

By the end of this guide, you’ll know how to:

Wire a DHT11 to the ESP32-C3 SuperMini
Read temperature and humidity values using MicroPython
View live sensor data in the REPL / Serial Monitor
Run a tiny built-in web server for real-time readings
Use this setup as a building block for a DIY home-automation system

What You Need

To start, you will need some hardware components:

Item	Notes
ESP32-C3 SuperMini board	with MicroPython installed (see the previous article)
DHT11 temperature & humidity sensor module	digital single-wire sensor
Female-to-female Dupont connectors	for direct pin-to-pin connections
USB-C cable	for power and serial communication

Buying through these affiliate links helps keep this blog running and the projects coming. Thanks for the support!

Hardware Wiring & Pin Diagram

Here’s how to wire the ESP32-C3 SuperMini directly to the DHT11 sensor module:

 ESP32-C3 SuperMini     DHT11 Sensor
 ───────────────        ─────────────
 3V3  ─────────────►    VCC
 GND  ─────────────►    GND
 GPIO5 ────────────►    DATA

All connections are made pin-to-pin using female-to-female Dupont connectors, no breadboard or additional components required.

On the ESP32-C3 Super Mini you connect like this:

You can see the red wire connected to 3.3V, the brown wire connected to GND and the orange connected to the PIN5 on the ESP32.

For the sensor, you connect like this:

You can see the red wire connected to '+', the brown wire connected to '-' and the orange connected to 'out'.

That’s it, three wires total.

Why GPIO5?

GPIO5 is a regular, general-purpose I/O pin that works well for digital sensors like the DHT11. You can use other free GPIOs if needed, but it’s best to avoid pins with special boot or flashing functions. GPIO5 is a safe and convenient default.

Reading the DHT11 in MicroPython (Serial Output)

Before building the web server, let's start with a simple script to verify your sensor is working correctly. This will help you troubleshoot any wiring issues early.

The dht module is included in most MicroPython builds for ESP32, but if you encounter an import error, you may need to install it.

Step 1: Create the Sensor Reading Script

In Thonny or your preferred editor, save this as dht_read.py:

import time
import machine
import dht

sensor_pin = machine.Pin(5)
sensor = dht.DHT11(sensor_pin)

print("Starting DHT11 sensor readings...")
print("Press Ctrl+C to stop\n")

while True:
    try:
        # DHT11 needs at least 2 seconds between readings
        time.sleep(2)

        # Trigger a measurement
        sensor.measure()

        # Read the values
        temp = sensor.temperature()
        hum = sensor.humidity()

        # Print formatted output
        print(f"Temperature: {temp}°C   Humidity: {hum}%")

    except Exception as e:
        print(f"Read error: {e}")
        print("Check your wiring and pull-up resistor!")

Step 2: Upload and Run

Upload the file to your ESP32-C3 SuperMini using Thonny (File → Save As → MicroPython device).
Open the Serial Monitor (View → Shell in Thonny).
Run the script by pressing F5 or clicking Run.

You should see an output like this:

Starting DHT11 sensor readings...
Press Ctrl+C to stop

Temperature: 21°C   Humidity: 34%
Temperature: 21°C   Humidity: 34%
Temperature: 21°C   Humidity: 34%
Temperature: 21°C   Humidity: 34%
Temperature: 21°C   Humidity: 34%
...

Understanding the Output:

Temperature is in Celsius (you can convert to Fahrenheit: (temp * 9/5) + 32)
Humidity is a percentage (0-100%)
The DHT11 has ±2°C accuracy for temperature and ±5% for humidity
Readings update every 2 seconds (the sensor's minimum interval)

Important: The DHT11 can only update about every 2 seconds, so keep your loop delay ≥ 2 seconds. Reading faster than this will cause errors or return stale data.

Running a Simple Web Server

Now that you know the DHT11 is working, it’s time to expose those readings over the network. This is where the project really starts to feel like a building block for your own Home Assistant–style system.

You’ll connect the ESP32-C3 SuperMini to Wi-Fi and run a tiny web server that serves the latest temperature and humidity readings to any browser on your local network; no external services, no cloud required.

Step 1: Create the Web Server Script

This script combines Wi-Fi connectivity, sensor readings, and a basic HTTP server. Save it as main.py, MicroPython automatically runs this file on boot.

import socket
import network
import dht
import machine
import time

# ===== Wi-Fi Configuration =====
SSID = "your-ssid"
PASSWORD = "your-password"

wlan = network.WLAN(network.STA_IF)
wlan.active(True)
wlan.connect(SSID, PASSWORD)

print("Connecting to Wi-Fi...")
timeout = 20
while not wlan.isconnected() and timeout > 0:
    time.sleep(1)
    timeout -= 1
    print(".", end="")

if not wlan.isconnected():
    raise RuntimeError("Wi-Fi connection failed")

ip = wlan.ifconfig()[0]
print(f"\nConnected. IP address: {ip}")

# ===== Sensor Setup =====
sensor = dht.DHT11(machine.Pin(5))
print("DHT11 initialized on GPIO5")

# ===== Web Server Setup =====
addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(addr)
server.listen(1)

print(f"Web server running at http://{ip}/")

# ===== Main Server Loop =====
while True:
    try:
        client, addr = server.accept()
        print(f"Client connected: {addr[0]}")
        client.recv(1024)  # Read and ignore request

        sensor.measure()
        temp = sensor.temperature()
        hum = sensor.humidity()
        fahrenheit = (temp * 9 / 5) + 32

        html = f"""<!DOCTYPE html>
<html>
<head>
    <title>ESP32-C3 Sensor Node</title>
    <meta http-equiv="refresh" content="5">
    <style>
        body {{
            font-family: Arial, sans-serif;
            max-width: 600px;
            margin: 40px auto;
            padding: 20px;
            background: #f4f4f4;
        }}
        h1 {{ text-align: center; }}
        .card {{
            background: white;
            padding: 20px;
            border-radius: 8px;
            margin-bottom: 15px;
        }}
        .value {{
            font-size: 2em;
            color: #0066cc;
            font-weight: bold;
        }}
    </style>
</head>
<body>
    <h1>ESP32-C3 Sensor Node</h1>
    <div class="card">
        <strong>Temperature</strong>
        <div class="value">{temp} C ({fahrenheit:.1f} F)</div>
    </div>
    <div class="card">
        <strong>Humidity</strong>
        <div class="value">{hum} %</div>
    </div>
    <p style="text-align:center;color:#777;">
        Auto-refreshes every 5 seconds
    </p>
</body>
</html>
"""

        client.send("HTTP/1.0 200 OK\r\nContent-Type: text/html\r\n\r\n")
        client.send(html)
        client.close()

        print(f"Served: {temp}°C, {hum}%")

    except Exception as e:
        print("Server error:", e)
        try:
            client.close()
        except:
            pass
        time.sleep(1)

Let's understand what this code does:

Connects to Wi-Fi with a timeout to avoid blocking forever
Reads the DHT11 fresh on each HTTP request
Serves a simple HTML dashboard directly from the ESP32
Auto-refreshes every 5 seconds for near-real-time updates
Keeps running even if a request or read fails

This pattern, sensor → local web endpoint, is exactly how many DIY home-automation systems start.

Note: Don't forget to input your Wi-Fi credentials before running the script.

Step 2: Upload and Run

Upload main.py to your ESP32-C3 SuperMini
Reset the board
Watch the serial output for the assigned IP address
Open a browser on the same network
Visit the indicated IP address

For reference, this is an example of the serial output on Thonny:

MPY: soft reboot
Connecting to Wi-Fi...
...
Connected. IP address: 192.168.2.73
DHT11 initialized on GPIO5
Web server running at http://192.168.2.73/

Opening the browser on the indicated address, you should see the web page showing live temperature and humidity readings, updating automatically every 5 seconds.

Conclusion

You’ve just built your first Wi-Fi-enabled sensor node, and more importantly, you’ve laid the groundwork for your own DIY home-automation system.

With the ESP32-C3 SuperMini and a simple DHT11 sensor, you now have a device that can read real-world data, connect to your network, and expose that data through a web interface, all running locally on MicroPython. No cloud services, no vendor lock-in, just hardware and code you control.

The concepts covered here, GPIO wiring, sensor polling, Wi-Fi connectivity, and serving data over HTTP, are the same building blocks used in larger systems like Home Assistant, MQTT-based sensor networks, and custom dashboards. From here, it’s easy to imagine adding more sensors, exposing a JSON API, or aggregating multiple nodes into a central controller.

Projects like this are even more fun when they’re shared. If you know someone who enjoys tinkering with hardware, building smart-home setups, or learning MicroPython, feel free to pass this along, it might be the starting point for their own sensor network.

Happy hacking.

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 a captive portal in ESP32 with MicroPython

Developer Service — Mon, 26 Jan 2026 08:34:32 +0000

You unbox a small IoT device, plug it in, and… nothing.

No screen. No keyboard. No buttons that make any sense.

This is a familiar situation if you've ever worked with microcontrollers like the ESP32. The hardware is powerful, but the very first interaction is often awkward. At some point, the device needs to connect to your Wi-Fi network, and the classic question appears:

"How does a tiny device ask for Wi-Fi credentials?"

Hardcoding credentials works for quick experiments, but it breaks down in the real world, every deployment would require re-flashing the firmware. Serial consoles are fine for developers, not for users who don't have USB cables or the patience for terminal commands. Mobile apps add friction and maintenance overhead, requiring separate iOS and Android versions that need constant updates.

What you want instead is simple: power on the device, connect to it once from any phone or laptop, configure it through a familiar web interface, and never think about it again. No apps to install, no cables to find, no special tools required.

This is exactly where captive portals shine, they turn Wi-Fi provisioning into a one-time, friction-free experience that works on every platform.

What Is a Captive Portal?

If you've ever connected to Wi-Fi at a coffee shop, hotel, or airport, you've used a captive portal.

You join the network, and a web page automatically pops up asking you to log in or accept terms.

You didn't type a URL—the network guided you there by intercepting your browser traffic.

A captive portal on an embedded device works the same way:

The device creates its own Wi-Fi network (acting as an access point)
The user connects from their phone or laptop (no password needed)
The device intercepts all internet requests using a DNS server
A web page automatically opens showing the configuration form
After submission, the device saves credentials and switches to normal Wi-Fi client mode

The beauty of this approach is that the ESP32 can handle everything on-device.

It acts as both an access point and a web server simultaneously, with no cloud services, external infrastructure, or internet connection required during setup. It's significantly simpler than building a companion mobile app (which requires maintaining iOS and Android versions) and far more user-friendly than hardcoded credentials or serial console configuration.

For IoT products, this is often the difference between "technically works" and "actually ships to customers."

Architecture Overview

The device goes through two distinct phases: setup mode (captive portal) and normal operation (Wi-Fi client).

Setup Mode runs the first time you power on the device, or anytime no Wi-Fi configuration exists.

In this mode, the ESP32:

Creates its own temporary Wi-Fi network
Runs a DNS server that redirects all domains to itself
Runs an HTTP server that serves the configuration page
Saves user-submitted credentials to flash storage

Normal Operation begins after successful configuration.

In this mode, the ESP32:

Disables the access point
Connects to the user's Wi-Fi network as a client
Runs your actual application (fetching data, controlling hardware, etc.)

The key decision happens at boot: does config.json exist? If yes, connect to Wi-Fi. If no, start the captive portal.

If you have already MicroPython installed on your ESP32, you are good to go and experiment with the following examples.

If not, see this guide for more detail on how to setup an ESP32 with MicroPython.

Step 1 - Create the Wi-Fi Access Point

The ESP32 creates its own Wi-Fi network using the built-in network module. This is the foundation of the entire captive portal, without the access point, users have no way to reach the configuration page.

The ESP32 has two Wi-Fi interfaces:

AP_IF (Access Point Interface) - Makes the ESP32 act like a router
STA_IF (Station Interface) - Makes the ESP32 act like a Wi-Fi client

During setup mode, we use AP_IF to create a temporary network that users can join from their phones.

In your local project, create a new file wifi_ap.py:

import network

def start_access_point():
    ap = network.WLAN(network.AP_IF)
    ap.active(True)

    ap.config(
        essid="MyDevice-Setup",
        authmode=network.AUTH_OPEN
    )

    print("Access point active:", ap.active())
    print("AP IP address:", ap.ifconfig()[0])

    return ap

Once this runs, the ESP32 broadcasts a Wi-Fi network named "MyDevice-Setup" and assigns itself IP address 192.168.4.1. Any device that connects to this network can communicate with the ESP32 using this IP.

Why use an open network?

You'll notice we use AUTH_OPEN (no password). This might seem insecure, but it's actually the right choice for a temporary setup network:

Better compatibility: Some devices have issues with WPA2 on captive portals
Faster setup: Users don't need to type a password
Auto-detection works better: Operating systems are more likely to show the "Sign in to network" popup on open networks
Temporary by design: The AP only exists during initial setup and disappears once configured

The security risk is minimal because:

The network only exists during the 1-2 minutes of setup
The user is typically standing right next to the device
Once configured, the AP shuts down completely

For production devices, you can add strategies like auto-disabling the AP after 10 minutes, or requiring a physical button press to re-enter setup mode.

Step 2 - Build the HTML Setup Page

Before we create the HTTP server, let's build the setup page that users will see. This is the critical user-facing component, it needs to work flawlessly on every mobile device.

Design Principles for Captive Portal Pages:

Mobile-first: Most users will access this from their phones
Minimal dependencies: No external CSS/JS frameworks (we don't have internet!)
Fast loading: Every byte travels over a slow access point connection
Clear purpose: Users should immediately understand what to do
Forgiving inputs: Don't assume users type perfectly

The page needs only two inputs: SSID and password. Everything else can be configured later through a different interface (if needed at all).

On your local project folder, create portal.html:

<!DOCTYPE html>
<html>
<head>
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Device Setup</title>
    <style>
        body {
            font-family: sans-serif;
            padding: 16px;
            max-width: 400px;
            margin: 0 auto;
        }
        h2 {
            margin-top: 0;
        }
        label {
            display: block;
            margin-top: 12px;
            font-weight: bold;
        }
        input, button {
            width: 100%;
            padding: 10px;
            margin-top: 6px;
            font-size: 16px;
            box-sizing: border-box;
        }
        button {
            margin-top: 20px;
            background: #007bff;
            color: white;
            border: none;
            cursor: pointer;
        }
        button:hover {
            background: #0056b3;
        }
    </style>
</head>
<body>
    <h2>Wi-Fi Setup</h2>

    <form method="POST" action="/save">
        <label>
            Wi-Fi SSID
            <input name="ssid" required>
        </label>

        <label>
            Password
            <input name="password" type="password">
        </label>

        <button type="submit">Save & Connect</button>
    </form>
</body>
</html>

Design notes:

viewport meta tag ensures proper scaling on mobile devices
Inline CSS keeps everything in one file (no extra HTTP requests)
box-sizing: border-box prevents input overflow issues
Large touch targets (10px padding) work better on phones
Simple color scheme (#007bff) looks professional without complexity

Step 3 - Create HTTP Server with Form Handling

Now create an HTTP server that serves the HTML file and handles form submissions.

This is more complex than a typical web server because it needs to:

Serve the same page for any URL - Operating systems make requests to various paths like /generate_204, /hotspot-detect.html, etc. to detect captive portals
Handle form submissions - Parse POST data and save credentials
Provide error feedback - Show clear messages when things go wrong
Be resilient - The ESP32's networking stack is fragile, so we need careful error handling

The server handles both GET requests (show the form) and POST requests (save configuration).

On your local project folder, create http_server.py:

import socket
import json

def parse_form_data(body):
    """Parse URL-encoded form data"""
    data = {}
    pairs = body.split("&")
    for pair in pairs:
        if "=" in pair:
            key, value = pair.split("=", 1)
            # URL decode
            data[key] = value.replace("+", " ").replace("%40", "@")
    return data

def save_config(data):
    """Save Wi-Fi credentials to config.json"""
    if not data.get("ssid"):
        raise ValueError("SSID is required")

    config = {
        "ssid": data.get("ssid"),
        "password": data.get("password", ""),
    }

    with open("config.json", "w") as f:
        json.dump(config, f)

    print("Configuration saved:", config)

def start_http_server():
    addr = socket.getaddrinfo("0.0.0.0", 80)[0][-1]
    s = socket.socket()
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(addr)
    s.listen(1)

    print("HTTP server listening on port 80")

    while True:
        try:
            conn, addr = s.accept()
            request = conn.recv(2048).decode()

            if request.startswith("POST /save"):
                # Handle form submission
                body = request.split("\r\n\r\n", 1)[1]
                form_data = parse_form_data(body)
                save_config(form_data)

                response_body = """<!DOCTYPE html>
<html>
<head>
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Setup Complete</title>
    <style>
        body { font-family: sans-serif; padding: 16px; text-align: center; }
        h2 { color: #28a745; }
    </style>
</head>
<body>
    <h2>Configuration saved</h2>
    <p>Device will reboot and connect to Wi-Fi.</p>
    <p><small>You can close this window.</small></p>
</body>
</html>"""
            else:
                # Serve the setup page for any other request
                try:
                    with open("portal.html") as f:
                        response_body = f.read()
                except:
                    response_body = "<h1>Setup Page Not Found</h1><p>Please upload portal.html</p>"

            response = (
                "HTTP/1.1 200 OK\r\n"
                "Content-Type: text/html\r\n"
                "Connection: close\r\n\r\n"
                + response_body
            )

            conn.send(response.encode())

        except Exception as e:
            print("HTTP error:", e)
            try:
                error = f"<h2>Error</h2><p>{e}</p>"
                conn.send(
                    "HTTP/1.1 500 Internal Server Error\r\n"
                    "Content-Type: text/html\r\n"
                    "Connection: close\r\n\r\n"
                    .encode() + error.encode()
                )
            except:
                pass
        finally:
            try:
                conn.close()
            except:
                pass

Key implementation details:

Form parsing: The parse_form_data() function handles URL-encoded data (the format HTML forms use). We manually decode common characters like + (space) and %40 (@) since MicroPython doesn't have a built-in URL decoder.

File handling: The server reads portal.html from the filesystem. If the file is missing, it shows an error message instead of crashing, this helps with debugging during development.

Connection management: Connection: close is critical. MicroPython's socket implementation can leak memory with persistent connections. We explicitly close every connection, even in error cases, to prevent the ESP32 from running out of sockets.

Error handling: The triple-nested try/except blocks might look excessive, but they prevent the server from crashing when:

A client disconnects mid-request
The request is malformed
The filesystem is corrupted
Memory runs low

Response encoding: Notice we call .encode() when sending responses. MicroPython's socket.send() expects bytes, not strings.

Step 4 - Add DNS Redirection (The "Captive" Part)

To trigger the automatic "Sign in to network" popup on phones, we need a DNS server that redirects all domains to the ESP32's IP. This is the "captive" part of "captive portal", we're capturing all DNS requests.

How DNS redirection works:

When your phone connects to a new Wi-Fi network, it performs a "captive portal check" by trying to reach a known URL (like http://captive.apple.com on iOS or http://connectivitycheck.gstatic.com on Android).

Here's what happens:

Phone asks DNS: "What's the IP address of captive.apple.com?"
Our DNS server lies: "It's 192.168.4.1" (the ESP32's IP)
Phone tries to load that page
Our HTTP server responds with the setup page
Phone realizes: "This isn't the real captive.apple.com, must be a captive portal!"
Phone shows the "Sign in to network" popup

The key insight: we don't need to know which domain was requested. We just answer everything with our own IP address.

Create dns_server.py:

import socket

AP_IP = "192.168.4.1"

def ip_to_bytes(ip):
    return bytes([int(x) for x in ip.split('.')])

def start_dns_server():
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind((AP_IP, 53))
    print("DNS server listening on", AP_IP)

    while True:
        try:
            data, addr = sock.recvfrom(512)
            if len(data) < 12:
                continue

            tid = data[:2]
            flags = b"\x81\x80"
            qdcount = b"\x00\x01"
            ancount = b"\x00\x01"
            nscount = b"\x00\x00"
            arcount = b"\x00\x00"

            dns_header = tid + flags + qdcount + ancount + nscount + arcount

            # Parse the question section
            idx = 12
            while idx < len(data) and data[idx] != 0:
                idx += 1
            idx += 5
            question = data[12:idx]

            # Respond only to A-record queries
            qtype = data[idx-4:idx-2]
            if qtype != b"\x00\x01":
                sock.sendto(dns_header + question, addr)
                continue

            # Build answer pointing to AP_IP
            answer = (
                b"\xc0\x0c"
                b"\x00\x01"
                b"\x00\x01"
                b"\x00\x00\x00\x3c"
                b"\x00\x04"
                + ip_to_bytes(AP_IP)
            )

            response = dns_header + question + answer
            sock.sendto(response, addr)

        except Exception as e:
            print("DNS error:", e)

How the DNS packet parsing works:

DNS packets have a specific structure. We extract:

Transaction ID (tid): Unique identifier for each query, must be echoed back
Flags: 0x8180 means "this is a response, no error"
Question section: The domain name the client asked about
Answer section: Our fabricated response pointing to 192.168.4.1

We only respond to A-record queries (IPv4 address lookups). Other query types (AAAA for IPv6, MX for mail servers, etc.) are safely ignored.

Platform compatibility notes:

iOS is particularly strict about captive portal detection. The popup appears more reliably if:

DNS replies are fast (our implementation is)
HTTP responses are well-formed (we use proper HTML)
The server is already running before the phone connects

If the popup doesn't appear automatically, users can manually open any browser and type any URL (like example.com). The DNS redirection will still work, bringing them to your setup page.

Important: Captive Portal Auto-Detection Limitations

The automatic captive portal popup is not guaranteed to appear. Whether it shows up depends on:

Operating system: iOS is most reliable, Android varies by manufacturer (Samsung/Xiaomi less reliable), Windows 10/11 is hit-or-miss
Timing: The popup only appears if DNS/HTTP checks happen within a few seconds of connecting
Network history: If the phone previously connected to this SSID, it may skip checks
Phone settings: Some phones have "captive portal detection" disabled in developer options
Carrier settings: Some mobile carriers modify captive portal behavior

In real-world testing, automatic popups appear on approximately 70-80% of devices. The other 20-30% require users to manually open a browser.

Step 5 - Connect to User's Wi-Fi

After saving configuration, the device needs to switch from AP mode to STA mode (Wi-Fi client).

This is the transition from "setup" to "normal operation."

The connection logic needs to be robust because Wi-Fi connections fail more often than you'd expect:

Wrong password (most common)
Weak signal strength
Router configured to block new devices
Network congestion
5GHz vs 2.4GHz band issues (ESP32 only supports 2.4GHz)

Our implementation tries for 20 seconds (20 attempts × 1 second), which is long enough for most networks but not so long that users think the device is frozen.

Create wifi_client.py:

import network
import time
import json

def connect_to_wifi():
    """Try to connect to saved Wi-Fi network"""
    try:
        with open("config.json") as f:
            config = json.load(f)
    except:
        print("No config file found")
        return False

    ssid = config.get("ssid")
    password = config.get("password", "")

    if not ssid:
        print("No SSID in config")
        return False

    # Disable AP mode
    ap = network.WLAN(network.AP_IF)
    ap.active(False)

    # Enable STA mode
    sta = network.WLAN(network.STA_IF)
    sta.active(True)
    sta.connect(ssid, password)

    print(f"Connecting to {ssid}...")

    for _ in range(20):
        if sta.isconnected():
            print("Connected!")
            print("IP:", sta.ifconfig()[0])
            return True
        time.sleep(1)

    print("Connection failed")
    return False

Connection flow breakdown:

Check for config: Try to load config.json. If it doesn't exist, return False immediately
Disable AP: Turn off the access point interface (can't be AP and STA simultaneously effectively)
Enable STA: Activate the station (client) interface
Connect: Attempt to connect using saved credentials
Wait and retry: Check connection status every second for up to 20 seconds
Report result: Return True if connected, False if timeout

Why the 20-second timeout?

Most successful connections happen in 2-5 seconds
Some networks with complex auth take 10-15 seconds
Anything longer probably indicates a real problem (wrong password, etc.)
20 seconds is long enough to succeed but short enough that users don't think the device is frozen

Putting It All Together

Now let's create a complete main.py that integrates all the components. This is the entry point that decides: "Should I start the captive portal, or connect to Wi-Fi?"

The logic is simple:

Try to connect to saved Wi-Fi
If successful → run the main application
If failed → start captive portal for (re)configuration

This creates a self-healing system: if the user changes their Wi-Fi password or moves the device to a new network, it automatically falls back to setup mode.

import network
import time
import _thread
import machine

# Import our modules
from wifi_ap import start_access_point
from http_server import start_http_server
from dns_server import start_dns_server
from wifi_client import connect_to_wifi

def start_captive_portal():
    """Start the captive portal in AP mode"""
    print("\n=== Starting Captive Portal ===")

    # Create access point
    ap = start_access_point()
    time.sleep(1)  # Give AP time to start

    # Start DNS server in background thread
    _thread.start_new_thread(start_dns_server, ())
    time.sleep(0.5)

    # Start HTTP server (blocks here)
    print("Captive portal ready!")
    print("Connect to 'MyDevice-Setup' from your phone\n")
    start_http_server()

def main():
    print("\n" + "="*40)
    print("ESP32 Captive Portal Starting...")
    print("="*40 + "\n")

    # Try to connect to saved Wi-Fi network
    if connect_to_wifi():
        print("\n=== Connected to Wi-Fi ===")
        print("Starting main application...\n")

        # Your main application code here
        try:
            import urequests
            print("Testing internet connection...")
            response = urequests.get("https://worldtimeapi.org/api/ip")
            if response.status_code == 200:
                data = response.json()
                print(f"Success! Current time: {data.get('datetime', 'N/A')}")
            response.close()
        except Exception as e:
            print(f"Internet test failed: {e}")

        # Your app continues here...
        print("\nReady for normal operation!")

    else:
        # No config or connection failed - start captive portal
        print("\n=== Starting Setup Mode ===")
        start_captive_portal()

# Run the main function
if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n\nShutting down...")
    except Exception as e:
        print(f"\nFatal error: {e}")
        print("Rebooting in 5 seconds...")
        time.sleep(5)
        machine.reset()

Code walkthrough:

start_captive_portal(): This function orchestrates setup mode:

Starts the access point
Launches the DNS server in a background thread (using _thread)
Starts the HTTP server in the foreground (which blocks indefinitely)

The threading is important: both servers need to run simultaneously. The DNS server runs in the background while the HTTP server handles the main event loop.

main(): This is the decision point:

Calls connect_to_wifi() to attempt connection
On success: runs your application code (we show a simple internet connectivity test)
On failure: starts the captive portal

Error handling: The try/except blocks catch:

KeyboardInterrupt: Allows clean shutdown during development
General exceptions: Reboots the device to recover from crashes

File Structure

Your ESP32 should have these files in its root directory:

/
├── main.py           # Main entry point
├── wifi_ap.py        # Access point setup
├── http_server.py    # HTTP server with form handling
├── dns_server.py     # DNS server for captive portal
├── wifi_client.py    # Wi-Fi connection logic
└── portal.html       # Setup page HTML

Why modular files?

Breaking the code into separate modules has several advantages:

Easier debugging: Test each component independently
Cleaner code: Each file has one clear purpose
Reusability: Copy dns_server.py to other projects
Maintainability: Easier to find and fix bugs

All Python files should be in the root directory. MicroPython doesn't handle complex directory structures well, so keep it simple.

Full source code is available at: https://github.com/nunombispo/ESP32-Captive-Portal

Upload Files to ESP32

The easiest way to upload files to your ESP32 is using Thonny IDE, which has built-in MicroPython support.

Install Thonny: Download from thonny.org

Connect ESP32: Plug in your ESP32 via USB

Configure Thonny:

Go to Tools → Options → Interpreter
Select "MicroPython (ESP32)"
Choose your COM port (e.g., COM3 on Windows, /dev/ttyUSB0 on Linux)
Click OK

Upload Files:

In Thonny, go to View → Files
Open your local project folder in the top section
Right-click on each file and select 'Upload to /' (this will upload the file to the device)

Run:

Press the Stop/Restart button (or Ctrl+D in the Shell)
Click the green run button

Demo: Real-World Usage

Here's what the complete flow looks like in practice:

Step 1: Device Powers On

The ESP32 boots and starts the captive portal. In the serial monitor (Thonny's Shell), you'll see:

========================================
ESP32 Captive Portal Starting...
========================================

No config file found

=== Starting Setup Mode ===

=== Starting Captive Portal ===
Access point active: True
AP IP address: 192.168.4.1
DNS server listening on 192.168.4.1
Captive portal ready!
Connect to 'MyDevice-Setup' from your phone

HTTP server listening on port 80

Step 2: Connect from Phone

On your phone's Wi-Fi settings, you'll see "MyDevice-Setup" appear as an open network. Tap to connect.

Expected behavior:

Best case: A popup appears within 3-5 seconds saying "Sign in to network"
Common case: No popup appears - this is normal! Just open any browser and type any URL

Step 3: Configuration Page

Whether you got the popup or opened a browser manually, you'll see the setup page with two fields:

Wi-Fi SSID (your network name)
Password

Enter your actual Wi-Fi credentials and click "Save & Connect"

Step 4: Success Confirmation

You'll see a page saying "Configuration saved" with instructions to reboot the device.

In the Thonny shell, you can see the configuration save log:

HTTP server listening on port 80
Configuration saved: {'ssid': 'SSID_EXAMPLE', 'password': 'password example'}

Step 5: Normal Operation

Reboot the ESP32 and after reboot, the serial monitor shows:

========================================
ESP32 Captive Portal Starting...
========================================

Connecting to SSID_EXAMPLE...
Connected!
IP: 192.168.2.73

=== Connected to Wi-Fi ===
Starting main application...

Testing internet connection...
Success! Current time: 2026-01-23T10:03:12.498011+01:00

Ready for normal operation!

Conclusion

Captive portals solve a fundamental problem: getting a headless device onto a user's network without hardcoded credentials or mobile apps.

For ESP32 projects, it's the difference between a prototype and a product someone else can actually use.

The ESP32 is powerful enough to run an access point, DNS server, and web server simultaneously, making Wi-Fi provisioning a simple, app-free experience.

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

Forem: Developer Service

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

What You Will Build

Why Microdot?

System Overview

Hardware

Safety

Wiring Summary

Software: Project Structure

sensor.py - Hardware Reads

web.py - Routes and Inline UI

main.py - Boot and Server Start

What to Expect During a Charge Session

Conclusion

Python Decorators - The Three-Layer Pattern

Decorators Under the Hood

The identity problem

The Three Layers of a Parametrized Decorator

Why the extra layer exists

Closure capture

Building Your First Parameterized Decorator

Decorators You'll Actually Ship

@retry - handling flaky operations

@timer - measuring execution time

@require_role - protecting endpoints

Stacking Decorators

Conclusion

Build a Terminal Disk Analyzer in Python with Textual

What We're Building

Architecture

Step 1 - The Data Model and Scanner

Step 2 - Building the TUI with Textual

Step 3 - Non-Blocking Scan with @work

Step 4 - Rendering the Table

Step 5 - Delete with Confirmation

Step 6 - CLI Entrypoint with Typer

Conclusion

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

Key Concepts

Your First Workflow: Running Tests on Every Push

The Python Project

Creating the Workflow File

Line-by-Line Walkthrough

Pushing and Reading the Results

Making It Smarter

Linting with ruff

Caching pip Dependencies

Testing Across Multiple Python Versions

Type Checking with mypy (Optional)

Common Pitfalls

YAML Indentation Errors

Wrong Branch Name or Event Trigger

Forgetting requirements.txt

Hardcoding Credentials

Conclusion

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

nunombispo / CYD-OpenF1

CYD F1 Pit Wall Display

Hardware

Dependencies

Configuration

One Endpoint, One Request

Part 1: The Aggregator

Positions

Tyres

Gaps

Formatting and total laps

The Response

Part 2: The Display

Layout

Rendering

Mapping the response: parse_pitwall

Part 3: The Main Loop

Result

What works well

What doesn't

What's Next

Conclusion

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

The Project: A Tiny File Database

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