DEV Community: Haji Rufai

I Built an LSM-Tree in Pure Python and Finally Understood How RocksDB and Cassandra Handle Millions of Writes

Haji Rufai — Thu, 11 Jun 2026 06:32:59 +0000

I have built a write-ahead log and a bloom filter from scratch in the last week. Today I wanted to put them together into something bigger.

So I built an LSM-tree. An LSM-tree is the storage engine behind RocksDB, Cassandra, LevelDB, ScyllaDB, and the metadata layer of a dozen databases you use every day. If you have ever wondered why these systems can swallow millions of writes per second, this is the answer.

The whole thing fits in about 150 lines of pure Python. No dependencies. By the end you will understand exactly why an LSM-tree trades read speed for absurd write speed, and how it claws some of that read speed back.

The problem LSM-trees solve

A B-tree updates data in place. Every write finds the right page on disk and modifies it. That means random I/O, and random I/O on spinning disks or even SSDs is slow.

The LSM-tree makes a different bet. What if you never updated anything in place? What if every write was just an append to the end of a file?

Appends are the fastest thing a disk can do. The catch is that reads become harder, because the latest value for a key could be anywhere. The genius of the LSM-tree is the set of tricks that make those reads fast again.

Here is the shape of the design:

Writes go to an in-memory sorted structure called a memtable.
When the memtable fills up, it gets flushed to disk as a sorted, immutable file called an SSTable.
Reads check the memtable first, then the SSTables newest to oldest.
A background process merges SSTables together to keep reads fast. This is compaction.

Let me build each piece.

The memtable

The memtable is just a sorted dictionary in memory. Writes are O(log n) inserts. I will use a plain dict and sort on flush, which keeps the code simple and is fine for our purposes.

class Memtable:
    def __init__(self):
        self.data = {}
        self.size_bytes = 0

    def put(self, key, value):
        self.data[key] = value
        self.size_bytes += len(key) + len(value)

    def get(self, key):
        return self.data.get(key)

    def items_sorted(self):
        return sorted(self.data.items())

Notice there is no delete method that removes a key. In an LSM-tree you never truly delete. Instead you write a special marker called a tombstone. I will represent it with a sentinel value.

TOMBSTONE = "__tombstone__"

def delete(self, key):
    self.put(key, TOMBSTONE)

Why a tombstone instead of removing the key? Because an older SSTable on disk might still have the real value. If you just dropped the key from memory, an old version would resurface on the next read. The tombstone says "this key is dead" and shadows everything older. Compaction cleans it up later.

The SSTable

When the memtable gets too big, we flush it to disk as an SSTable. The key property is that the entries are sorted by key. Sorted order is what makes everything else efficient.

import json

def flush_memtable(memtable, path):
    with open(path, "w") as f:
        for key, value in memtable.items_sorted():
            f.write(json.dumps([key, value]) + "\n")

That is it. A flat file of sorted key-value pairs, one per line. Real systems use a binary format with a block index, but the principle is identical. Sorted on disk, immutable once written.

Immutability is a feature, not a limitation. Because an SSTable never changes after it is written, there are no locks, no in-place updates, and no torn writes. Concurrent readers can scan it freely while new data lands in fresh files.

Reading from an SSTable

To read a key, we scan the file. Because it is sorted we can stop early once we pass the key we want.

def sstable_get(path, target):
    with open(path) as f:
        for line in f:
            key, value = json.loads(line)
            if key == target:
                return value
            if key > target:
                return None  # sorted, so it cannot appear later
    return None

A linear scan sounds slow, and it is. This is where the bloom filter from my last post earns its keep. Before scanning an SSTable, you ask its bloom filter "could this key be here?" If the answer is no, you skip the entire file without touching disk. Most negative lookups never read a single SSTable byte.

Tying it together: the LSM engine

Now the full engine. Writes hit the memtable. When the memtable crosses a size threshold, we flush it and start a new one. Reads walk from newest data to oldest and stop at the first hit.

import os
import time

class LSMTree:
    def __init__(self, directory, flush_threshold=4096):
        self.directory = directory
        self.flush_threshold = flush_threshold
        self.memtable = Memtable()
        self.sstables = []  # newest last
        os.makedirs(directory, exist_ok=True)

    def put(self, key, value):
        self.memtable.put(key, value)
        if self.memtable.size_bytes >= self.flush_threshold:
            self._flush()

    def delete(self, key):
        self.put(key, TOMBSTONE)

    def _flush(self):
        path = os.path.join(self.directory, f"sstable_{time.time_ns()}.log")
        flush_memtable(self.memtable, path)
        self.sstables.append(path)
        self.memtable = Memtable()

    def get(self, key):
        # 1. memtable is freshest
        value = self.memtable.get(key)
        if value is not None:
            return None if value == TOMBSTONE else value
        # 2. SSTables, newest to oldest
        for path in reversed(self.sstables):
            value = sstable_get(path, key)
            if value is not None:
                return None if value == TOMBSTONE else value
        return None

The ordering in get is the whole game. The memtable holds the most recent writes, so it wins. Then we check SSTables from newest to oldest. The first time we find the key, that is the current value, and we stop. A tombstone counts as a hit and returns None, which is how deletes work across files.

Let me prove it behaves like a real key-value store.

db = LSMTree("./data", flush_threshold=64)

db.put("user:1", "alice")
db.put("user:2", "bob")
db.put("user:1", "alice_updated")  # newer write shadows the old one
db.delete("user:2")                # tombstone

print(db.get("user:1"))  # alice_updated
print(db.get("user:2"))  # None
print(db.get("user:9"))  # None

Even after the memtable flushes to disk and later writes land in newer SSTables, the read path resolves the right version every time, because newer always shadows older.

The cost nobody warns you about: read amplification

Run that engine for a while and you will end up with hundreds of SSTables. A read for a missing key might have to check every one of them. That is read amplification, and it is the dark side of the write-optimized design.

Two things fight it. The bloom filter skips files that cannot contain the key. And compaction merges many small SSTables into fewer big ones.

Compaction

Compaction reads several SSTables, merges them by key keeping only the newest value for each, drops tombstones, and writes one fresh SSTable. Because the inputs are already sorted, this is a k-way merge.

def compact(self):
    merged = {}
    # oldest first so newer values overwrite older ones
    for path in self.sstables:
        with open(path) as f:
            for line in f:
                key, value = json.loads(line)
                merged[key] = value
    # drop tombstones during the final compaction
    clean = {k: v for k, v in merged.items() if v != TOMBSTONE}

    new_path = os.path.join(self.directory, f"sstable_{time.time_ns()}.log")
    tmp = Memtable()
    for k, v in clean.items():
        tmp.put(k, v)
    flush_memtable(tmp, new_path)

    for path in self.sstables:
        os.remove(path)
    self.sstables = [new_path]

After compaction, dozens of files collapse into one, stale versions vanish, and deleted keys are physically gone. Reads get fast again. The price is the I/O spent rewriting data, which is called write amplification. Tuning the balance between read amplification, write amplification, and space amplification is the entire art of running RocksDB or Cassandra in production. The RUM conjecture says you can optimize for at most two of read, update, and memory at once. You never get all three.

Where the durability comes from

You might have noticed a gap. If the process crashes while data is still in the memtable, those writes are gone. The memtable lives in RAM.

This is exactly where the write-ahead log comes in. Before a write touches the memtable, you append it to the WAL on disk. On restart you replay the WAL to rebuild the memtable. The WAL plus the memtable plus SSTables plus bloom filters plus compaction is the complete LSM-tree. Every piece I have built this week was a part of this one machine, and I did not plan it that way when I started.

What this taught me

The LSM-tree is not a clever data structure. It is a clever pipeline. Each component is simple. The memtable is a dict. An SSTable is a sorted file. A bloom filter is a bit array. Compaction is a merge. The power comes from how they compose.

That is the real lesson of building storage engines by hand. The famous systems are not magic. They are a handful of boring ideas wired together with care, and you can build the toy version in an afternoon.

If you want to actually understand the database you depend on, stop reading the docs and build the smallest version that runs. You will never look at RocksDB the same way again.

I am publishing one of these from-scratch builds every day. Follow along if you want to understand the systems under your stack one toy at a time, and tell me in the comments which engine I should tear apart next.

I built a feature store in pure Python to finally understand the point-in-time join

Haji Rufai — Wed, 10 Jun 2026 03:50:46 +0000

For a long time the phrase "feature store" sat in my head as a box on an architecture diagram. Feast, Tecton, Databricks Feature Store. I knew they served features to models and I knew they were a Real Thing that ML platform teams care about, but I could not have told you what the hard part actually was. So I did the thing that always works for me: I rebuilt a tiny version from scratch, with no pandas and no pyarrow, just the Python standard library and sqlite. The project is called Asof, and building it taught me that the whole category exists to solve a single deceptively nasty problem.

That problem is the point-in-time join, and once it clicked I could not unsee it.

Live demo and system map: https://hajirufai.github.io/asof/
Code: https://github.com/hajirufai/asof

The one problem a feature store is really for

Say you are training a churn model. You have a label: customer c1 churned on January 15. You want to attach features to that label, things like the customer's rolling 7 day spend, how many orders they placed recently, their account age. Those features change over time, so you have a whole history of them.

Here is the trap. If you join "customer c1's features" to "the January 15 label" by just grabbing c1's most recent feature row, you will probably grab a row from January 20 or February 3, because that is the latest one sitting in your table. You just trained your model on data from the future of the thing it is trying to predict. Your offline accuracy looks fantastic. Then you ship it, and in production the model only ever sees the past, and it falls apart.

That gap between what your training pipeline saw and what your serving pipeline sees has a name: train/serve skew. The specific flavour above, using data from after the label timestamp, is label leakage. The point-in-time join is the fix, and it is the thing a feature store gets right so you do not have to remember to.

The rule is simple to say and easy to get wrong: for a label at time t, use the most recent feature value with a timestamp less than or equal to t, and nothing newer.

What Asof actually is

Asof keeps two stores, which is the shape every feature store has:

an offline store, the full timestamped history of every feature, used to build training sets, and
an online store, just the latest value per entity, used to serve a model at inference where latency matters.

flowchart LR
  S[CSV / JSONL sources] --> FV[Feature views]
  FV --> OFF[(Offline store<br/>full history)]
  OFF -->|get_historical_features<br/>as-of join| TRAIN[Point-in-time<br/>training set]
  OFF -->|materialize| ON[(Online store<br/>latest per entity)]
  ON -->|get_online_features| SERVE[Feature server to model]

You register definitions, load history, and then two operations carry the weight. get_historical_features builds a training set with the as-of join. get_online_features serves the freshest values. In between, materialize sweeps the offline history into the online store.

Defining things looks like this:

from datetime import timedelta
from asof import FeatureStore, Entity, FeatureView, Field, ValueType

store = FeatureStore()

customer = Entity("customer", join_key="customer_id")
view = FeatureView(
    name="customer_stats",
    entity=customer,
    schema=[Field("rolling_7d_spend", ValueType.FLOAT), Field("tier", ValueType.STRING)],
    ttl=timedelta(days=14),
)
store.apply(customer, view)

The as-of join, the slow way and the fast way

The obviously correct version is a nested loop. For every label row, scan every feature row for that entity, throw away anything newer than the label, throw away anything older than the TTL, keep the latest of what remains. That is O(n times m) and it is the version I wrote first, on purpose, because it is impossible to get subtly wrong.

for ent_row in entity_rows:
    best = None
    for fr in feature_rows:
        if fr[join_key] != ent_row[join_key]:
            continue
        if fr["event_timestamp"] > ent_row["event_timestamp"]:
            continue  # this is the future. never use it.
        if ttl and (ent_row["event_timestamp"] - fr["event_timestamp"]) > ttl:
            continue  # too old, expired
        if best is None or fr["event_timestamp"] >= best["event_timestamp"]:
            best = fr

That is fine for a unit test and hopeless for real data. The fast version groups both sides by entity key, sorts each side by timestamp, and then sweeps a single pointer through the sorted feature events as the sorted label times move forward, remembering the most recent valid row as it goes. After the sort it is one linear pass, O(n plus m).

ent_list.sort(key=lambda pair: pair[1]["event_timestamp"])
feats.sort(key=lambda r: r["event_timestamp"])
ptr = 0
latest = None
for idx, ent_row in ent_list:
    label_ts = ent_row["event_timestamp"]
    while ptr < len(feats) and feats[ptr]["event_timestamp"] <= label_ts:
        latest = feats[ptr]
        ptr += 1
    chosen = latest
    # then apply the TTL check on chosen

The merge is the same idea behind a merge join in a database, just specialized so it keeps the latest match instead of every match. On the bundled benchmark it runs over 300 times faster than the nested loop on 50,000 feature rows, and here is the part I care about more: I never have to trust that it is correct. The test suite runs both implementations on random data every single run and asserts they return identical results. If the fast path ever disagrees with the brute-force path, CI goes red.

rows (feat x label)     keys    fast (s)    naive (s)   speedup
5000 x 1000             50      0.0034      0.1582      46.1
20000 x 4000            200     0.0197      2.6587      134.9
50000 x 10000           500     0.0470      15.7845     336.0

Proving the leak instead of asserting it

I did not want a README that says "naive joins leak, trust me." I wanted the repo to show it. So there is a third join in the code, naive_last_value_join, which does exactly the wrong thing on purpose: it grabs each entity's single most recent feature value and ignores the label timestamp entirely. That is the shortcut people actually reach for.

The demo builds the training set both ways and counts how many rows differ. On the example dataset every label row comes out different, and each difference is a value the naive join pulled from that customer's future. The benchmark gate in CI fails the build if the naive join ever stops leaking, which means the demonstration cannot quietly rot into something that no longer proves the point. A gate that can never reject anything is just decoration, so this one actively checks that the wrong path is detectably wrong.

The live demo has a little interactive timeline where you drag the label time t and watch the as-of join pick the right event while the naive join keeps clutching the last one, even when it is sitting in the future. Watching the two numbers diverge made the whole concept concrete for me in a way no paragraph did.

The boring details that actually matter

A few things bit me, and they are the kind of thing the production feature stores quietly handle:

Timestamps. Everything normalizes to timezone-aware UTC on the way in, and gets stored as epoch microseconds as an integer. That means sqlite orders them exactly and the TTL math is plain integer arithmetic, no float drift, no naive-versus-aware datetime explosions halfway through a join.

Ties. When two feature rows share the exact same timestamp at or before t, which one wins? You have to pick a rule and stick to it, because the fast path and the slow path must agree. Asof keeps the last row in source order, deterministically, in both implementations. I found this the hard way when the benchmark mismatched on dense data: the stable sort kept ties in source order, the brute-force loop kept the first one it saw, and they disagreed. Changing one comparison from greater-than to greater-than-or-equal fixed it.

Materialize must not regress. Sweeping offline rows into the online store is latest-wins, and the upsert refuses to move an entity backward in time. So if you re-run an old window after a newer one, it does not clobber the fresh value. That makes materialize idempotent and safe to retry, which is the only way a scheduled job is allowed to behave.

What I would build next

This is a learning project, not a Feast replacement, and the gaps are deliberate. On-demand transforms, where a feature is computed at request time from other features, would be the next thing. Push sources for streaming updates. A proper feature server with auth instead of the little stdlib http.server inspector I ship now. Maybe a Parquet-backed offline store instead of sqlite once the history gets big.

But the core is done, and it does the one hard thing correctly. If you have ever nodded along to "feature store" without being totally sure what the difficult part was, the difficult part is this: never, not by one microsecond, let a feature from the future touch a label from the past. Everything else is plumbing.

The code is on GitHub at https://github.com/hajirufai/asof, MIT licensed, zero dependencies, with the interactive system map at https://hajirufai.github.io/asof/. It runs on nothing but the standard library, so you can clone it and poke at the join in about thirty seconds.

I Built a Bloom Filter from Scratch in Pure Python and Finally Understood How Databases Skip Reading Data

Haji Rufai — Tue, 09 Jun 2026 06:31:45 +0000

A few years into building data pipelines, I kept running into the same trick in systems I depended on every day.

Cassandra used it to avoid reading SSTables that did not contain my key. Bigtable used it for the same reason. Spark used it to skip whole partitions during joins. My CDN used it to decide whether a URL had ever been cached. Postgres extensions used it to prune rows before touching disk.

Same idea, five different systems. A tiny structure that answers one question fast: "have I probably seen this before?"

It is called a Bloom filter. And the day I built one from scratch was the day I stopped treating it like magic.

This is the toy version I wrote. No libraries, no bitarray, no mmh3. Just pure Python and the math underneath. By the end you will know exactly why a structure that fits in a few kilobytes can replace a lookup table that would cost you gigabytes.

The problem it actually solves

Imagine you run an LSM-tree storage engine. Writes go into many small sorted files on disk. To read one key, you might have to check ten files. Most of them do not contain the key. Each miss is a wasted disk read.

A disk read is slow. A memory check is fast. So before touching a file, you ask a small in-memory structure: "could this key be in here?"

If it says no, you skip the file entirely. If it says maybe, you read the file and check for real.

The key word is probably. A Bloom filter never says "yes". It says "no" with certainty, or "maybe" with a small chance of being wrong. It can produce false positives. It can never produce a false negative.

That asymmetry is the whole point. A false positive costs you one unnecessary read. A false negative would cost you correctness. Bloom filters trade a little wasted work for a hard guarantee.

The core idea in one paragraph

You keep a big array of bits, all starting at zero. To add an item, you run it through several hash functions. Each hash points at one position in the bit array. You set those bits to one.

To check an item, you hash it the same way and look at those same positions. If every one of them is a one, the item is probably present. If even one of them is a zero, the item is definitely absent, because adding it would have set that bit.

That is it. No stored keys, no buckets, no resizing. Just bits flipping on.

Building it: the bit array

Python has no native bit array, so I packed bits into a bytearray. Each byte holds eight bits. To touch bit i, I find its byte with i // 8 and its position inside that byte with i % 8.

class BitArray:
    def __init__(self, size_in_bits):
        self.size = size_in_bits
        self.data = bytearray((size_in_bits + 7) // 8)

    def set(self, i):
        self.data[i // 8] |= (1 << (i % 8))

    def get(self, i):
        return (self.data[i // 8] >> (i % 8)) & 1

Setting a bit uses a bitwise OR with a mask. Reading shifts the byte right and masks off everything but the lowest bit. This is the same packing trick databases use to keep the filter small enough to live in memory.

Building it: the hash functions

A Bloom filter needs several independent hash functions. Writing several good hashes by hand is painful, so I used a standard shortcut called double hashing. You compute two base hashes once, then combine them to fake as many as you need.

import hashlib

def _two_hashes(item):
    item = item.encode() if isinstance(item, str) else item
    h1 = int.from_bytes(hashlib.sha256(item).digest()[:8], "big")
    h2 = int.from_bytes(hashlib.md5(item).digest()[:8], "big")
    return h1, h2

def _hash_positions(item, num_hashes, size):
    h1, h2 = _two_hashes(item)
    for i in range(num_hashes):
        yield (h1 + i * h2) % size

The formula h1 + i * h2 gives a different position for each i. It behaves close enough to independent hashes for real use, and Cassandra ships essentially this approach.

Building it: the filter

Now the structure ties together. Add sets bits. Check reads bits and bails out the moment it sees a zero.

class BloomFilter:
    def __init__(self, size_in_bits, num_hashes):
        self.bits = BitArray(size_in_bits)
        self.num_hashes = num_hashes
        self.count = 0

    def add(self, item):
        for pos in _hash_positions(item, self.num_hashes, self.bits.size):
            self.bits.set(pos)
        self.count += 1

    def __contains__(self, item):
        for pos in _hash_positions(item, self.num_hashes, self.bits.size):
            if not self.bits.get(pos):
                return False
        return True

The early return False is where the certainty comes from. One zero bit means the item was never added. No exceptions.

bf = BloomFilter(size_in_bits=1000, num_hashes=4)
bf.add("user_42")
bf.add("user_99")

print("user_42" in bf)   # True
print("user_99" in bf)   # True
print("user_500" in bf)  # False, almost certainly

The part everyone gets wrong: sizing it

Here is where most tutorials wave their hands. The size and number of hashes are not guesses. They come from two numbers you choose: how many items you expect, and how many false positives you can tolerate.

The formulas are short. Given n expected items and a target false positive rate p:

import math

def optimal_size(n, p):
    # number of bits
    m = -(n * math.log(p)) / (math.log(2) ** 2)
    return int(math.ceil(m))

def optimal_hashes(m, n):
    # number of hash functions
    k = (m / n) * math.log(2)
    return max(1, int(round(k)))

Say you expect one million keys and accept a one percent false positive rate.

n = 1_000_000
p = 0.01
m = optimal_size(n, p)        # about 9,585,059 bits
k = optimal_hashes(m, n)      # 7 hash functions

print(f"{m} bits = {m / 8 / 1024 / 1024:.2f} MB")
print(f"{k} hash functions")

That is roughly 1.2 megabytes to track a million keys with one percent error. A real hash set of a million strings would cost you tens of megabytes. That gap is why storage engines reach for this instead of a set.

Push the error rate down to 0.1 percent and the filter grows to about 1.8 megabytes. Cheaper accuracy is mostly a matter of spending a few more bits.

Proving the false positive rate is real

Theory is nice. I wanted to see the one percent with my own eyes, so I built a filter, added known items, then queried items I never added and counted how often it lied.

def make_optimal(n, p):
    m = optimal_size(n, p)
    k = optimal_hashes(m, n)
    return BloomFilter(m, k)

bf = make_optimal(10_000, 0.01)

for i in range(10_000):
    bf.add(f"present_{i}")

false_positives = 0
trials = 100_000
for i in range(trials):
    if f"absent_{i}" in bf:
        false_positives += 1

print(f"measured rate: {false_positives / trials:.4f}")
# measured rate: roughly 0.0102

The measured rate lands right on top of the one percent target. The math is not a vibe. It is a contract the structure actually keeps.

What you cannot do, and why that is fine

You cannot delete from a plain Bloom filter. Clearing the bits for one item could clear bits another item depends on, and that would create a false negative, the one thing the structure promises never to happen. If you need deletes, you move to a counting Bloom filter that stores small counters instead of single bits.

You also cannot list what is inside. The filter holds no keys, only the shadow they cast on the bit array. That is exactly why it is so small. It remembers that you saw something without remembering what.

Why this changed how I read systems

After building it, the trick stopped being a black box and started being a tool I could reach for. Dedup a stream of billions of events without holding every ID in memory. Skip remote lookups for keys that were obviously never written. Prune join partitions before they ever hit the network.

Every one of those is the same one million keys in 1.2 megabytes idea, wearing a different costume.

The systems we depend on are not magic. They are a handful of sharp ideas, implemented carefully and reused everywhere. The fastest way to stop being intimidated by them is to build the toy version yourself.

Pick the structure under the system you lean on most. Write it from scratch this week. You will read your own infrastructure differently afterward.

If you build a Bloom filter after reading this, reply and tell me what false positive rate you measured. I read every comment, and I am always curious whether the math holds on someone else's machine.

Follow me for more "build the toy version of what you depend on" deep dives. Next up I am tearing into how databases keep millions of keys sorted on disk without ever loading them all into memory.

I Built a Write-Ahead Log in Pure Python and Finally Understood How Databases Survive Crashes

Haji Rufai — Mon, 08 Jun 2026 06:31:52 +0000

A few years ago I shipped a nightly batch job that wrote aggregated metrics to a file. It worked for months. Then one night the box ran out of disk halfway through a write, the process died, and the next morning I opened a half-written file full of garbage bytes. The downstream dashboard happily loaded it and showed numbers that were quietly wrong for two days before anyone noticed.

That was the day I actually understood why every serious database on the planet writes to a log before it touches the real data. Postgres does it. SQLite does it. Kafka is basically a log with a fan club. The pattern has a boring name: the Write-Ahead Log, or WAL.

So I did what I always do when I want to stop being scared of something: I built the toy version. No database, no C extensions, no libraries. Just Python and a file. Here is what I learned.

The one rule

The Write-Ahead Log has exactly one rule, and the whole thing falls out of it:

Before you change the real data, write what you are about to do to an append-only log, and make sure that log is safely on disk.

That is it. "Safely on disk" is doing a lot of work in that sentence, and we will get to it. But the core idea is: the log is the source of truth. The actual data file is just a cache of the log that you can always rebuild. If the machine dies, you replay the log and you are back exactly where you were.

The magic comes from two properties:

Appends are simpler than in-place edits. You never go back and modify an old record, so a crash can only ever leave a torn record at the very end of the file. Everything before it is intact.
You can detect the torn record with a checksum and just throw it away.

Let me show you what I mean with code.

A record format that survives a crash

The first thing I needed was a way to write records so I could tell a complete record from a half-written one. Here is the format I landed on for each entry:

[ 4 bytes: length ][ 4 bytes: crc32 ][ N bytes: payload ]

Length first so the reader knows how many bytes to grab. A CRC32 checksum so the reader can verify the payload was fully and correctly written. Then the payload itself.

import struct, zlib, os, json

HEADER = struct.Struct(">II")  # length, crc32, big-endian

def encode_record(payload: bytes) -> bytes:
    crc = zlib.crc32(payload) & 0xFFFFFFFF
    return HEADER.pack(len(payload), crc) + payload

Writing a record is then just: encode it, append it, and flush it to disk.

class WAL:
    def __init__(self, path):
        self.path = path
        # append + binary, create if missing
        self.f = open(path, "ab")

    def append(self, payload: bytes):
        self.f.write(encode_record(payload))
        self.f.flush()          # push out of Python's buffer
        os.fsync(self.f.fileno())  # push out of the OS buffer onto the disk
        return self

fsync is the whole point

That os.fsync line is the single most important line in this entire article, and it is the one almost everyone forgets.

When you call f.write(), the bytes go into Python's buffer. When you call f.flush(), they go into the operating system's buffer. Neither of those is the disk. The OS is allowed to sit on those bytes for a while before actually writing them to physical storage, because writing to disk is slow and batching is fast.

That is great for performance and catastrophic for durability. If the power cuts after flush() but before the OS got around to the real write, your data is gone, and you had no idea it was still in flight. os.fsync() is you telling the kernel: "no, really, I will wait right here until this is physically on the disk."

This is the trade every database makes. fsync on every record is the safe, slow setting. Batching many records per fsync is the fast, slightly-riskier setting. Postgres exposes this as synchronous_commit. Now you know what that knob actually does.

Reading it back, and surviving the torn tail

Now the fun part. Reading the log means walking the file record by record. If we hit a record whose length runs past the end of the file, or whose checksum does not match, we know that is the torn record from a crash, and we stop. Everything before it is good.

def read_all(path):
    records = []
    with open(path, "rb") as f:
        data = f.read()
    offset = 0
    while offset + HEADER.size <= len(data):
        length, crc = HEADER.unpack(data[offset:offset + HEADER.size])
        start = offset + HEADER.size
        end = start + length
        if end > len(data):
            break  # truncated payload, torn write
        payload = data[start:end]
        if (zlib.crc32(payload) & 0xFFFFFFFF) != crc:
            break  # corrupt record, stop here
        records.append(payload)
        offset = end
    return records

Notice how forgiving this is. It does not raise, it does not panic. It just reads as far as it safely can and returns clean records. That is the recovery story for free, and it only works because we never edit old records in place.

Turning the log into an actual key-value store

A log of raw bytes is not very useful on its own. Let me put a tiny key-value store on top of it so you can see how a real system uses this. Every set and delete becomes a log entry first, then updates an in-memory dictionary.

class KVStore:
    def __init__(self, path):
        self.wal = WAL(path)
        self.data = {}
        self._recover(path)

    def _recover(self, path):
        for raw in read_all(path):
            op = json.loads(raw)
            if op["t"] == "set":
                self.data[op["k"]] = op["v"]
            elif op["t"] == "del":
                self.data.pop(op["k"], None)

    def set(self, key, value):
        self.wal.append(json.dumps({"t": "set", "k": key, "v": value}).encode())
        self.data[key] = value

    def delete(self, key):
        self.wal.append(json.dumps({"t": "del", "k": key}).encode())
        self.data.pop(key, None)

    def get(self, key):
        return self.data.get(key)

The order inside set matters and it is deliberate: log first, memory second. If the process dies between the two lines, recovery still replays the log entry and the memory state ends up correct. If you did it the other way around, you could acknowledge a write that never made it to durable storage. That is the exact bug that bit me with my batch job, just dressed up nicely.

Here is the whole thing working, including a simulated crash:

# first run
kv = KVStore("store.wal")
kv.set("balance", 100)
kv.set("balance", 250)
kv.delete("temp")
# pretend the process is killed right here, no clean shutdown

# brand new process, same file
kv2 = KVStore("store.wal")
print(kv2.get("balance"))  # 250, replayed straight from the log

No clean shutdown, no flush-on-exit, no try/finally. We just open the file again and replay. The state is exactly what it should be because every change was durable the moment it was acknowledged.

Compaction: the log cannot grow forever

There is an obvious problem. If I set the same key a million times, the log has a million entries but the store has one value. Replaying a giant log on startup gets slow, and the file eats disk.

Real systems fix this with compaction (or checkpointing). The idea is simple: write the current state out as a fresh, minimal log, then atomically swap it in.

def compact(self, path):
    tmp = path + ".compact"
    with open(tmp, "wb") as f:
        for k, v in self.data.items():
            payload = json.dumps({"t": "set", "k": k, "v": v}).encode()
            f.write(encode_record(payload))
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)  # atomic rename, the safe swap

os.replace is the hero here. On every major OS, renaming a file over another is atomic: a reader either sees the entire old file or the entire new file, never a mix. So even if you crash mid-compaction, you either have the old complete log or the new complete log. You can never lose data in the gap. This is the same trick people use for "write to temp file, then rename" config saves, and it is one of the most useful patterns in all of systems programming.

What building this actually taught me

I have used Postgres for a decade. I have run Kafka in production. I thought I understood durability. I did not, not really, until I wrote those forty lines and watched a "crashed" process come back with perfect state.

A few things clicked that no blog post had ever made stick:

fsync is the line between "I saved it" and "I hope I saved it." Performance tuning in databases is almost always a negotiation about how often you are allowed to skip it.
Append-only is not a limitation, it is the feature. It is the reason a crash can only ever damage the tail, and the reason recovery is trivial instead of terrifying.
Atomic rename is a load-bearing wall under more of your stack than you realize.

The toy version is never the real thing. Production WALs deal with log rotation, segment files, group commit, partial-page writes, and a dozen other things I cheerfully ignored. But the next time I read about synchronous_commit, or watch Postgres replay its WAL on startup, or reason about what Kafka actually guarantees, I am not looking at a black box anymore. I am looking at the forty-line version with more careful engineers and more edge cases.

That is the whole reason I keep doing these builds. The fastest way I know to stop being intimidated by a piece of infrastructure is to build the smallest honest version of it with my own hands.

So that is my pitch to you: pick the one piece of your stack that feels like magic and build the toy version this weekend. A WAL. A cache. A rate limiter. You will read the real docs differently forever after.

If you build a tiny WAL of your own, reply and tell me what surprised you. And if you want more of these from-scratch builds, follow me here. I do one whenever I get curious enough about something to stop trusting it and start understanding it.

I Built a Columnar File Format in Pure Python — a tiny, readable Parquet

Haji Rufai — Mon, 08 Jun 2026 03:48:51 +0000

A while back I caught myself in an interview saying "yeah, I use Parquet a lot" — and then realized I couldn't actually explain why it's faster than a CSV beyond hand-waving about "columnar." That bugged me. So I did the thing that always fixes this for me: I rebuilt it from scratch.

The result is Columna, a columnar storage engine and file format written in pure Python. No pandas, no pyarrow, no numpy. Just struct and zlib from the standard library. It's about 3,000 lines, it has 81 tests, and on a 50,000-row dataset it ends up 91% smaller than the equivalent CSV while reading 98% fewer bytes to answer a filtered query.

Live demo + file inspector: https://hajirufai.github.io/columna/
Code: https://github.com/hajirufai/columna

Here's what I learned building it.

Why columnar actually wins

A CSV is row-major. The bytes for row 1 come first, then row 2, and so on. That layout is great for "give me this whole record," but it's terrible for analytics. If you want the average of one column, you still have to read every byte of every other column just to step over them. There's no way to skip.

Columnar storage flips this. All the values for amount are stored together, then all the values for region, and so on. Three things fall out of that one decision:

First, compression gets dramatically better, because values in a single column are similar to each other — same type, often similar magnitude, frequently repeated. A column of four country codes compresses far better than a row mixing an integer id, a float, and three strings.

Second, you can read just the columns a query needs. Want two columns out of seven? Read two column chunks, skip the rest. That's projection pushdown.

Third — and this is the one that surprised me how much it matters — if you store min/max stats per chunk, you can skip whole blocks of rows without decoding them. That's predicate pushdown, and it's where the real speedups live.

The file format

I went with a footer-last layout, same idea as Parquet. The writer streams row groups out as it goes and only writes the metadata at the very end:

MAGIC "CNA1"
Row Group 0
  Column Chunk "order_id"  -> pages
  Column Chunk "region"    -> pages
  ...
Row Group 1 ...
FOOTER (zlib-compressed JSON: schema, offsets, encodings, min/max)
FOOTER_LEN (uint32) + MAGIC

Why footer-last? Because when you're writing, you don't know a column chunk's compressed size or its min/max until you've seen all the rows. Putting metadata at the front would mean buffering the entire file in memory or seeking backwards. Writing it at the end lets you stream. The reader just seeks to the last few bytes, grabs the footer length and magic, inflates the footer JSON, and now it has a map of the entire file: every column chunk's byte offset, which encoding it used, and its min/max. All of that before reading a single value. That map is the whole trick.

A row group is just a horizontal slice of rows — say 5,000 at a time. Each row group holds one chunk per column, and each chunk is split into self-describing pages with a tiny header (magic byte, encoding id, codec id, value count, null count) and a CRC32 checksum. The checksum caught two of my own bugs during development, so I'm keeping it.

Five encodings, and letting the data decide

This was my favorite part. There are five encodings:

PLAIN — varint integers, IEEE-754 floats, length-prefixed strings. The fallback.
DICTIONARY — for low-cardinality columns. Store the distinct values once, replace the column with small integer indices, then bit-pack those indices.
RLE — run-length encoding for long stretches of the same value.
DELTA — for monotonic data like ids and timestamps. Store the first value, then zig-zag varints of each gap. A sequence like 100000, 100001, 100003 becomes one full integer plus single-byte deltas.
BITPACK — subtract the column minimum and pack the leftovers into the fewest bits that fit. Quantities of 1 through 12 need 4 bits each, not a full byte.

Now, how do you pick the right one? Early on I wrote a clever heuristic — "if cardinality is low use dictionary, if it looks sorted use delta" — and it kept choosing badly on edge cases. So I threw it out and did the dumb thing that turned out to be the right thing: the writer actually encodes every column with every applicable encoding, measures the bytes, and keeps whichever is smallest. Then it runs an optional DEFLATE pass and keeps that only if it shrinks further.

It's more work at write time, but it can never pick something worse than PLAIN, and the encoding id lives in the page header so the reader stays completely generic. I'd rather burn write CPU than ship a format that sometimes makes files bigger.

def choose_encoding(values, dtype):
    best_enc, best_buf = Encoding.PLAIN, encode_with(Encoding.PLAIN, values, dtype)
    for enc in applicable(dtype):
        buf = encode_with(enc, values, dtype)
        if len(buf) < len(best_buf):
            best_enc, best_buf = enc, buf
    return best_enc, best_buf

Nulls are handled with a definition bitmap — one bit per row marking which positions are present. Only non-null values go into the value stream. An all-null column costs about one bit per row and zero value bytes; a column with no nulls skips the bitmap entirely.

Predicate pushdown without lying about it

Here's the part I actually cared about getting right. When you run scan(where=col("amount") > 4500), the engine walks the row groups and, for each one, checks the footer stats. If a group's max amount is 4500 or less, it cannot possibly contain a matching row, so the engine skips it — no decode, no read.

def can_skip(self, stats):
    mn, mx = stats[self.column]["min"], stats[self.column]["max"]
    if mn is None or mx is None:
        return False              # unreliable stats -> never skip
    if self.op == ">":  return mx <= self.value
    if self.op == "<":  return mn >= self.value
    if self.op == "==": return self.value < mn or self.value > mx

The if mn is None line matters more than it looks. The cardinal sin of a pushdown engine is dropping a row that should have matched. If stats are missing, or polluted by a NaN, the only safe move is to refuse to skip and fall back to a full scan of that group. Slower, but correct. Correctness beats speed every single time here, and I wrote a test that hammers this: it generates random data, runs random predicates through the engine, and compares the result against a brute-force Python filter. If pushdown ever changed the result set, that test would scream. It doesn't.

Every scan returns a ScanResult that reports rows_scanned, row_groups_skipped, and bytes_read, so the speedup is something you can measure instead of something I claim.

The numbers

I built a benchmark script that pits Columna against plain CSV on the same 50,000-row orders dataset:

operation	CSV	Columna
File size	2,592 KB	225 KB (91% smaller)
Read 1 of 7 columns	2,592 KB	154 KB (94% less)
Filter `amount > 4500`	2,592 KB	47 KB (98% less)

For that last filter, Columna touched only 3 of 10 row groups — it skipped 7 entirely using nothing but the footer min/max stats. A CSV has no choice but to read the whole file, every time, for every query.

What I'd tell my past self

Two things stuck with me. One: the "try everything and measure" approach beat my clever heuristic every time, and it was less code. Two: a file format is mostly an exercise in being honest about offsets and lengths. Once the footer correctly describes where every byte lives, projection and predicate pushdown are almost free — they're just decisions about which byte ranges to not read.

It's a teaching-grade engine, not a production store — single file, single thread, no nested types, no schema evolution. But every byte on disk is explained by readable Python, and I can finally answer that interview question properly.

If you want to poke at it, the live inspector renders a real .cna file's row groups and encodings in the browser: https://hajirufai.github.io/columna/report.html — and the source is at https://github.com/hajirufai/columna.

I built a mini columnar storage engine in pure Python — and finally understood why Parquet is so fast

Haji Rufai — Sun, 07 Jun 2026 06:32:14 +0000

Last month a "fast" reporting query at work was taking 40 seconds. The table had 90 columns. The query touched 3 of them. We were reading the entire row, every byte, just to throw 97% of it away.

That's the moment columnar storage finally clicked for me — not as a buzzword, but as a thing I could feel in a slow dashboard. So over a weekend I built a tiny columnar store in pure Python to actually understand what Parquet and Arrow are doing under the hood. No pandas, no pyarrow. Just struct, bytes, and a notebook full of scribbles.

Here's what I learned, with the code that taught me.

Rows vs columns: same data, different bytes

Say you have this data:

rows = [
    {"user_id": 1, "country": "KE", "amount": 12.5},
    {"user_id": 2, "country": "KE", "amount": 8.0},
    {"user_id": 3, "country": "NG", "amount": 19.9},
]

A row store writes it like this on disk:

[1,"KE",12.5][2,"KE",8.0][3,"NG",19.9]

To sum amount, you have to walk past user_id and country on every single row. The disk reads them whether you want them or not.

A column store flips it:

user_id:  [1, 2, 3]
country:  ["KE", "KE", "NG"]
amount:   [12.5, 8.0, 19.9]

Now SELECT sum(amount) reads exactly one contiguous block and ignores the rest. That single rearrangement is 80% of the magic. The other 20% is what you can do because values of the same type now sit next to each other.

Building the writer

Let me build a file format that stores each column separately. First, the layout. A column file will be: a small header, then one encoded block per column, then a footer that says where each column lives so we can seek straight to it.

import struct, json

MAGIC = b"COLLITE1"

def write_table(path, columns: dict):
    # columns = {"user_id": [1,2,3], "amount": [12.5,...], ...}
    offsets = {}
    with open(path, "wb") as f:
        f.write(MAGIC)
        for name, values in columns.items():
            offsets[name] = {"start": f.tell()}
            block = encode_column(values)
            f.write(struct.pack("<I", len(block)))
            f.write(block)
            offsets[name]["length"] = len(block) + 4
        # footer: JSON map of column -> offset, then its length
        footer = json.dumps(offsets).encode()
        f.write(footer)
        f.write(struct.pack("<I", len(footer)))

The footer-at-the-end trick is exactly what Parquet does. You write data first (streaming, you don't need to know total size up front), then write metadata, then write the metadata length as the very last 4 bytes. A reader opens the file, jumps to the end, reads those 4 bytes, then reads the footer, and now knows where every column starts — without scanning the file.

Encoding: where columns earn their keep

Here's the part that made me appreciate the format. Because a column holds one type and often repeats values, you can compress it cheaply before any general-purpose compressor touches it. I implemented three encodings and let each column pick the smallest.

Dictionary encoding (for low-cardinality strings)

country has 3 rows but only 2 distinct values. Storing the strings repeatedly is wasteful. Instead, build a dictionary and store integer codes:

def encode_dict(values):
    seen = {}
    codes = []
    for v in values:
        if v not in seen:
            seen[v] = len(seen)
        codes.append(seen[v])
    dictionary = list(seen.keys())     # ["KE", "NG"]
    # codes -> [0, 0, 1]
    return dictionary, codes

Now "KE","KE","NG" becomes [0,0,1] plus a 2-entry dictionary. On a column with millions of rows and 50 distinct countries, you go from storing long strings to storing tiny ints. This is also why columnar engines can filter country = 'KE' so fast — they resolve 'KE' to code 0 once, then scan a tight array of integers.

Run-length encoding (for sorted or repetitive data)

If the data is sorted or clumpy, store runs instead of values:

def encode_rle(values):
    runs = []
    for v in values:
        if runs and runs[-1][0] == v:
            runs[-1][1] += 1
        else:
            runs.append([v, 1])
    return runs   # [["KE",2],["NG",1]]

A column that's KE for a million rows becomes a single ["KE", 1_000_000] pair. Time-partitioned data (everything from the same day) compresses absurdly well this way.

Delta encoding (for monotonic integers)

user_id climbs: 1001, 1002, 1003... Storing the gaps instead of the absolute values keeps the numbers tiny, which then bit-packs nicely:

def encode_delta(values):
    out, prev = [], 0
    for v in values:
        out.append(v - prev)
        prev = v
    return out   # 1001,1002,1003 -> 1001,1,1,1

Now the smart part — pick the winner per column:

def encode_column(values):
    candidates = []
    # try each strategy, measure serialized size, keep the smallest
    candidates.append(("raw", serialize_raw(values)))
    if all(isinstance(v, int) for v in values):
        candidates.append(("delta", serialize_delta(encode_delta(values))))
    if len(set(values)) < len(values) * 0.5:
        candidates.append(("dict", serialize_dict(*encode_dict(values))))
    candidates.append(("rle", serialize_rle(encode_rle(values))))

    tag, payload = min(candidates, key=lambda c: len(c[1]))
    return tag.encode().ljust(4, b" ") + payload

Each column independently chooses its best encoding. That's why a real Parquet file can have one column dictionary-encoded, the next RLE, the next plain — all in the same file. There is no single right answer; the data decides.

Reading: only pay for what you select

The payoff. To answer SELECT amount FROM t, the reader touches only the amount block:

def read_column(path, name):
    with open(path, "rb") as f:
        f.seek(-4, 2)                       # last 4 bytes
        flen = struct.unpack("<I", f.read(4))[0]
        f.seek(-(4 + flen), 2)
        offsets = json.loads(f.read(flen))  # column -> {start, length}

        info = offsets[name]
        f.seek(info["start"])
        size = struct.unpack("<I", f.read(4))[0]
        block = f.read(size)
        return decode_column(block)         # dispatch on the 4-byte tag

Two seeks, one read of one block. The other 89 columns never leave the disk. This is "column pruning," and it's the single biggest reason analytical queries on columnar files fly.

The other free win: predicate pushdown

Once data is encoded per column, you can store cheap per-block statistics — min and max — in the footer. Then a query like WHERE amount > 1000 can skip entire blocks without decoding them:

def block_can_match(stats, op, value):
    if op == ">":
        return stats["max"] > value   # skip block if max <= value
    if op == "<":
        return stats["min"] < value
    return True

If a block's max amount is 50 and you're asking for > 1000, you skip it entirely. On a year of partitioned data, a date filter ends up reading a handful of blocks instead of all of them. This is the same idea behind Parquet row-group statistics and the reason WHERE clauses on sorted columns feel almost free.

What I actually took away

A few things stuck with me after building this:

Columnar isn't one trick, it's three stacked wins. Layout (read fewer columns), encoding (same-type values compress hard), and statistics (skip blocks). Each multiplies the others.
The footer-at-the-end pattern is everywhere — Parquet, ORC, even some log formats. Write data streaming, write metadata last, write metadata length as the final bytes. Random access without scanning.
Encoding choice is per-column and data-dependent. Don't reach for a global "best" codec. Let each column measure and pick.
It explains the gotchas, too. Columnar is brutal for SELECT * row lookups and single-row updates — exactly why you don't run your transactional app on Parquet. Knowing why helps you pick the right store instead of cargo-culting.

I'm not suggesting anyone ship a homemade format — use Parquet, use Arrow, they're battle-tested and handle a hundred edge cases I skipped (nested types, nulls, encryption, compression codecs). But building the 200-line toy version turned columnar storage from a word I nodded along to into something I can reason about when a query is slow.

The 40-second report? Once I understood column pruning, the fix was obvious: stop SELECT *, store it columnar, partition by date. It now runs in under a second.

If you build data systems, I'd push you to build the toy version of whatever you depend on — a tiny query engine, a mini WAL, a baby columnar store. A weekend of "this is dumb and slow but it works" teaches more than a month of reading docs.

I write one of these build-from-scratch breakdowns regularly — caches, API gateways, secrets vaults, data-contract validators, and now this. Follow me here on dev.to if you want the next one, and tell me in the comments: what's the system you rely on every day but have never built yourself? That's probably your next weekend project.

I built a data-contract validator in pure Python (no pandas, no PyYAML) and it caught a 30% revenue ghost

Haji Rufai — Sat, 06 Jun 2026 03:51:24 +0000

A few months ago I spent the better part of a day chasing a bug that turned out not to be a bug at all. A downstream dashboard showed revenue had jumped 30% overnight. No deploys, no schema changes, nothing in the logs. After far too long I found it: an upstream system had started sending a total column that no longer equaled subtotal + tax. The pipeline didn't crash. The data just lied, quietly, and everything downstream believed it.

That's the thing about data bugs. They rarely throw exceptions. A status field grows a new typo'd value. A join key starts producing orphans. A nullable column that was "never actually null in practice" suddenly is. None of it crashes anything — it just rots the numbers people make decisions on.

So I built DataPact: a small framework for writing down what your data is supposed to look like, and then enforcing it. It's a data quality and data-contract validation tool, and the whole thing runs on the Python standard library. No pandas, no PyYAML, no network calls.

Live demo report: https://hajirufai.github.io/datapact/report.html
Landing page: https://hajirufai.github.io/datapact/
Source: https://github.com/hajirufai/datapact

The idea: contracts, not assertions scattered everywhere

Most teams already validate data — but it's usually a pile of ad-hoc assert df["x"].notna().all() lines buried in notebooks and DAGs. Nobody can answer "what are the rules for the orders table?" without grepping three repos.

A data contract flips that. You write the rules down in one declarative document — column types, null rules, ranges, allowed sets, regexes, cross-column math, referential integrity — version it in git, and let producers and consumers share it. DataPact then validates any batch against that contract and tells you, precisely, what broke.

Here's a contract in DataPact's YAML-lite format:

name: orders
version: 1.0
strictness: lenient
columns:
  - name: order_id
    type: int
    nullable: false
    checks:
      - kind: column_values_unique
        severity: error
  - name: status
    type: str
    checks:
      - kind: column_values_in_set
        kwargs: { values: [new, paid, shipped, refunded] }
expectations:
  - kind: multicolumn_sum_to_equal
    kwargs: { columns: [subtotal, tax], total_column: total, tolerance: 0.01 }

That last expectation is the exact rule that would have caught my 30% revenue ghost. subtotal + tax must equal total, within a cent.

"Zero dependencies" wasn't a vanity thing

I want to be honest about why this is stdlib-only, because it sounds like a flex and it mostly isn't. Two real reasons:

First, a lot of data platforms are locked down. You can't always pip install half of PyPI on the box where the pipeline runs. A validation tool that drops in with nothing but Python is genuinely easier to adopt than one that drags pandas + pyarrow + a YAML parser behind it.

Second, I wanted to actually understand the problem instead of gluing libraries together. Writing my own YAML reader and type-inference ladder taught me more about the messy reality of "what type is this column" than any wrapper would have.

The downside is I had to write a YAML parser. Which brings me to the most annoying bug of the whole project.

The escape-sequence bug that broke every email

DataPact ships its own tiny YAML reader — a strict subset: maps, lists, scalars, comments, quotes, flow lists. No arbitrary-object deserialization, which is a nice security property for free.

My email validation regex in the contract looked like this:

checks:
  - kind: column_values_match_regex
    kwargs:
      pattern: "^[^@ ]+@[^@ ]+\\.[^@ ]+$"

When I ran it, every single email failed, including obviously valid ones. 14 out of 14, 100%. My first naive parser just stripped the surrounding quotes and handed back the raw string — so \\. stayed as a literal backslash-backslash-dot. In the compiled regex that means "a literal backslash followed by any character," and no email on earth has a backslash in it.

The fix was to make the parser do what real YAML does: process escape sequences inside double-quoted strings, while leaving single-quoted strings literal.

_ESCAPES = {"n": "\n", "t": "\t", "r": "\r", '"': '"', "\\": "\\", "/": "/", "0": "\0"}

def _unescape_double(s: str) -> str:
    out, i = [], 0
    while i < len(s):
        ch = s[i]
        if ch == "\\" and i + 1 < len(s):
            out.append(_ESCAPES.get(s[i + 1], "\\" + s[i + 1]))
            i += 2
        else:
            out.append(ch)
            i += 1
    return "".join(out)

After that, the regex compiled to \. and the dirty email got flagged on its own (not-an-email has no @, so it correctly fails) while real addresses passed. The lesson, for the hundredth time in my career: escaping rules are never as simple as you hope, and the bug always shows up as "100% of things fail" rather than something subtle.

Type inference, and why "1" is not a boolean

The other rabbit hole was type inference. DataPact reads CSVs where everything is a string, so it has to guess: is "42" an int, is "4.5" a float, is "2026-01-04" a date?

I built an inference ladder — try bool, then int, then float, then datetime, then date, then fall back to string. And immediately got bitten. My parse_bool accepted "1" and "0" as True/False (reasonable when you're explicitly coercing). But during inference, that meant a column full of 1s and 0s got classified as boolean — and then my stdev check refused to run on it because "this isn't a numeric column."

The fix was to make inference conservative. Only unambiguous words — true, false, yes, no — infer as boolean. "1", "0", "t", "f" are far more likely to be integers or category codes, so they stay numeric:

if isinstance(value, str):
    if value.strip().lower() in ("true", "false", "yes", "no"):
        return "bool"
    if parse_int(value) is not None:
        return "int"
    if parse_float(value) is not None:
        return "float"
    ...

parse_bool is still lenient when you explicitly tell DataPact a column is boolean — but it no longer guesses bool from a digit. Small change, but it's the kind of default that quietly saves users from a confusing afternoon.

The architecture

The whole thing is a handful of small, pure pieces:

flowchart LR
    A[CSV / JSON / JSONL<br/>SQLite / records] --> B[Dataset<br/>+ type inference]
    C[Contract<br/>YAML / JSON / builder] --> D[Validation Engine]
    B --> D
    D --> E[ValidationReport]
    E --> F[CLI exit code]
    E --> G[HTML report]
    E --> H[guard / raise]
    B --> P[Profiler] --> C

Sources normalize CSV, JSON, JSONL, SQLite and plain lists-of-dicts into one Dataset view.
Expectations are a registry of pure functions — one per check kind. There are 23 of them across column-level (not_null, unique, between, in_set, match_regex, mean_between...), table-level (row_count_between, compound_columns_unique...) and cross-column (a > b, sum_to_equal, referential integrity).
The engine runs every expectation, applies strictness rules for unexpected columns, and builds a structured ValidationReport.

Every column-level check supports a mostly= tolerance, so you can say "this should be non-null in at least 99% of rows" instead of demanding perfection — real data is messy and a single bad row shouldn't always fail a 10-million-row batch.

Using it: three ways

As a library, validating against a contract file:

import datapact as dp

report = dp.validate("orders.csv", dp.load_contract("orders.yaml"))
print(report.success, report.passed, report.failed)
for r in report.results:
    if not r.success:
        print(r.expectation.label(), "→", r.message)

As a pipeline gate, with a decorator that raises before bad data escapes:

from datapact import guard, DataContractError

@guard(contract)
def load_orders():
    return fetch_rows_from_somewhere()

try:
    rows = load_orders()
except DataContractError as exc:
    alert(exc.report)   # the full report is attached to the exception

As a CI check, where a contract breach fails the build like a unit test:

datapact validate orders.csv --contract orders.yaml --fail-on error
echo $?   # 1 on breach

That --fail-on flag is the part I'm most happy with. It makes data quality a gate, not a dashboard nobody looks at. My GitHub Actions workflow actually runs two jobs: one proves the clean sample passes, and one proves the dirty sample fails — because a gate that never rejects anything is worse than no gate at all.

The report is the part people actually see

A validation result is only useful if a human can read it. So every run renders to a single self-contained HTML file — no external assets, no JavaScript framework — with failures sorted to the top and a full data profile (null rates, distinct counts, distributions) underneath.

I built the live demo report from a deliberately dirty orders file with ten injected problems: a duplicate order ID, a null email, a malformed email, an unknown status value, a bad date format, a negative subtotal, a quantity of 40, an invalid country code, and three rows where the totals don't add up. DataPact catches all ten — 41 expectations, 31 passed, 10 failed — and the report lays them out so you can see exactly which rows and values broke each rule.

Where it fits

DataPact is rule-based contract validation. It answers one question well: "does this batch obey the rules we agreed on?" That's deliberately different from statistical drift detection (is the distribution shifting?) or ETL orchestration (move the data around). It complements both — you'd run DataPact as the gate between an extract step and a load step, or in CI on your fixtures.

It's about 3,000 lines of pure Python with 141 tests, all stdlib unittest, no test dependencies either. If you've ever lost an afternoon to data that lied to you, give it a look — and if you find a rule it can't express yet, that's exactly the kind of issue I want to see.

I Built a Cache Engine from Scratch in Python — and O(1) LFU Eviction Is Sneakier Than LRU

Haji Rufai — Thu, 04 Jun 2026 03:40:25 +0000

When you call cache.get("user:123") and the cache is full, something has to go. That decision — which key gets thrown out to make room — is the whole game in caching. Get it right and your hit rate stays high. Get it wrong and you're constantly evicting the one thing you were about to ask for again.

I'd used functools.lru_cache and Redis for years without really knowing what happens in that moment. So I built CacheLite: an in-memory cache engine in pure Python, no dependencies, where I had to write every eviction policy by hand. LRU, LFU, FIFO, Random — plus TTL expiry, a reader-writer lock, snapshots, an HTTP API and a CLI. The repo is here: https://github.com/hajirufai/cachelite and there's a live walkthrough at https://hajirufai.github.io/cachelite/.

This post is the part I found most interesting: how eviction actually works, and why O(1) LFU is sneakier than O(1) LRU.

The thing nobody tells you about LRU

Everybody knows LRU means "evict the least recently used key." The interview answer is "use an OrderedDict." That's fine, but it hides the actual data structure trick, so I didn't let myself use it.

To do LRU in O(1) you need two things at once:

Instant lookup by key (so a hash map / dict)
Instant reordering when a key is touched (so a doubly-linked list)

The dict maps a key to its node. The linked list keeps everything in recency order, most-recent at the front. When you access a key, you unlink its node and stick it back at the front. When you need to evict, you grab the node at the very back. Both are pointer swaps — no shifting, no scanning.

The detail that makes the code clean is sentinel nodes. Instead of a real head and tail that you have to null-check, you keep two dummy nodes that always exist:

class LRUPolicy(EvictionPolicy):
    def __init__(self) -> None:
        self._head = _Node("\x00head")   # sentinel
        self._tail = _Node("\x00tail")   # sentinel
        self._head.next = self._tail
        self._tail.prev = self._head
        self._nodes: dict[str, _Node] = {}

    def _remove_node(self, node):
        node.prev.next = node.next
        node.next.prev = node.prev
        node.prev = node.next = None

    def _add_to_head(self, node):
        node.prev = self._head
        node.next = self._head.next
        self._head.next.prev = node
        self._head.next = node

Because head and tail always sit on the ends, inserting into an empty list and inserting into a full one are the same four lines. There's no "if this is the first node" branch. The first time I wrote this without sentinels I had three special cases and a bug in each. With sentinels it just works.

Access and eviction fall out of those two helpers:

def on_access(self, key):
    node = self._nodes.get(key)
    if node is None:
        return
    self._remove_node(node)
    self._add_to_head(node)   # touched → move to front

def evict(self):
    victim = self._tail.prev          # the node before tail = oldest
    if victim is self._head:
        raise CacheEmptyError("nothing to evict")
    self._remove_node(victim)
    del self._nodes[victim.key]
    return victim.key

I gave _Node a __slots__ so each node only carries key, prev, next and skips the per-instance __dict__. With a few hundred thousand nodes that adds up.

LFU is where it gets weird

LFU — least frequently used — sounds like a tiny variation on LRU. It is not. The naive version is O(n): to find the least-used key you scan every counter. To do it in O(1) you need three structures pulling together, and this is the part I actually had to draw on paper.

key -> frequency. How many times each key has been hit.
frequency -> ordered set of keys at that frequency. A bucket per frequency level.
min_freq. The lowest frequency currently in use, tracked as you go so eviction never scans.

The buckets are the trick. Every key with frequency 1 lives in bucket 1, every key with frequency 2 in bucket 2, and so on. To evict, you go straight to the bucket at min_freq and drop a key from it.

But there's a tie to break. If five keys all sit at frequency 1, which one goes? You want the one that's been sitting there longest — basically LRU within the frequency level. So each bucket is an OrderedDict, and you pop from the front:

def evict(self):
    if not self._key_freq:
        raise CacheEmptyError("nothing to evict")
    bucket = self._freq_keys[self._min_freq]
    key, _ = bucket.popitem(last=False)   # oldest key at lowest freq
    del self._key_freq[key]
    if not bucket:
        del self._freq_keys[self._min_freq]
        self._min_freq = min(self._freq_keys) if self._freq_keys else 0
    return key

The genuinely fiddly bit is keeping min_freq honest on access. When you touch a key, its frequency goes up by one, so it has to move from bucket f to bucket f+1:

def on_access(self, key):
    freq = self._key_freq.get(key)
    if freq is None:
        return
    bucket = self._freq_keys[freq]
    del bucket[key]
    if not bucket:                       # bucket emptied out
        del self._freq_keys[freq]
        if self._min_freq == freq:       # we just removed the min bucket
            self._min_freq = freq + 1
    new_freq = freq + 1
    self._key_freq[key] = new_freq
    self._freq_keys.setdefault(new_freq, OrderedDict())[key] = None

Here's the insight that makes the min_freq update O(1): when you promote a key out of the minimum bucket and that bucket becomes empty, the new minimum can only be freq + 1. It can't jump to some arbitrary value, because the key you just moved landed there. So you never have to search for the new minimum on the hot path. On a fresh insert, frequency is 1, so min_freq resets to 1 — also free. I only fall back to an actual min() scan in on_remove, which is the rare path.

That asymmetry — insert and access are O(1), only deletion might scan — is the kind of thing you don't appreciate until you've written it wrong once.

FIFO and Random, the honest ones

After LFU these two felt like a holiday.

FIFO evicts in insertion order, ignoring access entirely. A deque would do it, except keys can be deleted before they reach the front, so I tombstone: mark a key dead and skip it when it surfaces during eviction. That keeps inserts and evicts O(1) amortized without rebuilding the queue every time something gets removed.

Random eviction has one interesting wrinkle: how do you delete a random element from a collection in O(1)? You can't index into a dict. The trick is a parallel array of keys plus a key -> index map. To remove a key, you swap it with the last element of the array, pop the end (O(1)), and fix up the moved key's index. random.choice over the array gives you a uniform victim. Same swap-remove pattern game engines use for entity pools.

Expiry: lazy plus active, the way Redis does it

TTL turned out to need two mechanisms, not one, and Redis settled on the same combo for good reasons.

Lazy expiry checks the clock when you read a key. If it's past its expiry, you treat it as a miss and delete it on the spot. Cheap, but a key that's never read again sits in memory forever taking up space.

Active expiry is a background thread that periodically samples keys and clears the dead ones. That reclaims memory from keys nobody is touching. I used time.monotonic() for all of this rather than time.time() — monotonic never jumps backward when the system clock gets adjusted, so a daylight-saving change can't accidentally resurrect or kill your cache entries.

One small Python gotcha bit me here: signal handlers for graceful shutdown only work in the main thread. The background sweeper raised ValueError: signal only works in main thread until I wrapped the registration in a try/except and let non-main contexts skip it.

Making it thread-safe without a single global lock

A threading.Lock around everything would work and would also be slow, because reads — the common case for a cache — would block each other for no reason. Reads don't conflict with reads.

So I wrote a reader-writer lock: any number of readers can hold it at once, but a writer gets exclusive access. Then, to stop one hot key from serializing the whole cache, I added striped locking — hash the key, mod it by the number of stripes, and lock only that stripe. Two writes to different keys usually land on different stripes and never wait on each other. It's the same idea behind Java's old ConcurrentHashMap. The concurrency stress test fires a hundred threads at the cache at once and checks nothing corrupts.

What I'd reach for this in real life

Honestly, for most apps Redis is still the answer. But there's a real gap CacheLite fills: an in-process cache where you want a TTL and a specific eviction policy and stats and the ability to snapshot warm data to disk — without standing up a separate server. functools.lru_cache can't do any of those. I also now actually understand what my cache config is doing when I tune maxmemory-policy on a real Redis box, which is worth something on its own.

The whole thing is about 3,700 lines of Python, zero dependencies, 173 passing tests including the concurrency stress test, with CI on 3.11 and 3.12.

Code: https://github.com/hajirufai/cachelite
Live walkthrough: https://hajirufai.github.io/cachelite/

If you've only ever used a cache as a black box, try writing the LFU eviction path by hand. It's a couple of hours and you'll never look at a maxmemory-policy setting the same way again.

I built a task scheduler with cron parsing from scratch in Python — zero dependencies

Haji Rufai — Mon, 01 Jun 2026 12:53:57 +0000

Every cron library I looked at pulled in a pile of dependencies for what is, fundamentally, a string-to-schedule conversion. So I decided to write one from scratch — and then kept going until I had a complete task scheduler.

CronLite is ~5,300 lines of Python. Zero external packages. Everything from the cron parser down to the SQLite storage layer uses nothing beyond the standard library.

GitHub · Live Demo

What it does

Full POSIX cron parsing (wildcards, ranges, steps, lists, month/day names, presets)
Priority-queue scheduler with a fixed-size thread pool
Retry strategies: fixed, linear, exponential backoff
DAG job dependencies with cycle detection
SQLite-backed persistence (WAL mode for concurrent reads)
REST API (~15 endpoints) built on http.server
CLI for everything: add, remove, trigger, explain, history, stats

The interesting part: next_run()

The most satisfying function to write was next_run(). Given a cron expression and a starting datetime, find the next moment that matches all five fields.

The naive approach — scan forward minute by minute — works but is painfully slow for something like 0 0 1 1 * (only runs once a year).

Instead, I walk fields from most-significant to least-significant. When a field doesn't match, I advance to the next matching value and reset all less-significant fields. Carries propagate naturally: if no minute matches in the current hour, advance the hour; if no hour matches today, advance the day.

def next_run(self, after: datetime) -> datetime | None:
    candidate = _truncate_to_minute(after) + timedelta(minutes=1)
    deadline = after + timedelta(days=_MAX_SEARCH_YEARS * 366)

    while candidate <= deadline:
        # Month
        if not self.month.matches(candidate.month):
            candidate = self._advance_month(candidate)
            if candidate is None or candidate > deadline:
                return None
            continue

        # Day (POSIX union semantics for DOM + DOW)
        if not self._day_matches(candidate):
            candidate = candidate.replace(hour=0, minute=0) + timedelta(days=1)
            continue

        # Hour
        if not self.hour.matches(candidate.hour):
            next_hour = self._next_value(candidate.hour, self.hour.values)
            if next_hour is not None and next_hour > candidate.hour:
                candidate = candidate.replace(hour=next_hour, minute=0)
                if self.minute.matches(0):
                    return candidate
                next_min = self._next_value(0, self.minute.values)
                if next_min is not None:
                    return candidate.replace(minute=next_min)
            candidate = candidate.replace(hour=0, minute=0) + timedelta(days=1)
            continue

        # Minute
        if not self.minute.matches(candidate.minute):
            next_min = self._next_value(candidate.minute, self.minute.values)
            if next_min is not None and next_min > candidate.minute:
                return candidate.replace(minute=next_min)
            # No more minutes this hour — advance
            next_hour = self._next_value(candidate.hour + 1, self.hour.values)
            if next_hour is not None:
                candidate = candidate.replace(
                    hour=next_hour, minute=min(self.minute.values)
                )
                continue
            candidate = candidate.replace(hour=0, minute=0) + timedelta(days=1)
            continue

        return candidate

    return None

The POSIX spec has a subtle rule: if both day-of-month and day-of-week are restricted (not wildcards), a day matches if either field matches — it's a union, not an intersection. Most tutorials get this wrong.

Retry engine

When a job fails, the retry policy decides what happens next. I implemented three strategies, each computing the delay before the next attempt:

class RetryPolicy:
    def compute_delay(self, attempt: int) -> float:
        if self.strategy == RetryStrategy.FIXED:
            delay = self.base_delay
        elif self.strategy == RetryStrategy.LINEAR:
            delay = self.base_delay * attempt
        elif self.strategy == RetryStrategy.EXPONENTIAL:
            delay = self.base_delay * (2 ** (attempt - 1))
        return min(delay, self.max_delay)

For a base delay of 10 seconds and exponential backoff: 10s → 20s → 40s → 80s → 160s, capped at whatever max_delay is set to. The cap prevents a job from waiting hours between retries after a dozen failures.

DAG dependencies

Jobs can declare dependencies. Before firing a job, the engine checks that all upstream jobs have completed successfully:

class JobDAG:
    def add_dependency(self, job_id: str, depends_on: str) -> None:
        # Check for cycles using DFS 3-color algorithm
        self._edges[job_id].add(depends_on)
        if self._has_cycle():
            self._edges[job_id].discard(depends_on)
            raise CyclicDependencyError(f"Adding {depends_on} → {job_id} creates a cycle")

Cycle detection uses the standard three-color DFS: white (unvisited), gray (in current path), black (fully explored). If we reach a gray node, there's a cycle. Topological sort uses Kahn's algorithm — start from nodes with zero in-degree, peel them off one at a time.

Storage backends

Both MemoryStore and SQLiteStore implement the same Store ABC, so swapping between them is a one-line change. The SQLite backend uses WAL journal mode for concurrent reads and parameterized queries throughout:

class SQLiteStore(Store):
    def __init__(self, db_path: str):
        self._conn = sqlite3.connect(db_path)
        self._conn.execute("PRAGMA journal_mode=WAL")
        self._create_tables()

    def save_job(self, job: Job) -> None:
        self._conn.execute(
            "INSERT OR REPLACE INTO jobs VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            self._job_to_row(job),
        )
        self._conn.commit()

Running it

git clone https://github.com/hajirufai/cronlite.git
cd cronlite

# Explain a schedule
python -m cronlite explain "*/15 9-17 * * MON-FRI"
# → Every 15 minutes, between 09:00 and 17:59, on weekdays

# Add and trigger a job
python -m cronlite add backup "0 2 * * *" "pg_dump mydb > backup.sql"
python -m cronlite trigger backup

# Start the scheduler with API
python -m cronlite start --workers 4 --port 8080

Or use the HTTP API:

curl -X POST http://localhost:8080/api/v1/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "name": "etl-pipeline",
    "expression": "0 */6 * * *",
    "command": "python run_etl.py",
    "max_retries": 3,
    "retry_strategy": "exponential"
  }'

Numbers

22 source modules
219 passing tests
5 example scripts
0 external dependencies
Python 3.11+ only

The test suite covers parsing edge cases (February 30, leap years, step patterns), executor behavior (timeouts, zombie processes, output truncation), storage persistence, DAG cycle detection, retry delay calculations, and full end-to-end scheduling scenarios.

Things I learned

POSIX cron semantics are full of edge cases. Day-of-month + day-of-week union caught me off guard. February 30 expressions parse successfully but next_run() correctly returns None within the search window.

Signal handlers only work in the main thread. The scheduler's graceful shutdown handler needs a try/except around signal.signal() for when it's used as a library in a non-main thread.

Python's or with containers is subtle. empty_list or default returns default because empty containers are falsy. Had to use explicit is not None checks when distinguishing between "no value" and "empty value."

SQLite WAL mode matters. Without it, concurrent reads during a write transaction block. With WAL, readers and writers don't interfere with each other.

Source: github.com/hajirufai/cronlite

I Implemented AES-128 from Scratch and Built a Secrets Vault in Python

Haji Rufai — Mon, 01 Jun 2026 06:56:45 +0000

Most developers import cryptography or pycryptodome and call it a day. I wanted to understand what happens inside the black box — so I implemented AES-128 byte-by-byte from the NIST FIPS 197 specification and built a full secrets manager around it.

VaultLite is the result: a lightweight HashiCorp Vault alternative with zero external dependencies. Everything from the S-Box substitution to the HTTP API runs on Python's standard library.

Why AES from Scratch?

AES (Advanced Encryption Standard) is everywhere. Your browser uses it right now. AWS KMS, HashiCorp Vault, Signal — they all rely on AES. But what actually happens when you encrypt a block of data?

I spent a week with the NIST FIPS 197 spec open in one tab and Python open in the other. Here's what I learned.

The AES-128 Pipeline

AES operates on a 4×4 matrix of bytes called the "state." A 16-byte plaintext block goes through 10 rounds of four transformations:

Plaintext (16 bytes)
    ↓
[AddRoundKey] ← round key 0
    ↓
┌─────────────────┐
│ Round 1-9:      │
│  SubBytes       │  ← S-Box substitution (GF(2⁸) inversion)
│  ShiftRows      │  ← Row rotation
│  MixColumns     │  ← Column mixing (Galois Field math)
│  AddRoundKey    │  ← XOR with round key
└─────────────────┘
    ↓
[SubBytes → ShiftRows → AddRoundKey] ← round 10 (no MixColumns)
    ↓
Ciphertext (16 bytes)

The S-Box: Where the Security Lives

The S-Box is a 256-entry lookup table. Each byte in the state gets substituted with its corresponding S-Box entry. It's the only non-linear part of AES — which is what makes it hard to break.

The S-Box is computed by finding the multiplicative inverse in GF(2⁸), then applying an affine transformation. In my implementation:

def _build_sbox():
    sbox = [0] * 256
    sbox[0] = 0x63  # 0 has no inverse; spec maps it to 0x63

    for i in range(1, 256):
        inv = gf_inverse(i)  # Multiplicative inverse in GF(2^8)
        # Affine transformation over GF(2)
        b = inv
        result = 0
        for bit in range(8):
            val = ((b >> bit) & 1)
            val ^= ((b >> ((bit + 4) % 8)) & 1)
            val ^= ((b >> ((bit + 5) % 8)) & 1)
            val ^= ((b >> ((bit + 6) % 8)) & 1)
            val ^= ((b >> ((bit + 7) % 8)) & 1)
            result |= (val << bit)
        sbox[i] = result ^ 0x63

    return sbox

MixColumns: Galois Field Arithmetic

MixColumns is the transformation that had me staring at my screen the longest. Each column of the state matrix gets multiplied by a fixed polynomial in GF(2⁸).

The key operation is xtime — multiplication by x (i.e., by 2) in GF(2⁸):

def xtime(a):
    result = a << 1
    if result & 0x100:  # If bit 8 is set, reduce
        result ^= 0x11B  # x^8 + x^4 + x^3 + x + 1
    return result & 0xFF

The magic number 0x11B is the irreducible polynomial that defines AES's finite field. Without it, multiplication would overflow — with it, results always stay within one byte.

Key Expansion: 16 Bytes → 176 Bytes

The 16-byte encryption key expands into 11 round keys (176 bytes total). Each new word is derived from the previous word, with every fourth word going through an additional transformation using the S-Box and a round constant:

def key_expansion(key):
    words = [key[4*i:4*i+4] for i in range(4)]

    for i in range(4, 44):
        temp = list(words[i-1])
        if i % 4 == 0:
            temp = temp[1:] + temp[:1]      # RotWord
            temp = [SBOX[b] for b in temp]   # SubWord
            temp[0] ^= RCON[i//4]            # Round constant
        words.append(bytes(a ^ b for a, b in zip(words[i-4], temp)))

    return [words[4*r:4*r+4] for r in range(11)]

Beyond the Cipher: Building a Vault

AES gives you a way to encrypt 16 bytes. A secrets manager needs a lot more:

Envelope Encryption

Each secret gets its own randomly generated Data Encryption Key (DEK). The DEK encrypts the data. The master key (KEK) encrypts the DEK. This way, rotating the master key only means re-encrypting the DEKs, not every secret.

Master Key (KEK)
    │
    ▼
┌─────────┐
│ Wrap DEK │──→ Encrypted DEK (stored alongside ciphertext)
└─────────┘
    │
Random DEK
    │
    ▼
┌───────────────────┐
│ AES-CBC(DEK, data)│──→ Ciphertext + IV + HMAC
└───────────────────┘

Seal/Unseal Ceremony

The master key never touches disk. On startup, the vault is sealed — it can't decrypt anything. To unseal, you provide key shares that were generated at initialization via XOR-based secret splitting.

This means no single person (or compromised server) can access the secrets without all the shares.

Hash-Chained Audit Log

Every vault operation gets logged in an append-only audit trail. Each entry contains the SHA-256 hash of the previous entry — forming a chain like a simplified blockchain. Tamper with any entry and the chain verification fails.

entries = vault.audit_log(token)
# [write] secret/database/prod  → allow
# [read]  secret/api/stripe     → deny (no capability)

integrity = vault.verify_audit_chain(token)
# {"valid": True, "total_entries": 47}

The Result

VaultLite ended up at ~6,300 lines of Python with 258 passing tests:

Crypto layer: AES-128, CBC mode, PKCS7 padding, HMAC-SHA256, PBKDF2, envelope encryption
Auth: Token trees with cascading revocation, app-role for machines
Access control: Path-based policies with glob matching
Versioning: Full version history, soft-delete/undelete, permanent destroy
Leasing: TTL-based leases with renewal and cleanup
API: 25+ RESTful endpoints using only http.server
Storage: In-memory, file, and SQLite backends

Zero dependencies. Just Python 3.10+.

Try It

git clone https://github.com/hajirufai/vaultlite.git
cd vaultlite
python -m vaultlite demo

Or use it as a library:

from vaultlite.vault import Vault

vault = Vault()
result = vault.initialize(shares=3, threshold=3)
for share in result.unseal_keys:
    vault.unseal(share)

vault.write("secret/db", {"password": "s3cur3"}, result.root_token)
secret = vault.read("secret/db", result.root_token)

The full source is at github.com/hajirufai/vaultlite. The landing page is at hajirufai.github.io/vaultlite.

What I Learned

AES is elegant once you see the math. The S-Box, MixColumns, and key expansion all work together to achieve diffusion (small input changes cascade) and confusion (no simple relationship between key and ciphertext).
Envelope encryption is underrated. Separating the data key from the master key makes rotation and key management dramatically simpler.
Audit chains catch more than you'd think. Even in testing, the hash chain caught bugs where I accidentally mutated entries.
The stdlib is more capable than people think. http.server, hashlib, hmac, sqlite3 — Python ships with enough to build serious security infrastructure.

Building AES from scratch won't make your code more secure (use cryptography in production). But it will make you understand what "128-bit encryption" actually means — and that understanding makes every security decision you make afterward a little bit sharper.

I Built a Stream Processing Engine from Scratch in Python — Here's How Windowing Actually Works

Haji Rufai — Sun, 31 May 2026 12:59:26 +0000

Most stream processing tutorials teach you how to connect Kafka to Postgres. They don't explain what happens inside the engine — how records get assigned to windows, how watermarks handle late data, or what a checkpoint actually serializes.

I built StreamLite to answer those questions. It's a complete stream processing engine in Python — 24 modules, 281 tests, zero external dependencies. Everything from tumbling windows to keyed state to checkpoint coordination, implemented from scratch.

The API

Three lines for a word count:

from streamlite import StreamLite

result = (
    StreamLite.from_collection(["the quick brown fox", "the lazy dog"])
    .flat_map(str.split)
    .key_by(lambda w: w)
    .count()
)
# [("the", 2), ("quick", 1), ("brown", 1), ...]

But the interesting part isn't the API. It's what happens underneath.

How Windowing Actually Works

When people say "tumbling window" they usually mean "group by minute." But the actual implementation has subtle decisions that change behavior.

Here's the core of TumblingWindow.assign_windows():

class TumblingWindow(WindowAssigner):
    def __init__(self, size_ms, offset_ms=0):
        self.size_ms = size_ms
        self.offset_ms = offset_ms

    def assign_windows(self, timestamp):
        start = timestamp - (timestamp - self.offset_ms) % self.size_ms
        return [TimeWindow(start=start, end=start + self.size_ms)]

One line of math: timestamp - (timestamp - offset) % size. That's it. But this formula handles:

Alignment: all windows snap to the same grid regardless of when the first record arrives
Offset: shift windows by offset_ms to align with business hours (e.g., daily windows starting at 6 AM, not midnight)
Determinism: the same timestamp always maps to the same window, even across restarts

Sliding windows are trickier

A sliding window with size=10s and slide=5s means each record belongs to two windows. A record at timestamp 7000 belongs to [0, 10000) and [5000, 15000):

class SlidingWindow(WindowAssigner):
    def assign_windows(self, timestamp):
        windows = []
        last_start = timestamp - (timestamp - self.offset_ms) % self.slide_ms
        for start in range(last_start, last_start - self.size_ms, -self.slide_ms):
            if start + self.size_ms > timestamp >= start:
                windows.append(TimeWindow(start=start, end=start + self.size_ms))
        return windows

The key insight: you walk backward from the latest window start, adding windows until the record no longer falls within the window bounds.

Session windows merge

Session windows don't have a fixed size. They group records by activity gaps:

class SessionWindow(WindowAssigner):
    def merge_windows(self, windows):
        sorted_windows = sorted(windows, key=lambda w: w.start)
        merged = [sorted_windows[0]]
        for window in sorted_windows[1:]:
            if window.start <= merged[-1].end + self.gap_ms:
                merged[-1] = TimeWindow(merged[-1].start, max(merged[-1].end, window.end))
            else:
                merged.append(window)
        return merged

Each new record creates a point window. Then merge_windows combines overlapping windows (where the gap between them is less than gap_ms). This merge step is what makes session windows expensive — it's O(n log n) per merge pass.

The Watermark Problem

Out-of-order data is the fundamental challenge in stream processing. A record with timestamp t=5000 might arrive after t=8000. When do you close the [0, 10000) window?

Watermarks are the answer. A watermark at time W means: "I believe all records with timestamp ≤ W have arrived."

class BoundedOutOfOrderness(WatermarkStrategy):
    def __init__(self, max_delay_ms):
        self._max_delay = max_delay_ms
        self._max_timestamp = Watermark.MIN_WATERMARK

    def on_event(self, timestamp):
        if timestamp > self._max_timestamp:
            self._max_timestamp = timestamp

    def current_watermark(self):
        if self._max_timestamp == Watermark.MIN_WATERMARK:
            return Watermark.MIN_WATERMARK
        return self._max_timestamp - self._max_delay

With max_delay_ms=5000, after seeing timestamp 10000 the watermark advances to 5000. Any record with timestamp ≤ 5000 arriving now is considered "on time." This tolerance is what makes bounded out-of-orderness practical — you trade latency for completeness.

Keyed State: The Secret Sauce

Stateful processing is what separates stream engines from batch map-reduce. StreamLite has four state primitives:

class ValueState:
    def get(self): ...      # returns current value or None
    def update(self, v): ...  # set new value
    def clear(self): ...     # reset

class ListState:
    def add(self, v): ...   # append
    def get(self): ...      # returns list

class MapState:
    def get(self, key): ...
    def put(self, key, value): ...

class ReducingState:
    def add(self, v): ...   # accumulate with reduce function

These are scoped by namespace + key, managed by a StateBackend that handles snapshot/restore for checkpointing:

def running_average(value, ctx):
    total = ctx.get_value_state("total", default=0)
    count = ctx.get_value_state("count", default=0)
    t = (total.get() or 0) + value["temp"]
    c = (count.get() or 0) + 1
    total.update(t)
    count.update(c)
    return [{"sensor": value["sensor"], "avg": t / c}]

stream.key_by(lambda r: r["sensor"]).process(running_average)

Each key gets its own isolated state. That isolation is what enables parallel processing in real engines — different keys can be processed on different threads without coordination.

Checkpointing: Crash Recovery

StreamLite checkpoints by serializing the entire StateBackend to disk:

class CheckpointCoordinator:
    def trigger_checkpoint(self, watermarks=None):
        self._checkpoint_count += 1
        checkpoint_id = f"chk-{self._checkpoint_count:06d}"
        snapshot = self._state_backend.snapshot()
        self._storage.save(checkpoint_id, snapshot, metadata)
        return metadata

The snapshot() method walks every namespace → key → state object and serializes to nested dicts. On restore, it reconstructs the state objects from those dicts.

Real engines like Flink use aligned checkpoint barriers across parallel operators. StreamLite keeps it simple — single-threaded snapshot — but the concept is the same: freeze state, write to durable storage, resume.

Stream Joins

StreamLite supports three join patterns:

Hash join — group both sides by key, match:

join = StreamJoin(left, right, join_fn=lambda l, r: {**l, **r}, join_type=JoinType.INNER)

Interval join — match records within a time window:

join = IntervalJoin(clicks, purchases, lower_bound_ms=-1000, upper_bound_ms=5000, join_fn=correlate)

Window join — join within aligned time windows:

join = WindowJoin(left, right, window_assigner=TumblingWindow(size_ms=60000), join_fn=merge)

The Full Architecture

Sources → Operators → Windowing → State → Sinks
              ↕            ↕         ↕
         Watermarks    Triggers  Checkpoints
              ↕
          Metrics

24 modules in total: types, errors, source, sink, operators, stream, keyed, window, trigger, watermark, state, context, join, split, merge, serializer, partition, checkpoint, metrics, pipeline, executor, time_utils, utils, and the main __init__.

Numbers

24 modules, each focused on one concept
281 tests across 20 test files
~8,000 lines of Python
Zero dependencies — stdlib only
5 examples: word count, clickstream, sensors, ETL, split-rejoin

Try it

git clone https://github.com/hajirufai/streamlite.git
cd streamlite
python -m pytest tests/ -v
python examples/word_count.py

The entire codebase is designed to be read. If you've ever wondered how Flink or Kafka Streams work under the hood, start with window.py and watermark.py. Those two files contain the core ideas that make stream processing work.

GitHub · Live docs

I Built an API Gateway from Scratch in Python — Zero Dependencies

Haji Rufai — Sun, 31 May 2026 06:57:40 +0000

Every backend developer eventually encounters API gateways — NGINX, Kong, Envoy. They're powerful, but how do they actually work? I decided to find out by building one from scratch.

GateLite is a fully functional API gateway written in pure Python with zero external dependencies. It handles routing, load balancing, rate limiting, authentication, circuit breaking, caching, and more — all in ~7,000 lines using only the standard library.

GitHub Repository | Live Demo

The Request Pipeline

Every request flows through a carefully ordered pipeline:

Client → CORS → Middleware → Auth → Rate Limit → Route Match
  → Cache → Transform → Circuit Breaker → Load Balance
  → Proxy Forward → Transform → Cache Store → Client

Each stage can short-circuit the pipeline with an early response. A rate-limited request never reaches the upstream. A cached response skips the proxy entirely.

Core Architecture

HTTP Parser

Since we can't use flask or fastapi, we need our own HTTP parser. GateLite reads raw bytes from TCP sockets and parses HTTP/1.1 requests:

class HTTPParser:
    def parse_request(self, data: bytes) -> Request:
        lines = data.split(b"\r\n")
        method, path, version = lines[0].decode().split(" ", 2)

        headers = Headers()
        body_start = 0
        for i, line in enumerate(lines[1:], 1):
            if line == b"":
                body_start = i + 1
                break
            name, value = line.decode().split(": ", 1)
            headers.add(name, value)

        body = b"\r\n".join(lines[body_start:]) if body_start else None
        return Request(method=method, path=path, headers=headers, body=body)

Smart Routing

Routes support four match types with priority ordering:

# Exact match (highest priority)
Route(name="health", path="/health", upstream="internal")

# Parameterized paths
Route(name="user", path="/users/{id}", upstream="user-svc")

# Prefix matching
Route(name="api", path="/api/*", upstream="backend")

# Regex (lowest priority)
Route(name="versioned", path="/v[0-9]+/.*", upstream="backend")

The router compiles patterns once and sorts by priority. Parameterized routes extract path params into a dict:

match = router.match("GET", "/users/42")
# match = (Route("user"), {"id": "42"})

Load Balancing

GateLite implements five load balancing algorithms:

class RoundRobin:
    def select(self, upstreams):
        self._index = (self._index + 1) % len(upstreams)
        return upstreams[self._index]

class LeastConnections:
    def select(self, upstreams):
        return min(upstreams, key=lambda u: u.active_connections)

class IPHash:
    def select(self, upstreams, client_ip):
        idx = hash(client_ip) % len(upstreams)
        return upstreams[idx]

Weighted round-robin distributes traffic proportionally — a server with weight 3 gets 3x the traffic of weight 1.

Token Bucket Rate Limiting

The token bucket algorithm provides smooth rate limiting with burst handling:

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # tokens per second
        self.capacity = capacity  # max burst size
        self.tokens = capacity
        self.last_refill = time.monotonic()

    def allow(self) -> bool:
        self._refill()
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now

GateLite also implements fixed window and sliding window log algorithms.

Circuit Breaker

Circuit breakers prevent cascading failures. When an upstream fails repeatedly, the circuit "opens" and rejects requests immediately instead of waiting for timeouts:

class CircuitBreaker:
    # States: CLOSED (normal) → OPEN (rejecting) → HALF_OPEN (testing)

    def allow_request(self) -> bool:
        if self.state == State.CLOSED:
            return True
        if self.state == State.OPEN:
            if time.monotonic() - self.last_failure > self.recovery_timeout:
                self.state = State.HALF_OPEN
                return True  # Allow one test request
            return False
        # HALF_OPEN: allow limited requests
        return self._half_open_requests < self.max_requests

    def record_success(self):
        if self.state == State.HALF_OPEN:
            self._successes += 1
            if self._successes >= self.success_threshold:
                self.state = State.CLOSED  # Recovery!

    def record_failure(self):
        self._failures += 1
        if self._failures >= self.failure_threshold:
            self.state = State.OPEN

JWT Authentication (No Libraries!)

Implementing JWT from scratch means working with Base64 and HMAC directly:

class JWTAuth:
    def decode(self, token: str) -> tuple[dict, dict]:
        parts = token.split(".")
        header = json.loads(self._b64decode(parts[0]))
        payload = json.loads(self._b64decode(parts[1]))

        # Verify signature
        signing_input = f"{parts[0]}.{parts[1]}".encode()
        expected = hmac.new(
            self.secret, signing_input, hashlib.sha256
        ).digest()
        actual = self._b64decode(parts[2])

        if not hmac.compare_digest(expected, actual):
            raise ValueError("Invalid signature")

        return header, payload

What I Learned

1. HTTP parsing is deceptively tricky. Edge cases everywhere: missing Content-Length, malformed headers, partial reads. The parser needs to be robust without being slow.

2. Thread safety matters. Rate limiters, circuit breakers, and metrics all need thread-safe access. Python's threading.Lock is your friend, but you need to minimize critical sections.

3. The pipeline pattern is powerful. Each middleware stage is independent and composable. Adding a new feature means adding a new stage, not modifying existing ones.

4. Circuit breakers need careful thresholds. Too sensitive and you get false positives. Too lenient and you don't protect against real failures. The half-open state is crucial for automatic recovery.

5. Caching is a protocol. Cache-Control headers, ETag validation, conditional 304 responses — proper HTTP caching follows a detailed spec that's worth understanding.

The Numbers

Metric	Value
Source modules	24
Test modules	19
Total lines	~7,000
Passing tests	298
External deps	0

Try It

git clone https://github.com/hajirufai/gatelite.git
cd gatelite
python -m pytest tests/ -v  # Run all 298 tests
python examples/basic_proxy.py  # Start a proxy

The full source is on GitHub. Contributions and feedback welcome!

This is project #16 in my "Building from Scratch" series. Previous projects include a message broker, search engine, compiler, and distributed key-value store.