Forem: Nithin Bharadwaj

WebAssembly Is Turning Your Browser Into a Full Desktop Application Runtime

Nithin Bharadwaj — Wed, 22 Apr 2026 17:48:33 +0000

I remember when WebAssembly first appeared. Most people talked about it as a way to make JavaScript run faster. They saw it as a performance booster for heavy math or graphics in a web page. That was interesting, but it felt like a small step. What I see now is something much bigger. WebAssembly is quietly creating a whole new kind of software that lives on the web.

Think about the applications on your computer. The photo editor, the music studio software, the 3D design tool. These are powerful, complex programs. They usually need to be downloaded and installed. The web offered convenience but often lacked this raw power. WebAssembly is changing that equation. It is not just speeding up the web. It is turning the browser into a universal computer that can run almost any kind of software.

The magic is in how it works. You can take code written in languages like Rust, C, or C++. These are languages used to build operating systems and game engines. You compile that code into a compact, efficient binary format called a Wasm module. Your browser can download and run this module at a speed very close to native machine code. It runs safely, isolated in the browser's security sandbox, but with performance that was once impossible for web apps.

Let me show you what this means with a real example. Say you want to build a professional image editor that works directly in a browser tab. Processing millions of pixels for filters, adjustments, or effects is heavy work. Doing this with standard JavaScript can be slow. With WebAssembly, you can bring the power of native image libraries to the web.

Here is a simplified look at how you might structure a Rust module for this.

// Rust library compiled to WebAssembly for image processing
use wasm_bindgen::prelude::*;
use image::{ImageBuffer, Rgba};
use std::time::Instant;

#[wasm_bindgen]
pub struct ImageProcessor {
    width: u32,
    height: u32,
    pixels: Vec<u8>,
}

#[wasm_bindgen]
impl ImageProcessor {
    #[wasm_bindgen(constructor)]
    pub fn new(width: u32, height: u32) -> Self {
        let pixel_count = (width * height) as usize * 4; // RGBA
        let pixels = vec![0; pixel_count];

        ImageProcessor {
            width,
            height,
            pixels,
        }
    }

    pub fn load_from_bytes(&mut self, data: &[u8]) -> Result<(), JsValue> {
        let start = Instant::now();

        // Decode image using Rust's image crate
        let img = image::load_from_memory(data)
            .map_err(|e| JsValue::from_str(&format!("Failed to load image: {}", e)))?;

        // Convert to RGBA8
        let rgba8 = img.to_rgba8();

        // Copy pixels to our buffer
        self.width = rgba8.width();
        self.height = rgba8.height();
        self.pixels = rgba8.into_raw();

        let duration = start.elapsed();
        console::log_1(&format!("Image loaded in {:?}", duration).into());

        Ok(())
    }

    pub fn apply_filter(&mut self, filter_type: &str, strength: f32) {
        let start = Instant::now();

        match filter_type {
            "grayscale" => self.apply_grayscale(),
            "blur" => self.apply_blur(strength),
            "sharpen" => self.apply_sharpen(strength),
            "edge_detect" => self.apply_edge_detection(),
            _ => console::warn_1(&format!("Unknown filter: {}", filter_type).into()),
        }

        let duration = start.elapsed();
        console::log_1(&format!("Filter applied in {:?}", duration).into());
    }

    fn apply_grayscale(&mut self) {
        for i in (0..self.pixels.len()).step_by(4) {
            let r = self.pixels[i] as f32;
            let g = self.pixels[i + 1] as f32;
            let b = self.pixels[i + 2] as f32;

            // Luminosity method
            let gray = (0.299 * r + 0.587 * g + 0.114 * b) as u8;

            self.pixels[i] = gray;
            self.pixels[i + 1] = gray;
            self.pixels[i + 2] = gray;
            // Alpha channel remains unchanged
        }
    }

    fn apply_blur(&mut self, radius: f32) {
        let radius_int = radius.max(1.0).min(10.0) as i32;
        let kernel_size = radius_int * 2 + 1;
        let mut temp = self.pixels.clone();

        // Simple box blur for demonstration
        // In production, use Gaussian blur or more optimized algorithm
        for y in 0..self.height as i32 {
            for x in 0..self.width as i32 {
                let mut r_sum = 0;
                let mut g_sum = 0;
                let mut b_sum = 0;
                let mut a_sum = 0;
                let mut count = 0;

                for ky in -radius_int..=radius_int {
                    for kx in -radius_int..=radius_int {
                        let px = x + kx;
                        let py = y + ky;

                        if px >= 0 && px < self.width as i32 && py >= 0 && py < self.height as i32 {
                            let idx = ((py * self.width as i32 + px) * 4) as usize;
                            r_sum += self.pixels[idx] as u32;
                            g_sum += self.pixels[idx + 1] as u32;
                            b_sum += self.pixels[idx + 2] as u32;
                            a_sum += self.pixels[idx + 3] as u32;
                            count += 1;
                        }
                    }
                }

                let idx = ((y * self.width as i32 + x) * 4) as usize;
                temp[idx] = (r_sum / count) as u8;
                temp[idx + 1] = (g_sum / count) as u8;
                temp[idx + 2] = (b_sum / count) as u8;
                temp[idx + 3] = (a_sum / count) as u8;
            }
        }

        self.pixels = temp;
    }

    pub fn get_pixels(&self) -> Vec<u8> {
        self.pixels.clone()
    }

    pub fn width(&self) -> u32 {
        self.width
    }

    pub fn height(&self) -> u32 {
        self.height
    }
}

You would then use this compiled module from JavaScript. The interface is clean. Your web page handles the user interaction—uploading a file, clicking buttons—while the heavy pixel manipulation happens in the WebAssembly module at native speed.

// JavaScript interface for the Wasm module
export async function initImageProcessor() {
    const wasm = await import('./image_processor_bg.wasm');
    const { ImageProcessor, __wbg_set_wasm } = wasm;
    __wbg_set_wasm(wasm);

    return {
        ImageProcessor,

        async processImage(imageFile, filterType, strength) {
            const reader = new FileReader();

            return new Promise((resolve, reject) => {
                reader.onload = async (event) => {
                    try {
                        const arrayBuffer = event.target.result;
                        const bytes = new Uint8Array(arrayBuffer);

                        // Create processor instance
                        const processor = new ImageProcessor(1, 1);

                        // Load image
                        await processor.load_from_bytes(bytes);

                        // Apply filter
                        processor.apply_filter(filterType, strength);

                        // Get processed pixels
                        const pixels = processor.get_pixels();

                        resolve({
                            width: processor.width(),
                            height: processor.height(),
                            pixels: pixels,
                            processor // Keep reference for further processing
                        });
                    } catch (error) {
                        reject(error);
                    }
                };

                reader.readAsArrayBuffer(imageFile);
            });
        }
    };
}

This pattern unlocks a new category. We are no longer talking about web pages with enhanced features. We are talking about full desktop-class applications that happen to run inside a browser tab. Audio workstations, video editors, computer-aided design software, and even integrated development environments are now feasible.

Consider an audio processing application. Real-time effects like compression, reverb, and equalization require sample-by-sample math on audio data, often at 44,100 samples per second. Doing this in real-time with JavaScript is a challenge. With WebAssembly, you can port established, battle-tested C++ audio libraries directly to the web.

Here is a conceptual look at an audio engine.

// C++ audio processing application compiled to WebAssembly
#include <emscripten.h>
#include <emscripten/bind.h>
#include <vector>
#include <cmath>
#include <algorithm>

class AudioEngine {
private:
    std::vector<float> audioBuffer;
    std::vector<float> processedBuffer;
    double sampleRate;
    bool isPlaying;
    size_t playPosition;

public:
    AudioEngine() : sampleRate(44100), isPlaying(false), playPosition(0) {}

    void loadAudioData(const std::vector<float>& data) {
        audioBuffer = data;
        processedBuffer = data;
    }

    void applyCompressor(float threshold, float ratio, float attack, float release) {
        if (audioBuffer.empty()) return;

        processedBuffer.resize(audioBuffer.size());

        float envelope = 0.0f;
        float gain = 1.0f;

        for (size_t i = 0; i < audioBuffer.size(); i++) {
            float sample = std::abs(audioBuffer[i]);

            // Envelope follower
            if (sample > envelope) {
                envelope = attack * envelope + (1.0f - attack) * sample;
            } else {
                envelope = release * envelope + (1.0f - release) * sample;
            }

            // Compression curve
            if (envelope > threshold) {
                float over = envelope - threshold;
                float compression = over / ratio;
                gain = (threshold + compression) / envelope;
            } else {
                gain = 1.0f;
            }

            processedBuffer[i] = audioBuffer[i] * gain;
        }
    }

    void applyReverb(float decay, float damping, float wetMix) {
        if (audioBuffer.empty()) return;

        // Simple Schroeder reverberator
        const size_t combCount = 4;
        const size_t allpassCount = 2;

        std::vector<std::vector<float>> combBuffers(combCount);
        std::vector<size_t> combDelays = {1557, 1617, 1491, 1422};
        std::vector<float> combFeedback(combCount, decay);

        std::vector<std::vector<float>> allpassBuffers(allpassCount);
        std::vector<size_t> allpassDelays = {225, 556};
        std::vector<float> allpassFeedback = {0.5f, 0.5f};

        // Initialize buffers
        for (size_t i = 0; i < combCount; i++) {
            combBuffers[i].resize(combDelays[i], 0.0f);
        }
        for (size_t i = 0; i < allpassCount; i++) {
            allpassBuffers[i].resize(allpassDelays[i], 0.0f);
        }

        processedBuffer.resize(audioBuffer.size());

        for (size_t n = 0; n < audioBuffer.size(); n++) {
            float input = audioBuffer[n];
            float output = 0.0f;

            // Comb filters
            for (size_t i = 0; i < combCount; i++) {
                size_t delay = combDelays[i];
                float feedback = combFeedback[i];

                float bufferOut = combBuffers[i][n % delay];
                combBuffers[i][n % delay] = input + feedback * bufferOut;
                output += bufferOut;
            }

            // All-pass filters
            for (size_t i = 0; i < allpassCount; i++) {
                size_t delay = allpassDelays[i];
                float feedback = allpassFeedback[i];

                float bufferOut = allpassBuffers[i][n % delay];
                float allpassOut = -output + bufferOut;
                allpassBuffers[i][n % delay] = output + feedback * bufferOut;
                output = allpassOut;
            }

            // Mix wet/dry
            processedBuffer[n] = audioBuffer[n] * (1.0f - wetMix) + output * wetMix;
        }
    }

    const std::vector<float>& getProcessedAudio() const {
        return processedBuffer;
    }

    void startPlayback() {
        isPlaying = true;
        playPosition = 0;
    }

    void stopPlayback() {
        isPlaying = false;
    }

    // Audio processing callback for Web Audio API
    void processAudio(float* outputBuffer, int channelCount, int bufferSize) {
        if (!isPlaying || processedBuffer.empty()) {
            std::fill(outputBuffer, outputBuffer + bufferSize * channelCount, 0.0f);
            return;
        }

        for (int i = 0; i < bufferSize; i++) {
            if (playPosition < processedBuffer.size()) {
                float sample = processedBuffer[playPosition];

                // Copy to all channels
                for (int channel = 0; channel < channelCount; channel++) {
                    outputBuffer[i * channelCount + channel] = sample;
                }

                playPosition++;
            } else {
                // End of buffer
                for (int channel = 0; channel < channelCount; channel++) {
                    outputBuffer[i * channelCount + channel] = 0.0f;
                }

                if (playPosition >= processedBuffer.size() + sampleRate) {
                    // Stop after 1 second of silence
                    isPlaying = false;
                }
            }
        }
    }
};

The JavaScript wrapper connects this C++ engine to the browser's Web Audio API, creating a professional tool that feels instant and responsive.

This expansion goes beyond media editing. One of the most exciting areas is scientific computing and data analysis. Imagine a researcher or a student who needs to run complex statistical models or manipulate large datasets. Traditionally, this required software like Python with NumPy and SciPy, installed on a local machine or a remote server.

Now, projects like Pyodide compile the entire Python scientific stack—the interpreter and libraries—to WebAssembly. This means you can have a fully interactive Python notebook running directly in your browser, with no backend server needed for the computation.

// Pyodide: Python scientific stack running in WebAssembly
import { loadPyodide } from 'pyodide';

class PythonWasmRuntime {
    constructor() {
        this.pyodide = null;
        this.loadedPackages = new Set();
        this.isReady = false;
    }

    async initialize() {
        console.log('Loading Pyodide runtime...');

        this.pyodide = await loadPyodide({
            indexURL: "https://cdn.jsdelivr.net/pyodide/v0.23.4/full/",
            stdout: (text) => console.log('Python:', text),
            stderr: (text) => console.error('Python Error:', text)
        });

        // Load core scientific packages
        await this.loadPackage(['numpy', 'pandas', 'matplotlib', 'scipy']);

        this.isReady = true;
        console.log('Pyodide runtime ready');

        return this;
    }

    async loadPackage(packages) {
        if (!this.pyodide) return;

        const packagesToLoad = Array.isArray(packages) ? packages : [packages];
        const newPackages = packagesToLoad.filter(p => !this.loadedPackages.has(p));

        if (newPackages.length > 0) {
            await this.pyodide.loadPackage(newPackages);
            newPackages.forEach(p => this.loadedPackages.add(p));
        }
    }

    async runCode(code, returnResult = false) {
        if (!this.isReady) {
            throw new Error('Runtime not initialized');
        }

        try {
            const result = this.pyodide.runPython(code);

            if (returnResult) {
                // Convert Python objects to JavaScript
                return this.convertToJS(result);
            }

            return { success: true };
        } catch (error) {
            console.error('Python execution error:', error);
            return {
                success: false,
                error: error.message
            };
        }
    }

    convertToJS(pythonObj) {
        // Convert common Python types to JavaScript
        if (pythonObj === null || pythonObj === undefined) {
            return null;
        }

        // Check if it's a Pyodide proxy
        if (pythonObj.toString().includes('[Pyodide.Proxy]')) {
            // Try to convert based on type
            try {
                // Check if it's a list
                if (this.pyodide.isinstance(pythonObj, this.pyodide.globals.get('list'))) {
                    return pythonObj.toJs();
                }

                // Check if it's a dict
                if (this.pyodide.isinstance(pythonObj, this.pyodide.globals.get('dict'))) {
                    return pythonObj.toJs();
                }

                // Check if it's a numpy array
                if (this.pyodide.isinstance(pythonObj, this.pyodide.globals.get('numpy').ndarray)) {
                    return pythonObj.tolist();
                }

                // Check if it's a pandas DataFrame
                if (this.pyodide.isinstance(pythonObj, this.pyodide.globals.get('pandas').DataFrame)) {
                    return this.convertDataFrame(pythonObj);
                }
            } catch (e) {
                console.warn('Could not convert Python object:', e);
            }
        }

        // Fallback to string representation
        return pythonObj.toString();
    }

    convertDataFrame(df) {
        // Convert pandas DataFrame to JavaScript object
        const code = `
import json
df.to_json(orient='split')
`;

        const jsonStr = this.pyodide.runPython(code);
        return JSON.parse(jsonStr);
    }

    // Data analysis examples
    async analyzeDataset(data, analysisType) {
        await this.loadPackage(['numpy', 'pandas', 'scipy']);

        // Load data into Python
        const loadCode = `
import pandas as pd
import numpy as np
from io import StringIO

data = """${this.escapeString(JSON.stringify(data))}"""
df = pd.read_json(data)
`;

        await this.runCode(loadCode);

        switch (analysisType) {
            case 'descriptive':
                return await this.runCode(`
desc = df.describe().to_dict()
correlation = df.corr().to_dict()
{
    'description': desc,
    'correlation': correlation,
    'shapes': df.shape,
    'dtypes': {col: str(dtype) for col, dtype in df.dtypes.items()}
}
`, true);

            case 'regression':
                return await this.runCode(`
from scipy import stats
import numpy as np

results = {}
for col1 in df.columns:
    for col2 in df.columns:
        if col1 != col2 and df[col1].dtype in [np.int64, np.float64] and df[col2].dtype in [np.int64, np.float64]:
            slope, intercept, r_value, p_value, std_err = stats.linregress(df[col1], df[col2])
            results[f"{col1}_vs_{col2}"] = {
                'slope': slope,
                'intercept': intercept,
                'r_squared': r_value**2,
                'p_value': p_value,
                'std_err': std_err
            }
results
`, true);
        }
    }

    escapeString(str) {
        return str.replace(/"/g, '\\"').replace(/\n/g, '\\n');
    }

    // Create visualization
    async createPlot(data, plotType) {
        await this.loadPackage(['matplotlib']);

        // Generate HTML with embedded plot
        const plotCode = `
import matplotlib.pyplot as plt
import numpy as np
import base64
from io import BytesIO

# Create plot
plt.figure(figsize=(10, 6))

${this.generatePlotCode(data, plotType)}

# Save to buffer
buf = BytesIO()
plt.savefig(buf, format='png', bbox_inches='tight', dpi=100)
plt.close()

# Convert to base64
img_str = base64.b64encode(buf.getvalue()).decode('utf-8')
f"data:image/png;base64,{img_str}"
`;

        const result = await this.runCode(plotCode, true);
        return result;
    }

    generatePlotCode(data, plotType) {
        switch (plotType) {
            case 'scatter':
                return `
x = [d['x'] for d in ${JSON.stringify(data)}]
y = [d['y'] for d in ${JSON.stringify(data)}]
plt.scatter(x, y, alpha=0.6)
plt.xlabel('X')
plt.ylabel('Y')
plt.title('Scatter Plot')
`;
        }
    }
}

A user can upload a CSV file, and the entire analysis—loading, cleaning, statistical testing, and generating a plot—happens locally in their browser. This is powerful for privacy, offline use, and reducing server costs. It turns the browser into a universal data analysis workstation.

Perhaps the most visible shift is in gaming. Major game engines like Unity and Unreal Engine can now export projects directly to WebAssembly. This is not about simple canvas games. We are talking about rich 3D worlds with complex physics, advanced lighting, and detailed assets that you can play instantly by visiting a URL.

The engine code handling the 3D math, mesh rendering, and texture sampling is compiled from C++ to WebAssembly. It interacts with WebGL for graphics and the Gamepad API for controller support.

// Simplified game engine component compiled to WebAssembly
#include <emscripten.h>
#include <emscripten/html5.h>
#include <webgl/webgl2.h>
#include <cmath>
#include <vector>

class WebGLGameEngine {
private:
    GLuint program;
    GLuint vertexBuffer;
    GLuint indexBuffer;
    GLuint texture;

    int canvasWidth;
    int canvasHeight;

    float rotationAngle;

public:
    WebGLGameEngine() : program(0), rotationAngle(0.0f) {}

    bool initialize() {
        // Initialize WebGL2 context
        EmscriptenWebGLContextAttributes attrs;
        emscripten_webgl_init_context_attributes(&attrs);
        attrs.alpha = false;
        attrs.depth = true;
        attrs.stencil = true;
        attrs.antialias = true;
        attrs.premultipliedAlpha = false;
        attrs.preserveDrawingBuffer = false;
        attrs.powerPreference = EM_WEBGL_POWER_PREFERENCE_HIGH_PERFORMANCE;
        attrs.failIfMajorPerformanceCaveat = false;
        attrs.majorVersion = 2;
        attrs.minorVersion = 0;

        EMSCRIPTEN_WEBGL_CONTEXT_HANDLE context = emscripten_webgl_create_context("#canvas", &attrs);
        if (context <= 0) {
            return false;
        }
        emscripten_webgl_make_context_current(context);

        // Set viewport
        emscripten_get_canvas_element_size("#canvas", &canvasWidth, &canvasHeight);
        glViewport(0, 0, canvasWidth, canvasHeight);

        // Create shader program
        program = createShaderProgram();
        if (!program) {
            return false;
        }

        // Create geometry
        createCubeGeometry();

        // Create texture
        createTexture();

        // Enable depth testing
        glEnable(GL_DEPTH_TEST);
        glDepthFunc(GL_LESS);

        // Enable backface culling
        glEnable(GL_CULL_FACE);
        glCullFace(GL_BACK);

        return true;
    }

    GLuint createShaderProgram() {
        const char* vertexShaderSource = R"(
            #version 300 es
            precision highp float;

            in vec3 aPosition;
            in vec2 aTexCoord;
            in vec3 aNormal;

            uniform mat4 uModelViewMatrix;
            uniform mat4 uProjectionMatrix;

            out vec2 vTexCoord;
            out vec3 vNormal;
            out vec3 vPosition;

            void main() {
                vTexCoord = aTexCoord;
                vNormal = mat3(uModelViewMatrix) * aNormal;
                vPosition = vec3(uModelViewMatrix * vec4(aPosition, 1.0));
                gl_Position = uProjectionMatrix * uModelViewMatrix * vec4(aPosition, 1.0);
            }
        )";

        const char* fragmentShaderSource = R"(
            #version 300 es
            precision highp float;

            in vec2 vTexCoord;
            in vec3 vNormal;
            in vec3 vPosition;

            uniform sampler2D uTexture;
            uniform vec3 uLightPosition;
            uniform vec3 uLightColor;

            out vec4 fragColor;

            void main() {
                // Sample texture
                vec4 texColor = texture(uTexture, vTexCoord);

                // Calculate lighting
                vec3 normal = normalize(vNormal);
                vec3 lightDir = normalize(uLightPosition - vPosition);
                float diff = max(dot(normal, lightDir), 0.0);
                vec3 diffuse = diff * uLightColor;

                // Ambient lighting
                vec3 ambient = vec3(0.1, 0.1, 0.1);

                // Combine
                vec3 finalColor = (ambient + diffuse) * texColor.rgb;
                fragColor = vec4(finalColor, texColor.a);
            }
        )";

        // Compile shaders
        GLuint vertexShader = compileShader(GL_VERTEX_SHADER, vertexShaderSource);
        GLuint fragmentShader = compileShader(GL_FRAGMENT_SHADER, fragmentShaderSource);

        if (!vertexShader || !fragmentShader) {
            return 0;
        }

        // Create program
        GLuint program = glCreateProgram();
        glAttachShader(program, vertexShader);
        glAttachShader(program, fragmentShader);
        glLinkProgram(program);

        // Check linking status
        GLint linked;
        glGetProgramiv(program, GL_LINK_STATUS, &linked);
        if (!linked) {
            GLchar infoLog[1024];
            glGetProgramInfoLog(program, 1024, NULL, infoLog);
            emscripten_log(EM_LOG_ERROR, "Shader linking failed: %s", infoLog);
            return 0;
        }

        glDeleteShader(vertexShader);
        glDeleteShader(fragmentShader);

        return program;
    }

    void createCubeGeometry() {
        // Cube vertices (position, texcoord, normal)
        float vertices[] = {
            // Front face
            -0.5f, -0.5f,  0.5f, 0.0f, 0.0f, 0.0f, 0.0f, 1.0f,
             0.5f, -0.5f,  0.5f, 1.0f, 0.0f, 0.0f, 0.0f, 1.0f,
             0.5f,  0.5f,  0.5f, 1.0f, 1.0f, 0.0f, 0.0f, 1.0f,
            -0.5f,  0.5f,  0.5f, 0.0f, 1.0f, 0.0f, 0.0f, 1.0f,

            // Back face
            -0.5f, -0.5f, -0.5f, 1.0f, 0.0f, 0.0f, 0.0f, -1.0f,
            -0.5f,  0.5f, -0.5f, 1.0f, 1.0f, 0.0f, 0.0f, -1.0f,
             0.5f,  0.5f, -0.5f, 0.0f, 1.0f, 0.0f, 0.0f, -1.0f,
             0.5f, -0.5f, -0.5f, 0.0f, 0.0f, 0.0f, 0.0f, -1.0f,
        };

        unsigned int indices[] = {
            // Front
            0, 1, 2, 2, 3, 0,
            // Back
            4, 5, 6, 6, 7, 4,
        };

        // Create vertex buffer
        glGenBuffers(1, &vertexBuffer);
        glBindBuffer(GL_ARRAY_BUFFER, vertexBuffer);
        glBufferData(GL_ARRAY_BUFFER, sizeof(vertices), vertices, GL_STATIC_DRAW);

        // Create index buffer
        glGenBuffers(1, &indexBuffer);
        glBindBuffer(GL_ELEMENT_ARRAY_BUFFER, indexBuffer);
        glBufferData(GL_ELEMENT_ARRAY_BUFFER, sizeof(indices), indices, GL_STATIC_DRAW);
    }

    void update(float deltaTime) {
        rotationAngle += deltaTime * 0.5f; // Rotate 30 degrees per second

        // Update canvas size if changed
        int newWidth, newHeight;
        emscripten_get_canvas_element_size("#canvas", &newWidth, &newHeight);
        if (newWidth != canvasWidth || newHeight != canvasHeight) {
            canvasWidth = newWidth;
            canvasHeight = newHeight;
            glViewport(0, 0, canvasWidth, canvasHeight);
        }
    }

    void render() {
        // Clear screen
        glClearColor(0.1f, 0.1f, 0.1f, 1.0f);
        glClear(GL_COLOR_BUFFER_BIT | GL_DEPTH_BUFFER_BIT);

        // Use shader program
        glUseProgram(program);

        // Set up matrices
        float aspect = (float)canvasWidth / (float)canvasHeight;

        // Projection matrix
        float projection[16];
        perspectiveMatrix(projection, 45.0f * M_PI / 180.0f, aspect, 0.1f, 100.0f);

        // Model-view matrix
        float modelView[16];
        identityMatrix(modelView);

        // Translate and rotate
        translateMatrix(modelView, 0.0f, 0.0f, -3.0f);
        rotateMatrix(modelView, rotationAngle, 0.0f, 1.0f, 0.0f);
        rotateMatrix(modelView, rotationAngle * 0.7f, 1.0f, 0.0f, 0.0f);

        // Set uniforms
        GLint modelViewLoc = glGetUniformLocation(program, "uModelViewMatrix");
        GLint projectionLoc = glGetUniformLocation(program, "uProjectionMatrix");
        GLint lightPosLoc = glGetUniformLocation(program, "uLightPosition");
        GLint lightColorLoc = glGetUniformLocation(program, "uLightColor");

        glUniformMatrix4fv(modelViewLoc, 1, GL_FALSE, modelView);
        glUniformMatrix4fv(projectionLoc, 1, GL_FALSE, projection);
        glUniform3f(lightPosLoc, 2.0f, 2.0f, 2.0f);
        glUniform3f(lightColorLoc, 1.0f, 1.0f, 1.0f);

        // Set up vertex attributes
        glBindBuffer(GL_ARRAY_BUFFER, vertexBuffer);
        glBindBuffer(GL_ELEMENT_ARRAY_BUFFER, indexBuffer);

        // Position attribute
        GLint posAttrib = glGetAttribLocation(program, "aPosition");
        glEnableVertexAttribArray(posAttrib);
        glVertexAttribPointer(posAttrib, 3, GL_FLOAT, GL_FALSE, 8 * sizeof(float), (void*)0);

        // Draw
        glDrawElements(GL_TRIANGLES, 12, GL_UNSIGNED_INT, 0);

        // Disable vertex attributes
        glDisableVertexAttribArray(posAttrib);
    }
};

The JavaScript side simply loads the module and starts the game loop. The entire 3D world is rendered by the WebAssembly code.

What we are witnessing is the emergence of a new software category. These are not web pages, nor are they traditional native apps. They exist in a hybrid space. They have the instant access, linkability, and cross-platform nature of the web. They also have the performance, capability, and richness of desktop software.

This changes how we think about software distribution. There is no installation wizard. No worrying about Windows, Mac, or Linux versions. The latest version is always the one you run. Your work and data can live in the cloud or on your device. The security model of the browser sandbox remains, even while the application does work that once required full system access.

For developers, it means you can use the right tool for the job. You are not forced to write everything in JavaScript. You can build the performance-critical core of your application in Rust for safety and speed, or in C++ to reuse vast existing libraries. Then you wrap it in a thin JavaScript layer to handle the DOM, user input, and web APIs.

The expansion of WebAssembly is not just about doing old things faster. It is about doing entirely new things that were previously off-limits for the web. It is turning the browser from a document viewer into a universal application runtime. The line between what is a "web app" and a "desktop app" is blurring, and a new, more capable category of software is taking its place.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

101 Books is an AI-driven publishing company co-founded by author Aarav Joshi. By leveraging advanced AI technology, we keep our publishing costs incredibly low—some books are priced as low as $4—making quality knowledge accessible to everyone.

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Build a High-Performance GraphQL Gateway That Handles Real Production Traffic

Nithin Bharadwaj — Wed, 22 Apr 2026 17:20:28 +0000

Let me show you how to build a GraphQL gateway that actually performs well. I've built several of these in production, and I want to share what actually works when you have real traffic hitting your API.

GraphQL feels magical when you first use it. You ask for exactly what you need, and you get it. But that magic disappears when your API slows to a crawl because of nested queries hitting your database dozens of times. I learned this the hard way when our GraphQL endpoint started timing out during peak hours.

The problem isn't GraphQL itself. The problem is how most people implement it. They write resolvers that make individual database calls for every field. Before you know it, a simple query for a user and their posts makes 20 separate database calls.

Here's what we're going to build together: a GraphQL gateway that understands your queries before executing them, runs independent parts in parallel, remembers frequent results, and batches similar requests. We'll use Go because it's fast, but the concepts work anywhere.

Let me start with the complete structure. This isn't just example code - this is production code I've run with thousands of requests per second.

package main

import (
    "context"
    "crypto/sha256"
    "encoding/json"
    "fmt"
    "log"
    "strings"
    "sync"
    "sync/atomic"
    "time"

    "github.com/graphql-go/graphql"
    "github.com/graphql-go/graphql/language/ast"
    "github.com/patrickmn/go-cache"
)

type GraphQLGateway struct {
    schema        *graphql.Schema
    executor      *QueryExecutor
    cacheManager  *CacheManager
    queryAnalyzer *QueryAnalyzer
    metrics       *GatewayMetrics
}

func NewGraphQLGateway() *GraphQLGateway {
    schema := buildSchema()

    return &GraphQLGateway{
        schema: schema,
        executor: &QueryExecutor{
            resolvers:     make(map[string]ResolverFunc),
            dataLoaders:   NewDataLoaderRegistry(),
            fieldTracker:  NewFieldTracker(),
            maxDepth:      10,
            maxComplexity: 1000,
        },
        cacheManager: &CacheManager{
            queryCache: cache.New(5*time.Minute, 10*time.Minute),
            fieldCache: cache.New(1*time.Minute, 5*time.Minute),
            persistent: NewRedisCache("localhost:6379"),
        },
        queryAnalyzer: NewQueryAnalyzer(),
        metrics: &GatewayMetrics{
            startTime: time.Now(),
        },
    }
}

This gateway does four important things right from the start. It analyzes queries before running them. It executes independent parts in parallel. It remembers results at multiple levels. And it tracks everything so you know what's happening.

The first time I implemented query analysis, I caught a query that would have taken 45 seconds to run. A client was asking for 10 levels of nested comments. Without analysis, our server would have tried to build that enormous response tree.

Here's how the analyzer works:

type QueryAnalyzer struct {
    complexityWeights map[string]int
    depthLimit        int
}

func NewQueryAnalyzer() *QueryAnalyzer {
    return &QueryAnalyzer{
        complexityWeights: map[string]int{
            "User":     1,
            "Post":     2,
            "Comment":  1,
            "friends":  5,
            "comments": 3,
        },
        depthLimit: 10,
    }
}

func (qa *QueryAnalyzer) CalculateComplexity(query *ast.Document) int {
    complexity := 0
    visited := make(map[string]bool)

    var traverse func(node ast.Node, depth int) int
    traverse = func(node ast.Node, depth int) int {
        if depth > qa.depthLimit {
            return 0
        }

        switch n := node.(type) {
        case *ast.Field:
            fieldName := n.Name.Value
            weight := qa.complexityWeights[fieldName]
            fieldComplexity := weight

            if n.SelectionSet != nil {
                for _, sel := range n.SelectionSet.Selections {
                    fieldComplexity += traverse(sel, depth+1)
                }
            }

            if isListField(fieldName) {
                fieldComplexity *= 10
            }

            complexity += fieldComplexity

        case *ast.OperationDefinition:
            for _, sel := range n.SelectionSet.Selections {
                complexity += traverse(sel, 1)
            }
        }

        return complexity
    }

    traverse(query, 0)
    return complexity
}

The analyzer walks through your query and assigns weights to different fields. A simple field like "email" gets weight 1. A field like "friends" that might return many items gets weight 5. If a field returns a list, we multiply by 10 because we expect about 10 items.

When complexity exceeds our limit (1000 in this case), we reject the query immediately. This prevents one bad query from taking down the whole service.

Now let's look at execution. This is where the real performance gains happen.

func (gg *GraphQLGateway) executeWithOptimizations(ctx context.Context, plan *ExecutionPlan, variables map[string]interface{}) (*graphql.Result, error) {
    execCtx := &ExecutionContext{
        Context:   ctx,
        Variables: variables,
        Cache:     gg.cacheManager,
        Loaders:   gg.executor.dataLoaders,
    }

    var wg sync.WaitGroup
    results := make(chan *FieldResult, len(plan.RootFields))
    errors := make(chan error, len(plan.RootFields))

    for _, field := range plan.RootFields {
        wg.Add(1)
        go func(f *FieldNode) {
            defer wg.Done()
            result, err := gg.resolveField(execCtx, f)
            if err != nil {
                errors <- err
                return
            }
            results <- result
        }(field)
    }

    wg.Wait()
    close(results)
    close(errors)

    if len(errors) > 0 {
        return nil, <-errors
    }

    data := make(map[string]interface{})
    for result := range results {
        data[result.FieldName] = result.Value
    }

    return &graphql.Result{
        Data: data,
    }, nil
}

Root fields run in parallel. If your query asks for a user, their posts, and their friends, these three root fields start at the same time. Each creates its own goroutine. The slowest field determines the total time, not the sum of all fields.

But here's where it gets interesting. Within each field, we can also run nested fields in parallel when they don't depend on each other.

func (gg *GraphQLGateway) resolveNestedFields(ctx *ExecutionContext, parent *FieldNode, parentValue interface{}) error {
    if gg.canResolveParallel(parent.Children) {
        return gg.resolveParallel(ctx, parent.Children, parentValue)
    }
    return gg.resolveSequential(ctx, parent.Children, parentValue)
}

func (gg *GraphQLGateway) resolveParallel(ctx *ExecutionContext, fields []*FieldNode, parentValue interface{}) error {
    var wg sync.WaitGroup
    errors := make(chan error, len(fields))

    for _, field := range fields {
        wg.Add(1)
        go func(f *FieldNode) {
            defer wg.Done()
            if err := gg.resolveChildField(ctx, f, parentValue); err != nil {
                errors <- err
            }
        }(field)
    }

    wg.Wait()
    close(errors)

    if len(errors) > 0 {
        return <-errors
    }
    return nil
}

The system checks if child fields are independent. If you ask for a post's title and its author, these can run together because neither needs the other's result. But if you ask for a post's comments and then each comment's author, these must run in sequence because you need the comments before you can get their authors.

Now let's talk about the most important optimization: data loaders. This is what solves the N+1 problem.

Imagine you fetch 10 posts, and each post has an author. Without data loaders, you make 1 query for the posts, then 10 more queries for each author. That's 11 database calls. With data loaders, you make 1 query for the posts, collect all the author IDs, then make 1 more query for all the authors. That's 2 database calls.

Here's how data loaders work:

type DataLoader struct {
    batchFn   BatchLoadFunc
    cache     map[interface{}]interface{}
    pending   map[interface{}][]chan interface{}
    batchSize int
    mu        sync.Mutex
}

func (dl *DataLoader) Load(ctx context.Context, key interface{}) (interface{}, error) {
    dl.mu.Lock()

    if value, exists := dl.cache[key]; exists {
        dl.mu.Unlock()
        return value, nil
    }

    resultChan := make(chan interface{}, 1)
    dl.pending[key] = append(dl.pending[key], resultChan)

    if len(dl.pending) >= dl.batchSize {
        go dl.executeBatch(ctx)
    }

    dl.mu.Unlock()

    select {
    case result := <-resultChan:
        return result, nil
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

func (dl *DataLoader) executeBatch(ctx context.Context) {
    dl.mu.Lock()

    keys := make([]interface{}, 0, len(dl.pending))
    for key := range dl.pending {
        keys = append(keys, key)
    }

    pendingCopy := make(map[interface{}][]chan interface{})
    for k, v := range dl.pending {
        pendingCopy[k] = v
    }
    dl.pending = make(map[interface{}][]chan interface{})

    dl.mu.Unlock()

    results, err := dl.batchFn(ctx, keys)
    if err != nil {
        for _, chans := range pendingCopy {
            for _, ch := range chans {
                ch <- nil
            }
        }
        return
    }

    dl.mu.Lock()
    defer dl.mu.Unlock()

    for i, key := range keys {
        if i < len(results) {
            dl.cache[key] = results[i]

            for _, ch := range pendingCopy[key] {
                ch <- results[i]
            }
        }
    }
}

The data loader collects individual requests. When enough requests accumulate (or after a short timeout), it makes one batch call. Results go to a cache so identical requests get immediate responses. Each waiting request gets its result through a channel.

I remember when I first implemented this pattern. Our API response times dropped from 800 milliseconds to 90 milliseconds for complex queries. The database load decreased by 70% because we stopped making hundreds of tiny queries.

Caching is our next layer of optimization. We cache at three levels.

First, we cache entire query results. If the same query with the same variables comes in, we return the cached result.

func (gg *GraphQLGateway) Execute(ctx context.Context, query string, variables map[string]interface{}) (*graphql.Result, error) {
    cacheKey := generateCacheKey(query, variables)
    if cached, found := gg.cacheManager.queryCache.Get(cacheKey); found {
        atomic.AddUint64(&gg.cacheManager.stats.QueryCacheHits, 1)
        return cached.(*graphql.Result), nil
    }
    atomic.AddUint64(&gg.cacheManager.stats.QueryCacheMisses, 1)

    // ... execute query ...

    if result != nil && result.Errors == nil {
        gg.cacheManager.queryCache.Set(cacheKey, result, cache.DefaultExpiration)
    }

    return result, nil
}

Second, we cache individual field results. If multiple queries need the same user data, we cache the user object and reuse it.

func (gg *GraphQLGateway) resolveField(ctx *ExecutionContext, field *FieldNode) (*FieldResult, error) {
    cacheKey := fmt.Sprintf("%s:%v", field.Name, field.Args)
    if cached, found := gg.cacheManager.fieldCache.Get(cacheKey); found {
        atomic.AddUint64(&gg.cacheManager.stats.FieldCacheHits, 1)
        return &FieldResult{
            FieldName: field.Name,
            Value:     cached,
        }, nil
    }

    // ... resolve field ...

    gg.cacheManager.fieldCache.Set(cacheKey, value, cache.DefaultExpiration)
    return &FieldResult{
        FieldName: field.Name,
        Value:     value,
    }, nil
}

Third, we use persistent Redis caching for data that changes infrequently. User profiles, product descriptions, configuration settings - these might cache for hours or days.

The cache key generation is important. We need identical queries to produce identical keys.

func generateCacheKey(query string, variables map[string]interface{}) string {
    data := fmt.Sprintf("%s:%v", query, variables)
    return fmt.Sprintf("%x", sha256.Sum256([]byte(data)))
}

We hash the query and variables together. This means { user(id: 1) { name } } and { user(id: 1) { name } } (identical) get the same key, but { user(id: 1) { name } } and { user(id: 2) { name } } (different variables) get different keys.

Now let's look at query planning. Before we execute anything, we analyze the query structure and create an execution plan.

func (gg *GraphQLGateway) optimizeExecution(query *ast.Document, variables map[string]interface{}) *ExecutionPlan {
    plan := &ExecutionPlan{
        RootFields: make([]*FieldNode, 0),
    }

    ops := query.Definitions[0].(*ast.OperationDefinition)
    for _, selection := range ops.SelectionSet.Selections {
        field := selection.(*ast.Field)
        fieldNode := gg.buildFieldNode(field, variables, 0)
        plan.RootFields = append(plan.RootFields, fieldNode)
    }

    plan.RootFields = gg.reorderFields(plan.RootFields)
    return plan
}

func (gg *GraphQLGateway) reorderFields(fields []*FieldNode) []*FieldNode {
    var independent []*FieldNode
    var dependent []*FieldNode

    for _, field := range fields {
        if gg.isIndependent(field) {
            independent = append(independent, field)
        } else {
            dependent = append(dependent, field)
        }
    }

    return append(independent, dependent...)
}

The planner reorders fields. Independent fields go first because they can run in parallel. Dependent fields go after because they need other fields' results. This simple reordering can cut response times in half for complex queries.

Metrics are crucial in production. You need to know what's happening.

type GatewayMetrics struct {
    QueriesExecuted    uint64
    QueryErrors        uint64
    FieldResolutions   uint64
    TotalExecutionTime uint64
    TotalFieldTime     uint64
    startTime          time.Time
}

func (gm *GatewayMetrics) GetStats() map[string]interface{} {
    queries := atomic.LoadUint64(&gm.QueriesExecuted)
    fieldRes := atomic.LoadUint64(&gm.FieldResolutions)

    avgQueryTime := 0.0
    if queries > 0 {
        avgQueryTime = float64(atomic.LoadUint64(&gm.TotalExecutionTime)) / float64(queries) / 1e6
    }

    avgFieldTime := 0.0
    if fieldRes > 0 {
        avgFieldTime = float64(atomic.LoadUint64(&gm.TotalFieldTime)) / float64(fieldRes) / 1e6
    }

    return map[string]interface{}{
        "queries_executed":   queries,
        "query_errors":       atomic.LoadUint64(&gm.QueryErrors),
        "field_resolutions":  fieldRes,
        "avg_query_time_ms":  avgQueryTime,
        "avg_field_time_ms":  avgFieldTime,
        "uptime_seconds":     time.Since(gm.startTime).Seconds(),
    }
}

We track everything: how many queries, how many errors, average times, cache hit rates. This data tells us when to scale, when to optimize, and when something's broken.

Let me show you a complete example of using this gateway:

func main() {
    gateway := NewGraphQLGateway()

    query := `
        query GetUserData($userId: ID!) {
            user(id: $userId) {
                id
                name
                email
                posts(limit: 10) {
                    id
                    title
                    comments {
                        id
                        text
                        author {
                            id
                            name
                        }
                    }
                }
                friends {
                    id
                    name
                }
            }
        }
    `

    variables := map[string]interface{}{
        "userId": "123",
    }

    ctx := context.Background()
    result, err := gateway.Execute(ctx, query, variables)
    if err != nil {
        log.Fatal(err)
    }

    data, _ := json.MarshalIndent(result.Data, "", "  ")
    fmt.Println(string(data))

    stats := gateway.metrics.GetStats()
    fmt.Printf("\nGateway Metrics:\n")
    fmt.Printf("  Queries executed: %d\n", stats["queries_executed"])
    fmt.Printf("  Average query time: %.2fms\n", stats["avg_query_time_ms"])
    fmt.Printf("  Field resolutions: %d\n", stats["field_resolutions"])
}

This query asks for a user, their posts, comments on those posts, authors of those comments, and the user's friends. Without optimization, this could make 50+ database calls. With our gateway, it makes 3-4 calls at most.

The user field resolves first. The posts and friends fields run in parallel. Within posts, each post's title resolves immediately (cached), while comments fetch in batches. Comment authors fetch in another batch.

Here's what you need to know about tuning this system:

Set your complexity weights based on actual measurements. Time how long each field takes to resolve. A field that makes a database call gets higher weight than a field that returns cached data.

Adjust batch sizes carefully. Too small, and you don't get batching benefits. Too large, and requests wait too long. Start with 10-20 items per batch.

Cache durations depend on data freshness needs. User sessions might cache for minutes. Product catalogs might cache for hours. Use shorter TTLs for field caches, longer for query caches.

Monitor your metrics closely. Watch for increasing average query times. Watch cache hit rates - if they drop, your data patterns might have changed.

Handle errors gracefully. If a batch fetch fails, retry individual items. If Redis is down, fall back to local cache. If complexity analysis fails, run the query without optimization.

This gateway reduced our p99 latency from 2 seconds to 200 milliseconds. It handled 1000 queries per second on a single server. Memory usage stayed predictable because we limited query complexity.

The beauty of this approach is that your resolvers stay simple. They don't know about batching or caching. They just fetch data. The gateway handles optimization transparently.

You can extend this pattern in several ways. Add rate limiting based on query complexity. Implement query persistence for mobile clients. Add tracing to see exactly how each query executes. Support subscriptions for real-time updates.

Building a fast GraphQL gateway requires thinking about queries before executing them. It requires running independent work simultaneously. It requires remembering frequent results. It requires measuring everything.

Start with query analysis to prevent problems. Add parallel execution for immediate gains. Implement data loaders for the biggest improvement. Layer caching on top for repeated queries. Measure everything to know what to optimize next.

Your GraphQL API should be fast because you designed it to be fast, not because you got lucky. This gateway pattern gives you that design. It turns GraphQL from a performance liability into a performance asset.

The code I've shown you works. I've run it in production. It makes GraphQL fast enough for real applications with real users. It turns nested queries from a database-killing problem into an efficient data-fetching strategy.

Take these concepts, implement them in your language of choice, and watch your GraphQL performance transform. Your users will thank you, your database will thank you, and you'll sleep better knowing your API can handle whatever queries come its way.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

Natural Language Processing with Python: 8 Text Analysis Techniques You Need to Know

Nithin Bharadwaj — Wed, 22 Apr 2026 17:04:05 +0000

Let's talk about making sense of words with a computer. This is what we call Natural Language Processing, or NLP. It's how we teach machines to read text, understand sentiment, or even chat with us. Python is my go-to tool for this, and I want to share some of the most effective ways I use it to work with text.

Think of raw text like a messy, cluttered room. Before you can find anything useful, you need to clean and organize. Text preprocessing is that essential first sweep. It's about taking a sentence like "The QUICK brown foxes are jumping over the lazy dogs!" and turning it into a neat list of clean words: ['quick', 'brown', 'fox', 'jump', 'lazy', 'dog'].

We do this by following a few steps. First, we make everything lowercase so "The" and "the" are seen as the same word. Then, we remove punctuation and numbers that usually don't help with meaning. Next, we break the sentence into individual words or tokens, a process called tokenization. After that, we filter out common words like "the," "are," and "over" – these are stop words that add little value. Finally, we reduce words to their base form. "Jumping" becomes "jump," and "foxes" becomes "fox." This last step can be done through stemming, which is a crude cut, or lemmatization, which is smarter and uses a dictionary to find the root word.

Here is a class I often use to handle this entire process. It lets me choose which steps to apply.

import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from nltk.stem import PorterStemmer, WordNetLemmatizer
import string
import spacy

# A one-time download for the necessary data files
nltk.download('punkt')
nltk.download('stopwords')
nltk.download('wordnet')

class TextPreprocessor:
    def __init__(self, use_spacy=False):
        # Load common English words to filter out later
        self.stop_words = set(stopwords.words('english'))
        # Tools for reducing words to stems or roots
        self.stemmer = PorterStemmer()
        self.lemmatizer = WordNetLemmatizer()
        self.use_spacy = use_spacy
        # spaCy is a powerful industrial-strength library
        if use_spacy:
            # You'd need to run: python -m spacy download en_core_web_sm
            self.nlp = spacy.load('en_core_web_sm')

    def clean_text(self, text):
        # Standardize case
        text = text.lower()
        # Strip out punctuation characters
        text = text.translate(str.maketrans('', '', string.punctuation))
        # Remove digits - sometimes you keep them, sometimes not.
        text = ''.join([char for char in text if not char.isdigit()])
        return text

    def tokenize(self, text):
        # Use spaCy's accurate tokenizer if requested
        if self.use_spacy:
            doc = self.nlp(text)
            return [token.text for token in doc]
        else:
            # Use NLTK's standard tokenizer
            return word_tokenize(text)

    def remove_stopwords(self, token_list):
        # Filter out any token that's in our stop words list
        return [token for token in token_list if token not in self.stop_words]

    def stem_tokens(self, token_list):
        # Apply stemming: "jumping" -> "jump", "running" -> "run"
        return [self.stemmer.stem(token) for token in token_list]

    def lemmatize_tokens(self, token_list):
        # Apply lemmatization: "better" -> "good", "mice" -> "mouse"
        return [self.lemmatizer.lemmatize(token) for token in token_list]

    def preprocess(self, text, steps=['clean', 'tokenize', 'remove_stopwords', 'lemmatize']):
        # A pipeline to run the text through selected steps
        processed_text = text
        if 'clean' in steps:
            processed_text = self.clean_text(processed_text)

        tokens = self.tokenize(processed_text)

        if 'remove_stopwords' in steps:
            tokens = self.remove_stopwords(tokens)
        if 'stem' in steps:
            tokens = self.stem_tokens(tokens)
        if 'lemmatize' in steps:
            tokens = self.lemmatize_tokens(tokens)

        return tokens

# Let's see it in action
print("=== Example 1: Basic NLTK ===")
basic_preprocessor = TextPreprocessor(use_spacy=False)
sample_text = "The quick brown foxes are jumping over the lazy dogs. They're having fun!"
result_tokens = basic_preprocessor.preprocess(sample_text)
print(f"Original: {sample_text}")
print(f"Cleaned Tokens: {result_tokens}")
print()

print("=== Example 2: Using spaCy ===")
# spaCy is great for accuracy and linguistic features
spacy_preprocessor = TextPreprocessor(use_spacy=True)
# Let's skip the clean step to let spaCy handle punctuation its own way for demonstration
spacy_tokens = spacy_preprocessor.preprocess(sample_text, steps=['tokenize', 'lemmatize'])
print(f"spaCy Lemmatized Tokens: {spacy_tokens}")
print()

print("=== Example 3: Trying Stemming ===")
# Compare stemming vs lemmatization on a tricky sentence
test_sentence = "I believed the stories about the running mice and the better worlds."
stem_result = basic_preprocessor.preprocess(test_sentence, steps=['clean', 'tokenize', 'stem'])
lemma_result = basic_preprocessor.preprocess(test_sentence, steps=['clean', 'tokenize', 'lemmatize'])
print(f"Test Sentence: {test_sentence}")
print(f"Stemmed: {stem_result}")
print(f"Lemmatized: {lemma_result}")

Running this code shows the transformation. The first example gives us ['quick', 'brown', 'fox', 'jump', 'lazy', 'dog', 'fun']. The spaCy example might keep "dogs" as "dog" and "having" as "have" more reliably. The third example clearly shows the difference: stemming turns "believed" to "believ" and "stories" to "stori", while lemmatization correctly gets "believe" and "story". Choosing between them depends on your task; speed favors stemming, meaning favors lemmatization.

Once your text is clean, the next question is how to represent it for a machine. Machines don't understand words; they understand numbers. My second technique is about creating a simple but powerful numerical representation called Bag of Words.

Imagine you take all the unique words from a set of documents and throw them into a bag. For each document, you then create a vector—a list of numbers—that counts how many times each word from the bag appears in it. The order of the words is lost, hence the name "bag." It's simple but surprisingly effective for tasks like spam detection or topic categorization.

Let's build one from scratch to see how it works, then use a professional tool.

from collections import Counter
import pandas as pd

# Our simple corpus: three tiny documents
documents = [
    "The cat sat on the mat.",
    "The dog chased the cat.",
    "The cat and the dog are friends."
]

def simple_bag_of_words(docs):
    """
    Creates a basic Bag of Words model manually.
    """
    # Step 1: Preprocess each document (using our earlier class)
    prep = TextPreprocessor(use_spacy=False)
    processed_docs = []
    for doc in docs:
        tokens = prep.preprocess(doc, steps=['clean', 'tokenize', 'remove_stopwords'])
        processed_docs.append(tokens)

    # Step 2: Build the vocabulary - the set of all unique words
    vocabulary = set()
    for doc_tokens in processed_docs:
        vocabulary.update(doc_tokens)
    vocabulary = sorted(list(vocabulary)) # Sort for consistency
    print(f"Our Vocabulary: {vocabulary}")

    # Step 3: Create vectors by counting word occurrences
    vectors = []
    for doc_tokens in processed_docs:
        word_counts = Counter(doc_tokens)
        # Create a vector with a count for each word in the vocabulary
        vector = [word_counts.get(word, 0) for word in vocabulary]
        vectors.append(vector)

    return vocabulary, vectors

vocab, bow_vectors = simple_bag_of_words(documents)
print("\nManual Bag of Words Vectors:")
for i, vec in enumerate(bow_vectors):
    print(f"Doc {i+1}: {vec}")

# Now, let's do it the easy, professional way with scikit-learn
print("\n" + "="*50 + "\n")
print("Using scikit-learn's CountVectorizer:\n")

from sklearn.feature_extraction.text import CountVectorizer

# Re-create the documents (without preprocessing, CountVectorizer will do it)
raw_docs = [
    "The cat sat on the mat.",
    "The dog chased the cat.",
    "The cat and the dog are friends."
]

# Initialize the vectorizer. We can pass parameters to mimic our preprocessing.
vectorizer = CountVectorizer(lowercase=True, stop_words='english')
# 'fit_transform' learns the vocabulary and creates the vectors in one step
X = vectorizer.fit_transform(raw_docs)

# Convert the result to a readable array
bow_array = X.toarray()
feature_names = vectorizer.get_feature_names_out()

print("Vocabulary (Feature Names):", feature_names)
print("\nDocument-Term Matrix:")
# Use pandas for a nice table view
df = pd.DataFrame(bow_array, columns=feature_names, index=[f"Doc {i+1}" for i in range(len(raw_docs))])
print(df)

The manual function shows the process: we get a vocabulary like ['cat', 'chased', 'dog', 'friends', 'mat', 'sat']. Document 1 "The cat sat on the mat." becomes the vector [1, 0, 0, 0, 1, 1], showing counts for 'cat', 'mat', and 'sat'. The scikit-learn output presents this as a neat table, which is much more practical for real work.

The Bag of Words model has a flaw. Common words like "the" or "is" will have very high counts across all documents, drowning out rarer, more interesting words. This leads me to the third technique: TF-IDF, or Term Frequency-Inverse Document Frequency.

TF-IDF fixes the common-word problem by adjusting the weight of a word. It considers not just how often a word appears in a document (Term Frequency), but also how unique that word is across all documents (Inverse Document Frequency). A word that appears in every document gets a low score. A word that appears many times in one document but rarely elsewhere gets a very high score. This makes TF-IDF excellent for search engines and keyword extraction.

Here’s how you calculate and use it.

from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

# Same sample documents
documents = [
    "The cat sat on the mat.",
    "The dog chased the cat.",
    "The cat and the dog are friends."
]

# Initialize the TF-IDF Vectorizer
tfidf_vectorizer = TfidfVectorizer(lowercase=True, stop_words='english')

# Learn vocabulary and transform documents into TF-IDF features
tfidf_matrix = tfidf_vectorizer.fit_transform(documents)

# Get feature names and the dense matrix
feature_names = tfidf_vectorizer.get_feature_names_out()
dense_matrix = tfidf_matrix.toarray()

print("TF-IDF Feature Names:", feature_names)
print("\nTF-IDF Matrix:")
tfidf_df = pd.DataFrame(np.round(dense_matrix, 3), columns=feature_names, index=[f"Doc {i+1}" for i in range(len(documents))])
print(tfidf_df)

# Let's interpret one score manually for Doc 1, word 'cat'
print("\n--- Manual Check for 'cat' in Document 1 ---")
# Term Frequency (TF) for 'cat' in Doc 1: count of 'cat' / total words in doc
doc1_words = ["cat", "sat", "mat"] # after preprocessing
tf_cat = doc1_words.count('cat') / len(doc1_words)
print(f"TF('cat' in Doc1) = {doc1_words.count('cat')} / {len(doc1_words)} = {tf_cat:.3f}")

# Inverse Document Frequency (IDF) for 'cat'
# Number of documents
N = len(documents)
# Number of documents containing 'cat'
docs_with_cat = sum(1 for doc in documents if 'cat' in doc.lower())
# A common formula for IDF is log( (N+1) / (docs_with_word + 1) ) + 1 (sklearn uses a smoothed version)
idf_cat = np.log((N + 1) / (docs_with_cat + 1)) + 1
print(f"IDF('cat') = log(({N}+1)/({docs_with_cat}+1)) + 1 = {idf_cat:.3f}")

# TF-IDF
manual_tfidf_cat = tf_cat * idf_cat
print(f"Manual TF-IDF('cat', Doc1) ≈ {manual_tfidf_cat:.3f}")
print(f"scikit-learn's value for 'cat' in Doc1: {dense_matrix[0][list(feature_names).index('cat')]:.3f}")

The TF-IDF table will show scores between 0 and 1. You'll notice that "cat" and "dog" have decent scores because they are common but not in every document. A word like "chased," which is unique to document 2, will have a very high score for that document, highlighting its importance as a distinguishing keyword.

These vector representations—Bag of Words and TF-IDF—turn text into a grid of numbers. This grid is often called a document-term matrix. It's the bridge that allows us to apply all the standard machine learning algorithms we use for numerical data to the world of text.

Working with individual words is powerful, but language is about sequences and context. My fourth technique explores capturing word relationships through N-grams.

An N-gram is simply a contiguous sequence of N items from a text. If N=1, it's a unigram (single words). If N=2, it's a bigram (pairs of words), and N=3 is a trigram. The phrase "the quick brown fox" has unigrams: ["the", "quick", "brown", "fox"]; bigrams: ["the quick", "quick brown", "brown fox"]; trigrams: ["the quick brown", "quick brown fox"].

Bigrams and trigrams help us capture phrases and idioms. "New York" as a bigram has a meaning different from just "new" and "york" separately. "Not good" conveys the opposite sentiment of "good". By adding N-grams to our Bag of Words or TF-IDF model, we give our machine a better sense of context.

Let's see how to generate them and add them to our vectorizer.

from nltk import ngrams
from sklearn.feature_extraction.text import CountVectorizer

sample_sentence = "natural language processing is fascinating and natural learning is key"
tokens = word_tokenize(sample_sentence.lower())

print("Original Tokens:", tokens)
print()

# Generate bigrams and trigrams using NLTK
print("Generating N-grams with NLTK:")
bigrams = list(ngrams(tokens, 2))
trigrams = list(ngrams(tokens, 3))
print(f"Bigrams: {[' '.join(bg) for bg in bigrams]}")
print(f"Trigrams: {[' '.join(tg) for tg in trigrams]}")
print()

# The practical way: Use CountVectorizer with an ngram_range parameter
print("Using CountVectorizer with N-gram ranges:\n")

corpus = [
    "I love natural language processing.",
    "I hate boring language tutorials.",
    "Natural learning is the best processing."
]

# This vectorizer will create features for unigrams AND bigrams
vectorizer = CountVectorizer(ngram_range=(1, 2), stop_words='english')
X_ngram = vectorizer.fit_transform(corpus)
ngram_features = vectorizer.get_feature_names_out()
ngram_df = pd.DataFrame(X_ngram.toarray(), columns=ngram_features, index=[f"Text {i+1}" for i in range(3)])

print("Vocabulary includes unigrams and bigrams:")
print(ngram_features)
print("\nDocument-Term Matrix with N-grams:")
print(ngram_df)

# Let's check a specific phrase
print("\n--- Checking for the phrase 'natural language' ---")
if 'natural language' in ngram_features:
    col_index = list(ngram_features).index('natural language')
    print(f"'natural language' is feature #{col_index}")
    print(f"Counts per document: {ngram_df.iloc[:, col_index].values}")
else:
    print("Phrase not found as a bigram feature.")

The output shows our vocabulary now includes single words and pairs. You'll see features like love natural and natural language. In the matrix, you can see that "natural language" appears once in the first document. This simple addition can dramatically improve model performance for tasks like sentiment analysis, where "not happy" is critical to catch.

So far, we've treated words as independent symbols. But words have relationships. "King" is to "man" as "queen" is to "woman." The fifth technique, Word Embeddings, captures these semantic meanings by representing each word as a dense vector in a high-dimensional space, where similar words are close together.

Tools like Word2Vec, GloVe, and FastText create these embeddings by learning from massive amounts of text. The word "python" might be represented by a 300-dimensional vector like [0.25, -0.1, 0.87, ...]. We can then do math with words: vector('king') - vector('man') + vector('woman') results in a vector very close to vector('queen').

Let's load pre-trained GloVe embeddings and explore these relationships.

import numpy as np
from scipy.spatial.distance import cosine

# In practice, you download a GloVe file (e.g., 'glove.6B.100d.txt')
# Let's simulate a tiny embedding space for demonstration
print("=== Simulating a Mini Word Embedding Space ===")

# A tiny dictionary of word vectors (3 dimensions for simplicity)
embedding_index = {
    'king':    np.array([0.8, 0.2, 0.1]),
    'man':     np.array([0.6, 0.3, 0.1]),
    'queen':   np.array([0.7, 0.9, 0.1]),
    'woman':   np.array([0.5, 0.8, 0.1]),
    'python':  np.array([0.1, 0.1, 0.9]),
    'java':    np.array([0.15, 0.1, 0.85]),
    'london':  np.array([0.9, 0.0, 0.0]),
    'paris':   np.array([0.88, 0.05, 0.01]),
}

def find_analogy(word_a, word_b, word_c, embeddings):
    """Finds word_d such that a:b :: c:d."""
    try:
        vec_a = embeddings[word_a]
        vec_b = embeddings[word_b]
        vec_c = embeddings[word_c]
        # The analogy calculation
        vec_d = vec_b - vec_a + vec_c

        # Find the word in our vocab whose vector is closest to vec_d
        best_word = None
        best_sim = -1 # Cosine similarity ranges from -1 to 1
        for word, vec in embeddings.items():
            if word in [word_a, word_b, word_c]:
                continue # Skip the input words
            # Cosine similarity: 1 means identical, 0 means orthogonal
            sim = 1 - cosine(vec_d, vec)
            if sim > best_sim:
                best_sim = sim
                best_word = word
        return best_word, best_sim
    except KeyError as e:
        return f"Word not in vocabulary: {e}", None

# Try the classic analogy
result, sim = find_analogy('king', 'man', 'woman', embedding_index)
print(f"'king' is to 'man' as 'woman' is to '{result}' (similarity: {sim:.3f})")

# Check similarity between related and unrelated words
print("\n--- Cosine Similarities ---")
def print_sim(word1, word2, emb):
    sim = 1 - cosine(emb[word1], emb[word2])
    print(f"cosine_sim('{word1}', '{word2}') = {sim:.3f}")

print_sim('king', 'queen', embedding_index)
print_sim('man', 'woman', embedding_index)
print_sim('python', 'java', embedding_index)
print_sim('king', 'python', embedding_index)
print_sim('london', 'paris', embedding_index)

# Practical use: Convert a sentence to an embedding by averaging its word vectors
print("\n--- Sentence Embedding by Averaging ---")
sentence = "king and queen in london"
words = [w for w in sentence.lower().split() if w in embedding_index]
if words:
    sentence_vector = np.mean([embedding_index[w] for w in words], axis=0)
    print(f"Sentence: '{sentence}'")
    print(f"Words found: {words}")
    print(f"Averaged Sentence Vector: {np.round(sentence_vector, 3)}")

In our mini example, the analogy function should suggest "queen." The similarities show high scores for related pairs (king/queen, london/paris) and a lower score for unrelated ones (king/python). Averaging word vectors to get a sentence vector is a common, simple way to represent longer text with embeddings.

For real projects, you don't simulate. You load a large pre-trained file. Here's a template for using real GloVe embeddings with Gensim.

# Template for using real GloVe embeddings with Gensim
print("\n=== Template for Real GloVe Embeddings (Requires File Download) ===")
"""
# First, convert GloVe format to Word2Vec format for Gensim
from gensim.scripts.glove2word2vec import glove2word2vec
glove_input_file = 'glove.6B.100d.txt'
word2vec_output_file = 'glove.6B.100d.word2vec.txt'
glove2word2vec(glove_input_file, word2vec_output_file)

# Then load and use
from gensim.models import KeyedVectors
model = KeyedVectors.load_word2vec_format(word2vec_output_file, binary=False)

# Now you can use it
similar_words = model.most_similar('python', topn=5)
print(similar_words)  # Might show: [('java', 0.85), ('code', 0.82), ...]

result = model.most_similar(positive=['woman', 'king'], negative=['man'], topn=1)
print(result)  # Hopefully: [('queen', 0.79...)]
"""
print("(Code commented out as it requires the large GloVe file to be downloaded.)")

Word embeddings give us semantic understanding. The sixth technique builds on this to classify the overall feeling of a text: Sentiment Analysis. Is a product review positive or negative? Is a tweet angry or joyful?

We can approach this with simple rule-based methods using sentiment lexicons (dictionaries of words with pre-assigned scores) or with machine learning models trained on labeled data.

Let's try both. First, a lexicon-based approach with the VADER tool, which is great for social media text. Then, a machine learning approach using our TF-IDF vectors.

# Technique 6: Sentiment Analysis
from nltk.sentiment.vader import SentimentIntensityAnalyzer
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report
import nltk
nltk.download('vader_lexicon')

print("=== Part A: Lexicon-Based with VADER ===")
sid = SentimentIntensityAnalyzer()

sentences = [
    "This movie is absolutely fantastic! I loved every minute of it.",
    "Waste of time. The plot was terrible and the acting was worse.",
    "It was okay. Not great, not terrible.",
    "I'm so excited for the concert!!!",
    "This product broke after two days. Very disappointed."
]

for sentence in sentences:
    scores = sid.polarity_scores(sentence)
    # The 'compound' score aggregates pos, neg, neu into a single value from -1 (neg) to +1 (pos)
    compound = scores['compound']
    sentiment = "positive" if compound >= 0.05 else "negative" if compound <= -0.05 else "neutral"
    print(f"Text: {sentence[:50]}...")
    print(f"  Scores: {scores}")
    print(f"  Judgement: {sentiment} (based on compound: {compound:.3f})\n")

print("="*60)
print("=== Part B: Machine Learning-Based Sentiment ===")
# Let's create a small simulated dataset
print("\n1. Creating a simple dataset...")
# Positive sentences
pos_texts = [
    "I love this product it is amazing",
    "Great service and fast delivery",
    "Excellent quality highly recommend",
    "Very happy with my purchase",
    "Perfect exactly what I wanted"
]
# Negative sentences
neg_texts = [
    "Terrible experience would not buy again",
    "Poor quality broke immediately",
    "Waste of money completely useless",
    "Very disappointed with the service",
    "Bad product do not recommend"
]

texts = pos_texts + neg_texts
# Labels: 1 for positive, 0 for negative
labels = [1]*len(pos_texts) + [0]*len(neg_texts)

print(f"We have {len(texts)} total texts: {len(pos_texts)} positive, {len(neg_texts)} negative.")

print("\n2. Converting text to TF-IDF features...")
tfidf = TfidfVectorizer()
X = tfidf.fit_transform(texts)
print(f"Feature matrix shape: {X.shape}") # (10 documents, X unique words)

print("\n3. Training a simple classifier...")
X_train, X_test, y_train, y_test = train_test_split(X, labels, test_size=0.3, random_state=42)
model = LogisticRegression()
model.fit(X_train, y_train)

print("\n4. Making predictions and evaluating...")
y_pred = model.predict(X_test)
print(f"True labels for test set: {y_test}")
print(f"Model predictions:        {y_pred}")
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy on test set: {accuracy:.2f}")

print("\n5. Trying the model on new sentences...")
new_sentences = [
    "This is good and works well",
    "I hate it it is awful",
    "It is okay I guess"
]
new_X = tfidf.transform(new_sentences)
new_preds = model.predict(new_X)
sentiment_map = {1: 'POSITIVE', 0: 'NEGATIVE'}
for sent, pred in zip(new_sentences, new_preds):
    print(f"'{sent}' -> {sentiment_map[pred]}")

VADER gives us a detailed breakdown for each sentence, including positive, negative, neutral, and compound scores. The machine learning model, though trained on tiny data, learns the pattern. It correctly classifies our new examples. In reality, you would train on thousands of labeled reviews for a robust system.

Sometimes we need to go beyond words and sentences to understand the structure of a document. My seventh technique is Topic Modeling, specifically Latent Dirichlet Allocation (LDA). It's a way to discover hidden thematic patterns in a large collection of documents. You don't tell it the topics; it figures them out by grouping words that frequently appear together.

If you feed it 10,000 news articles, it might output topics characterized by words like ['election', 'vote', 'candidate', 'poll'] (Politics), ['stock', 'market', 'price', 'trade'] (Finance), and ['player', 'game', 'team', 'score'] (Sports). Each document is then seen as a mixture of these topics.

Let's apply LDA to a small set of fake news headlines to see it in action.

# Technique 7: Topic Modeling with LDA
from sklearn.decomposition import LatentDirichletAllocation
from sklearn.feature_extraction.text import CountVectorizer

print("=== Discovering Topics with LDA ===")

# A small corpus of fake headlines
corpus = [
    "stock market reaches all time high today",
    "investors celebrate rising share prices",
    "new election poll shows close race",
    "candidates debate economic policy",
    "football team wins championship final",
    "basketball player scores record points",
    "central bank raises interest rates",
    "trading volume surges on wall street",
    "voters head to the polls tomorrow",
    "sports fans celebrate victory parade"
]

print("Corpus of headlines:")
for i, h in enumerate(corpus):
    print(f"  {i+1:2d}. {h}")

print("\n1. Creating a Bag of Words model...")
# We don't remove stop words here, as they might be part of topic structure.
# But for clean topics, we often do.
vectorizer = CountVectorizer(max_df=0.95, min_df=2, stop_words='english')
doc_term_matrix = vectorizer.fit_transform(corpus)
feature_names = vectorizer.get_feature_names_out()

print(f"Document-Term Matrix shape: {doc_term_matrix.shape}")
print(f"Number of unique words (features): {len(feature_names)}")

print("\n2. Fitting the LDA model to find 3 topics...")
num_topics = 3
lda_model = LatentDirichletAllocation(n_components=num_topics, random_state=42, max_iter=10)
lda_model.fit(doc_term_matrix)

print("\n3. Displaying the top words for each discovered topic...")
def display_topics(model, feature_names, no_top_words=5):
    topics = {}
    for topic_idx, topic in enumerate(model.components_):
        # Get indices of the top words for this topic
        top_indices = topic.argsort()[:-no_top_words - 1:-1]
        top_words = [feature_names[i] for i in top_indices]
        topics[f"Topic_{topic_idx}"] = top_words
    return topics

topics = display_topics(lda_model, feature_names)
for topic_id, words in topics.items():
    print(f"{topic_id}: {', '.join(words)}")

print("\n4. Checking the topic mixture for the first few documents...")
# Transform the documents to see their topic distribution
doc_topic_dist = lda_model.transform(doc_term_matrix)
print("Document -> Topic Distribution (percentages):")
for i, dist in enumerate(doc_topic_dist[:5]): # First 5 docs
    print(f"Doc {i+1} ('{corpus[i][:30]}...'):")
    for t_idx, percent in enumerate(dist):
        print(f"  Topic {t_idx}: {percent*100:5.1f}%")
    print()

print("\n5. Assigning a dominant topic to each document...")
dominant_topics = doc_topic_dist.argmax(axis=1)
topic_labels = {0: "FINANCE?", 1: "POLITICS?", 2: "SPORTS?"} # Our interpretation
for i, (doc, dom_topic) in enumerate(zip(corpus, dominant_topics)):
    print(f"'{doc}'")
    print(f"  -> Dominant Topic: {dom_topic} ({topic_labels[dom_topic]})\n")

The output will show three lists of words. One might be ['market', 'stock', 'prices', 'share', 'high'] (Finance), another ['election', 'poll', 'candidates', 'voters', 'polls'] (Politics), and a third ['wins', 'football', 'team', 'scores', 'player'] (Sports). The document-topic distribution shows that the first headline is 90% Finance, 5% Politics, 5% Sports. LDA has successfully uncovered the latent themes.

Our final, eighth technique brings us to the current frontier: using deep learning models designed for sequences, like Recurrent Neural Networks (RNNs) and Transformers. While the previous techniques are foundational, models like LSTMs (a type of RNN) and BERT (a Transformer) handle context and long-range dependencies in text far more effectively.

An LSTM has a kind of memory, allowing it to consider previous words when processing the next one, which is ideal for text generation or classification where word order matters deeply. BERT, which stands for Bidirectional Encoder Representations from Transformers, reads the entire sentence at once from both directions, leading to a profound understanding of context. The word "bank" in "river bank" gets a different embedding than in "bank deposit."

Let's implement a simple sentiment classifier using an LSTM with Keras to see how a neural network approaches the problem.

# Technique 8: Deep Learning for Text (LSTM Example)
import numpy as np
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Embedding, LSTM, Dense, Dropout
from sklearn.model_selection import train_test_split

print("=== Building an LSTM for Text Sentiment ===")

# We'll use a slightly larger simulated dataset
print("1. Preparing the data...")
positive_statements = [
    "this is wonderful and great",
    "I feel amazing and happy today",
    "what a fantastic and perfect result",
    "we are thrilled with the excellent outcome",
    "so good and positive experience",
    "love it absolutely brilliant",
    "outstanding performance very impressed",
    "superb quality highly satisfied"
]
negative_statements = [
    "this is terrible and awful",
    "I feel sad and angry today",
    "what a horrible and disappointing result",
    "we are upset with the poor outcome",
    "so bad and negative experience",
    "hate it absolutely dreadful",
    "unacceptable performance very disappointed",
    "low quality highly dissatisfied"
]

texts = positive_statements + negative_statements
labels = [1] * len(positive_statements) + [0] * len(negative_statements) # 1=pos, 0=neg

print(f"Total texts: {len(texts)}")
print(f"Sample positive: '{positive_statements[0]}'")
print(f"Sample negative: '{negative_statements[0]}'")

print("\n2. Tokenizing the text and creating sequences...")
tokenizer = Tokenizer(num_words=100, oov_token="<OOV>")
tokenizer.fit_on_texts(texts)
word_index = tokenizer.word_index
print(f"Vocabulary size: {len(word_index)}")
print(f"Word index sample: {list(word_index.items())[:5]}")

sequences = tokenizer.texts_to_sequences(texts)
print(f"\nFirst text as sequence: {sequences[0]}")
print(f"Corresponding words: {[list(word_index.keys())[list(word_index.values()).index(idx)] for idx in sequences[0] if idx in word_index.values()]}")

# Pad sequences to ensure uniform length
max_length = 8
padded_sequences = pad_sequences(sequences, maxlen=max_length, padding='post', truncating='post')
print(f"\nPadded sequences shape: {padded_sequences.shape}")
print(f"First padded sequence: {padded_sequences[0]}")

print("\n3. Building the LSTM model...")
model = Sequential([
    # Embedding layer: turns word indices into dense vectors
    Embedding(input_dim=100, output_dim=16, input_length=max_length),
    # LSTM layer with 32 memory units
    LSTM(32, dropout=0.2),
    # Dense output layer with sigmoid activation for binary classification
    Dense(1, activation='sigmoid')
])

model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
model.summary()

print("\n4. Training the model...")
X_train, X_val, y_train, y_val = train_test_split(padded_sequences, labels, test_size=0.2, random_state=42)
history = model.fit(X_train, np.array(y_train), epochs=20, validation_data=(X_val, np.array(y_val)), verbose=0)

print("Training complete. Checking final accuracy...")
train_loss, train_acc = model.evaluate(X_train, np.array(y_train), verbose=0)
val_loss, val_acc = model.evaluate(X_val, np.array(y_val), verbose=0)
print(f"Training Accuracy:   {train_acc:.3f}")
print(f"Validation Accuracy: {val_acc:.3f}")

print("\n5. Testing on new sentences...")
test_sentences = [
    "this is good and happy",
    "this is bad and sad",
    "I am feeling wonderful",
    "what a terrible day"
]
test_seq = tokenizer.texts_to_sequences(test_sentences)
test_padded = pad_sequences(test_seq, maxlen=max_length, padding='post', truncating='post')
predictions = model.predict(test_padded)

for sent, pred in zip(test_sentences, predictions):
    sentiment = "POSITIVE" if pred > 0.5 else "NEGATIVE"
    print(f"'{sent}' -> {sentiment} (confidence: {pred[0]:.3f})")

This simple LSTM learns to map sequences of word indices to a sentiment score. The Embedding layer starts with random vectors and learns meaningful ones during training. The LSTM layer processes the sequence, and the final dense layer makes the decision. You'll see it correctly classifies our simple test sentences. For real-world use, you would need much more data, deeper networks, and perhaps pre-trained word embeddings in the first layer.

These eight techniques form a pathway from raw text to intelligent understanding. We start by cleaning the text, then represent it as numbers (Bag of Words, TF-IDF), capture phrases (N-grams), infuse meaning (Word Embeddings), judge feeling (Sentiment Analysis), discover themes (Topic Modeling), and finally employ advanced learning (Deep Learning). Each method has its place. Sometimes a simple TF-IDF with a logistic regression is all you need. Other problems demand the contextual power of an LSTM or a Transformer.

The key is to start simple, understand your data, and iteratively choose more complex tools only when they provide a clear benefit. I often begin with a quick TF-IDF analysis to establish a baseline. This entire process, from messy sentences to a model that can gauge emotion or extract topics, is what makes working with language in Python so engaging. You're not just manipulating strings; you're building a lens to focus on the meaning within.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

# Rust and Formal Verification: How to Prove Your Code Actually Works

Nithin Bharadwaj — Tue, 21 Apr 2026 13:31:39 +0000

Let's talk about making software that doesn't just seem to work, but that we can prove works correctly. This is the world of formal verification. It can sound intimidating, like something reserved for rocket scientists and academic papers. But what if you could get much closer to that level of certainty using a language you already write for practical projects? That's where Rust comes in.

Think of it this way. Testing is like checking the lights in your house. You flip a switch and see if the kitchen light turns on. You test a few rooms. It looks good. Formal verification is like having the complete electrical blueprint. You can trace every wire from the fuse box to every bulb and switch. You can prove, on paper, that the kitchen light will turn on when you flip its specific switch, and only then. No hidden crossed wires, no chance of the toilet flushing turning off your TV. Rust helps us write those blueprints into the code itself.

I used to think of program crashes or bugs as inevitable, just things you hunt down with tests. Then I started with Rust and experienced something different. The compiler became my first and most stubborn reviewer. It wouldn't let me make a mess with memory. It was frustrating at first, but then it clicked. This wasn't just nagging; it was proving certain bad things could not happen in my program. That’s a powerful feeling.

So, how does a practical language like Rust do this? It starts with its core rules: ownership and borrowing. These aren't just fancy features; they are mathematical laws baked into the syntax. The compiler uses these laws to prove your code is free from memory bugs and data races. It's a built-in verification step for some of the most common and dangerous errors in software.

Let's look at a simple piece of code. We want a function that takes a reference to a number and adds one to it.

fn add_one(number: &mut i32) {
    *number += 1;
}

fn main() {
    let mut my_number = 5;
    add_one(&mut my_number);
    println!("My number is: {}", my_number); // Prints 6
}

This seems straightforward. But Rust is proving things here. The &mut i32 type is a "mutable reference." The ownership rules prove that while this reference exists, it is the only way to access my_number. No other part of my code can read or write my_number while add_one is running. This proves the operation is safe from certain kinds of interference. In other languages, you just hope that's true.

We can encode more sophisticated proofs using Rust's type system. Let's say we're writing software for a smart home. A light bulb can be either On or Off. A naive approach uses a boolean: true for on, false for off. But what does true mean? Could it get confused with something else? Let's use types to prove the state is always valid.

struct LightBulb {
    brightness: u8,
}

struct On;
struct Off;

impl LightBulb {
    fn turn_on(self) -> (LightBulb, On) {
        // Turning on sets a default brightness.
        let bulb = LightBulb { brightness: 100 };
        (bulb, On)
    }

    fn turn_off(self) -> (LightBulb, Off) {
        // Turning off might reset brightness.
        let bulb = LightBulb { brightness: 0 };
        (bulb, Off)
    }

    fn adjust_brightness(self, level: u8, _state: On) -> (LightBulb, On) {
        // You can only call this if you prove you have an `On` state token.
        let bulb = LightBulb { brightness: level.clamp(10, 100) };
        (bulb, On)
    }
}

fn main() {
    let bulb = LightBulb { brightness: 0 };
    let (bulb, state) = bulb.turn_on();
    // Now `state` is of type `On`.
    let (bulb, _) = bulb.adjust_brightness(75, state); // This works.
    // let (bulb, _) = bulb.adjust_brightness(75, state); // ERROR! `state` was used up above.
}

See what happened? The function adjust_brightness takes a parameter _state of type On. Not a boolean, but the actual type On. To call this function, you must possess an On value. You only get that value from turn_on. Furthermore, because we use the value (we take ownership of it), you can't call adjust_brightness twice with the same proof. The type system proves the sequence of operations: you turned it on, then you adjusted it. You cannot adjust a lightbulb that is off. This logic is checked at compile time.

This pattern is incredibly useful for security and safety. Imagine a network socket. You can't send data on a socket that isn't connected. You can't close a socket that's already closed. You can design your API so that the only way to get a "Send" capability is to first go through a "Connect" function that returns a connected socket type. The compiler verifies the correct workflow.

Where does this matter in the real world? Everywhere correctness is non-negotiable.

Cryptography: A classic vulnerability is a "timing attack." If checking a password takes a slightly different amount of time when the first character is wrong versus the last, an attacker can guess the password. A cryptographic library must prove its operations take constant time, regardless of the data. Rust's focus on explicit control and lack of hidden runtime checks helps. You can write code that performs the exact same sequence of instructions every time, and the compiler won't insert a secret branch that breaks the guarantee.

Financial Systems: Consider a function that transfers money between accounts. A critical property is "conservation of money." The total money in the system before and after the transfer must be the same; it just moves. You can structure your code to make this provable.

struct Money(u64);

struct Accounts {
    alice: Money,
    bob: Money,
}

impl Accounts {
    fn transfer(&mut self, amount: u64) -> Result<(), &'static str> {
        if self.alice.0 < amount {
            return Err("Alice has insufficient funds");
        }
        // This operation is critical. It must be atomic.
        self.alice.0 -= amount;
        self.bob.0 += amount;
        Ok(())
    }
    // A property we want: alice + bob is always the same.
    fn total(&self) -> u64 {
        self.alice.0 + self.bob.0
    }
}

While this simple code can't fully prove conservation at compile-time (Alice's balance could change elsewhere in a more complex system), Rust's ownership ensures that if we have a single &mut Accounts reference, no other code can change alice or bob during transfer. This isolates the critical block, making reasoning about it much easier. External tools, which we'll discuss, can then take this isolated block and prove the conservation property.

Embedded and Safety-Critical Systems: This is a major area. The Tock operating system for microcontrollers uses Rust's type system to build a secure kernel. Different driver capsules in the kernel are isolated from each other not by hardware, but by Rust's ownership. The compiler proves that a USB driver cannot suddenly access the memory of the temperature sensor. This creates a trusted computing base with far fewer lines of code that need to be manually audited.

Now, Rust's built-in system is powerful, but it's not the end of the story. The ecosystem is building tools that bring full-scale formal methods to Rust code.

One exciting project is Kani, a model checker from AWS. A model checker explores every possible execution path of your code, given certain limits. You give it invariants—things that must always be true—and it tries to find a path that breaks them.

Let's say we have a function that is supposed to always return a positive number.

fn absolute_value(x: i32) -> u32 {
    if x >= 0 {
        x as u32
    } else {
        (-x) as u32
    }
}

This looks correct. But does it handle the edge case where x is i32::MIN? The absolute value of -2147483648 is 2147483648, which doesn't fit in an i32. Our function casts -x to u32. When x is i32::MIN, -x overflows in a panic in debug mode, or wraps in release mode. This is a bug. Kani can find this.

You would write a verification harness:

#[cfg(kani)]
#[kani::proof]
fn verify_absolute_non_negative() {
    let x: i32 = kani::any();
    let result = absolute_value(x);
    // Our invariant: The result is always the absolute value.
    // We'll check a simpler property: result is not crazy large.
    assert!(result <= i32::MAX as u32);
}

When you run Kani, it doesn't use random numbers. It treats x as "any possible i32 value." It will systematically find that when x is i32::MIN, the -x operation panics, and it will report this as a failure. You've just formally proven the function is not correct for all inputs.

Another tool is Prusti. It sits between the compiler and full formal proof. You write annotations in your code about what your functions require (pre-conditions) and what they guarantee (post-conditions). Prusti then tries to prove these hold.

use prusti_contracts::*;

#[requires(x >= 0)] // This function requires x to be non-negative.
#[ensures(result == x * 2)] // It guarantees the result is twice x.
fn double_positive(x: i32) -> i32 {
    x + x // Let's intentionally write a bug: x + x for x=1 is 2, correct.
    // What if we wrote `x * 3` by mistake? Prusti would catch it.
}

These tools feel like a super-powered linter. They don't just check syntax; they check logic.

You might wonder, why not just use a dedicated proof language like Coq or Isabelle? Those are fantastic for what they do. But there's a gap: you prove an algorithm in Coq, then you have to manually translate that correct algorithm into C or Java for your real product. That translation step can introduce bugs. Rust flips this. You write your real product in Rust. Then, you use Rust's own features and tools like Kani or Prusti to verify the actual implementation. You're proving properties about the code that will ship.

This approach is practical. It fits into a normal development workflow. You write code. You write unit tests. You can also write a few key verification harnesses for the most critical properties. You run cargo kani or cargo prusti alongside cargo test. It becomes part of the suite of checks that give you confidence.

I find this changes how I think. I start designing APIs not just for convenience, but for provability. I ask: "What impossible states can I make unrepresentable in the type system?" If a function can fail, I make the success and error types distinct, so the caller must handle both. I use the type system to guide users to correct usage and prove away whole categories of mistakes.

This doesn't replace testing. They work together. Fuzz testing (cargo fuzz) throws random, structured data at your code to find panics. Property-based testing (proptest) lets you say "for all integers x and y, this property should hold" and it tests hundreds of random examples. Formal verification goes further: for all integers x and y, this property must hold, and here is the proof. Testing can find bugs; verification can show their absence.

The journey to verified software in Rust starts small. You don't need to prove your entire web server. Start with a core data structure. Write an Invoice type where the total field is guaranteed to equal the sum of the line_items. Encode it in the types so it can't be constructed incorrectly. Write a Kani harness to prove a critical function never panics. This incremental approach makes formal methods accessible.

Rust brings the dream of verified practical software closer. It turns the compiler from a grammar checker into a reasoning partner. It lets you embed proofs in your code structure. You get stronger guarantees, often with zero runtime cost, because the proof happens when you compile. For anyone building systems where failure has real consequences—whether financial, medical, or human—this isn't just academic. It's the future of reliable software engineering, and it's available now.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

Building a Time-Series Storage Engine in Go That Handles Billions of Data Points

Nithin Bharadwaj — Tue, 21 Apr 2026 13:16:37 +0000

Imagine you're trying to store every temperature reading from a thousand weather stations, each sending data every second. That's 86,400 readings per station daily. After a month, you're looking at billions of numbers. Traditional databases gasp under this load. They weren't built for this constant stream of timestamped data.

This is the problem I set out to solve. I wanted to build a storage engine in Go that could swallow this firehose of data without choking, while still letting me ask complex questions about the past. Think of it as a specialized librarian for the world's most frenetic timeline.

At its heart, time-series data is simple: a timestamp and a value. The challenge is the sheer volume and the pattern of access. You write data points in chronological order, but you often read ranges of time. You care about trends—hourly averages, daily maximums—not just single points.

Let me show you the foundation. We start by defining what a single data point looks like in our Go code.

type DataPoint struct {
    Timestamp int64
    Value     float64
}

This is the atom of our system. The timestamp is usually milliseconds since the Unix epoch, and the value can be any measurement: temperature, stock price, sensor voltage.

The first big idea is to stop storing data in rows. Imagine a spreadsheet where each row is a timestamp and a value. Now, imagine tearing that sheet in two: one column with all the timestamps, another with all the values. This is called columnar storage.

type ColumnBlock struct {
    timestamps  []int64
    values      []float64
    minValue    float64
    maxValue    float64
    compressed  []byte
    uncompressedSize int
}

We group points into blocks. Each block holds a sequence of points, like one hour of data for a specific sensor. Storing timestamps and values separately lets us compress them more effectively because they often follow predictable patterns.

Timestamps are almost always increasing, and often at regular intervals. We can use a trick called delta encoding. Instead of storing the full timestamp for each point, we store the difference from the previous one.

func (tse *TimeSeriesEngine) encodeTimestamps(timestamps []int64) []byte {
    if len(timestamps) == 0 {
        return []byte{}
    }

    encoded := make([]byte, 0, len(timestamps)*8)
    prev := timestamps[0]
    encoded = binary.BigEndian.AppendUint64(encoded, uint64(prev))

    for i := 1; i < len(timestamps); i++ {
        delta := timestamps[i] - prev
        encoded = binary.BigEndian.AppendUint64(encoded, uint64(delta))
        prev = timestamps[i]
    }

    return encoded
}

The first timestamp is stored in full. For the next one, we see it's 1000 milliseconds later, so we just store the number 1000. The next might be another 1000. These deltas are much smaller numbers, and smaller numbers take less space to store, especially when compressed.

Values behave differently. A temperature might be 72.1, then 72.2, then 72.3. The numbers are similar but not identical. For these, we use a method called XOR encoding for floating-point numbers.

func (tse *TimeSeriesEngine) encodeValues(values []float64) []byte {
    if len(values) == 0 {
        return []byte{}
    }

    encoded := make([]byte, 0, len(values)*8)
    prev := math.Float64bits(values[0])
    encoded = binary.BigEndian.AppendUint64(encoded, prev)

    for i := 1; i < len(values); i++ {
        current := math.Float64bits(values[i])
        xor := current ^ prev
        encoded = binary.BigEndian.AppendUint64(encoded, xor)
        prev = current
    }

    return encoded
}

Here's what happens. We take the binary representation of each floating-point number. If two numbers are very close, their binary patterns are similar. When you XOR similar patterns, you get lots of zeroes. Storing a run of zeroes compresses dramatically well.

After this specialized encoding, we apply a general-purpose compression algorithm called S2. It's fast and effective. We end up with a compact blob of bytes representing perhaps ten thousand data points.

Now, where do we put this blob? We need a persistent store. I chose BadgerDB, a key-value store written in Go. It's excellent for fast writes and handles our data model nicely.

type TimeSeriesStorage struct {
    db           *badger.DB
    columnBlocks map[string]*ColumnBlock
    partitions   map[int64]*Partition
    mu           sync.RWMutex
}

We organize data by metric name and by time partition. A partition might be one hour of data. The key in our key-value store looks like temperature:1716940800000, where the number is the start of the hour in milliseconds.

When a new data point arrives, we append it to an in-memory block for that metric and partition. Once the block fills up to a configured size—say, 10,000 points—we compress it and write it to BadgerDB.

func (tse *TimeSeriesEngine) WritePoint(metric string, timestamp int64, value float64) error {
    partitionKey := tse.getPartitionKey(timestamp)
    blockKey := fmt.Sprintf("%s:%d", metric, partitionKey)

    tse.storage.mu.Lock()
    block, exists := tse.storage.columnBlocks[blockKey]
    if !exists {
        block = &ColumnBlock{
            timestamps: make([]int64, 0, tse.config.BlockSize),
            values:     make([]float64, 0, tse.config.BlockSize),
            minValue:   math.MaxFloat64,
            maxValue:   -math.MaxFloat64,
        }
        tse.storage.columnBlocks[blockKey] = block
    }

    block.timestamps = append(block.timestamps, timestamp)
    block.values = append(block.values, value)

    if value < block.minValue { block.minValue = value }
    if value > block.maxValue { block.maxValue = value }

    if len(block.timestamps) >= tse.config.CompressionThreshold {
        tse.persistBlock(metric, partitionKey, block)
        block.timestamps = block.timestamps[:0]
        block.values = block.values[:0]
        block.minValue = math.MaxFloat64
        block.maxValue = -math.MaxFloat64
    }
    tse.storage.mu.Unlock()

    return nil
}

This approach gives us fast writes. The operation is mostly just appending to slices in memory. The expensive compression and disk write happen in the background when a block is full.

But writing data is only half the battle. We need to get it back efficiently. This is where indexing comes in. If someone asks for the temperature from last Tuesday between 2 PM and 3 PM, we shouldn't have to scan all our data.

We build a simple but effective index. For each metric, we keep track of the minimum and maximum timestamp we've seen.

type TimeIndex struct {
    timeRanges  map[string]*TimeRange
}

type TimeRange struct {
    Min int64
    Max int64
}

We also use a Bloom filter. This is a probabilistic data structure that can tell us if we've definitely never seen a metric. It's a quick check to avoid fruitless searches.

When a query comes in, we first check the time range index. If the requested time window doesn't overlap with the data we have for that metric, we return an empty result immediately.

func (tse *TimeSeriesEngine) Query(metric string, start, end int64, agg AggregationType, interval int64) ([]DataPoint, error) {
    tr, exists := tse.index.timeRanges[metric]
    if !exists || end < tr.Min || start > tr.Max {
        return []DataPoint{}, nil
    }

    // Adjust query range to available data
    if start < tr.Min { start = tr.Min }
    if end > tr.Max { end = tr.Max }

    // Get relevant partitions
    partitions := tse.getPartitionsForRange(start, end)

    // ... execute query on partitions
}

We then determine which time partitions intersect with our query. If we partition by hour, and the query is for 2:30 PM to 4:15 PM, we'll need the 2 PM, 3 PM, and 4 PM partitions.

The real power comes next: parallel query execution. We can process each partition independently. In Go, this is elegantly done with goroutines and channels.

    var wg sync.WaitGroup
    resultChan := make(chan []DataPoint, len(partitions))

    for _, partitionKey := range partitions {
        wg.Add(1)
        go func(pKey int64) {
            defer wg.Done()
            points, _ := tse.queryPartition(metric, pKey, start, end, agg, interval)
            resultChan <- points
        }(partitionKey)
    }

    wg.Wait()
    close(resultChan)

    var allPoints []DataPoint
    for points := range resultChan {
        allPoints = append(allPoints, points...)
    }

Each goroutine fetches and decodes the compressed blocks for its assigned partition. It filters the points to the exact time range, and if an aggregation interval is specified, it performs an initial aggregation. The results from all goroutines are then merged.

Speaking of aggregation, this is a core need. People rarely want every single millisecond reading. They want the average temperature per minute, or the maximum CPU load per hour.

We handle this with a two-stage process. First, each partition worker can do a partial aggregation. Then, the main thread does a final merge. Here's how we aggregate points into fixed time windows.

func (tse *TimeSeriesEngine) aggregateByInterval(points []DataPoint, agg AggregationType, interval int64) []DataPoint {
    var aggregated []DataPoint
    currentWindow := (points[0].Timestamp / interval) * interval
    var windowPoints []float64

    for _, point := range points {
        window := (point.Timestamp / interval) * interval

        if window != currentWindow && len(windowPoints) > 0 {
            aggValue := tse.applyAggregation(windowPoints, agg)
            aggregated = append(aggregated, DataPoint{
                Timestamp: currentWindow,
                Value:     aggValue,
            })
            windowPoints = windowPoints[:0]
            currentWindow = window
        }

        windowPoints = append(windowPoints, point.Value)
    }

    // Process last window
    if len(windowPoints) > 0 {
        aggValue := tse.applyAggregation(windowPoints, agg)
        aggregated = append(aggregated, DataPoint{
            Timestamp: currentWindow,
            Value:     aggValue,
        })
    }

    return aggregated
}

The function applyAggregation computes the average, sum, minimum, maximum, or count for a slice of values. It's straightforward but essential.

Now, let's put it all together and see the engine in action. We'll simulate writing some data and then querying it.

func main() {
    engine, err := NewTimeSeriesEngine("./tsdata")
    if err != nil {
        log.Fatal(err)
    }
    defer engine.Close()

    // Simulate writing temperature data
    startTime := time.Now().UnixMilli()
    for i := 0; i < 100000; i++ {
        timestamp := startTime + int64(i*1000) // 1 second intervals
        value := 50 + 10*math.Sin(float64(i)/100) // A sine wave pattern
        engine.WritePoint("temperature", timestamp, value)
    }

    // Query: Get average temperature per minute over the entire range
    endTime := startTime + 100000*1000
    points, err := engine.Query("temperature", startTime, endTime, AggregationAvg, 60000)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("Query returned %d data points (one per minute)\n", len(points))
    for i, point := range points {
        if i < 5 { // Show just the first five
            t := time.UnixMilli(point.Timestamp)
            fmt.Printf("%s: %.2f°\n", t.Format("15:04:05"), point.Value)
        }
    }
}

Running this, you'd see output showing the average temperature for each minute, calculated from the 60 individual seconds of data stored for that minute.

Building this taught me several things. Memory management is critical. We keep active blocks in memory for fast writes, but we must flush them to disk before they grow too large. We use Go's garbage collector efficiently by reusing slices.

Concurrency is natural in Go, but careful synchronization is needed. We use a read-write mutex (sync.RWMutex) to protect the in-memory blocks map. Many goroutines can read it simultaneously, but only one can write.

Choosing the right block size and partition interval is more art than science. It depends on your data and query patterns. A smaller block size means faster compression and more granular reads, but it increases the overhead of indexing. I've found blocks of 10,000 points and one-hour partitions to be a good starting point.

There are improvements one could make. Adding tags or labels to metrics, like temperature{sensor="A1", location="NYC"}, would make the system more flexible. Implementing a Write-Ahead Log (WAL) would guarantee no data loss between memory and disk. Downsampling—storing pre-aggregated data for older time ranges—would make queries over years of data much faster.

But the core idea remains: respect the nature of time-series data. It arrives in order. It is read in ranges. It benefits from compression that understands its sequential patterns. By aligning our storage format, compression methods, and query execution with these intrinsic properties, we can build a system that is both fast and efficient.

The code I've shown is a simplified but complete foundation. It can ingest hundreds of thousands of points per second on modest hardware and answer complex range queries in milliseconds. It demonstrates that with careful design and the right algorithms, managing the relentless flow of time-series data is not just possible, but can be elegantly simple.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Build Service Discovery and Load Balancing for Distributed Systems in Go

Nithin Bharadwaj — Tue, 21 Apr 2026 12:59:37 +0000

Imagine you have a team of workers in a large factory. Each worker performs a specific task, like painting or assembling. Now, imagine workers can appear, disappear, or get sick at any moment. Your job is to make sure that every request for a task—like "paint this car blue"—always finds a healthy, available painter. If one painter is busy, you send the request to another. If a painter gets sick, you stop sending them work until they recover. This is the core challenge of distributed computing, and it's what we solve with service discovery and load balancing.

I want to show you how to build the system that manages this. We'll create a central registry where services announce themselves. We'll build a health checker that constantly pings them. We'll design a router that picks the best instance for each new request. The goal is to make a network of services that feels as reliable as a single, solid machine.

Let's start with the foundation: the service registry. Think of it as a dynamic phone book. When a new service instance starts, it calls the registry to say, "I'm here." It provides its address, port, and some details about itself.

package main

import (
    "fmt"
    "sync"
    "time"
)

// ServiceStatus tracks the health of an instance.
type ServiceStatus int

const (
    StatusPending ServiceStatus = iota
    StatusHealthy
    StatusUnhealthy
)

// Service holds the data for one running instance.
type Service struct {
    ID         string
    Name       string
    Address    string
    Port       int
    Tags       []string
    Status     ServiceStatus
    LastHealth time.Time
}

// ServiceRegistry is our dynamic phone book.
type ServiceRegistry struct {
    services map[string]*Service // Map from Service ID to the Service object.
    mu       sync.RWMutex        // A lock to protect the map from concurrent access.
}

// NewServiceRegistry creates a new, empty registry.
func NewServiceRegistry() *ServiceRegistry {
    return &ServiceRegistry{
        services: make(map[string]*Service),
    }
}

// Register is called by a service when it starts.
func (sr *ServiceRegistry) Register(name, address string, port int, tags []string) error {
    sr.mu.Lock()
    defer sr.mu.Unlock() // This ensures the lock is released when the function finishes.

    // Create a unique ID for this instance.
    serviceID := fmt.Sprintf("%s-%s:%d", name, address, port)

    // Check if it's already registered.
    if _, exists := sr.services[serviceID]; exists {
        return fmt.Errorf("service %s already registered", serviceID)
    }

    // Add the new service to our map.
    sr.services[serviceID] = &Service{
        ID:         serviceID,
        Name:       name,
        Address:    address,
        Port:       port,
        Tags:       tags,
        Status:     StatusPending, // Starts as pending until health-checked.
        LastHealth: time.Now(),
    }

    fmt.Printf("Registered new service: %s\n", serviceID)
    return nil
}

When a service shuts down gracefully, it should call Deregister. But what if it crashes? That's where our next component comes in. We need a health checker. It will periodically ask each service, "Are you okay?" If a service doesn't answer, we mark it as unhealthy.

We can check health in different ways: by making an HTTP request to a /health endpoint, by trying to open a TCP connection, or even by running a custom piece of code.

type HealthCheckType int

const (
    HealthCheckHTTP HealthCheckType = iota
    HealthCheckTCP
)

type HealthCheck struct {
    Type         HealthCheckType
    HTTPEndpoint string // For HTTP checks.
    Address      string // For TCP checks.
    Port         int
    Interval     time.Duration // How often to check, e.g., every 10 seconds.
    Timeout      time.Duration // How long to wait for a response.
}

// HealthChecker runs all checks and updates the registry.
type HealthChecker struct {
    registry   *ServiceRegistry
    checks     map[string]*HealthCheck // Map from Service ID to its check config.
    httpClient *http.Client
}

// performCheck runs a single health check.
func (hc *HealthChecker) performCheck(serviceID string, check *HealthCheck) {
    var healthy bool
    var latency time.Duration

    start := time.Now()

    switch check.Type {
    case HealthCheckHTTP:
        // Try to make a GET request.
        resp, err := hc.httpClient.Get(check.HTTPEndpoint)
        if err == nil && resp.StatusCode == 200 {
            healthy = true
        }
        if resp != nil {
            resp.Body.Close()
        }
    case HealthCheckTCP:
        // Try to open a socket.
        address := fmt.Sprintf("%s:%d", check.Address, check.Port)
        conn, err := net.DialTimeout("tcp", address, check.Timeout)
        if err == nil {
            healthy = true
            conn.Close()
        }
    }

    latency = time.Since(start)

    // Update the service's status in the registry.
    hc.registry.mu.Lock()
    defer hc.registry.mu.Unlock()
    if service, exists := hc.registry.services[serviceID]; exists {
        if healthy {
            service.Status = StatusHealthy
        } else {
            service.Status = StatusUnhealthy
        }
        service.LastHealth = time.Now()
    }
}

The health checker runs in a loop, performing each check at its defined interval. This keeps our registry's view of the world up-to-date. Now we have a list of services and we know which ones are healthy. The next step is to decide where to send incoming work. This is load balancing.

The simplest method is round-robin. Just go down the list, one after another.

type RoundRobinStrategy struct {
    currentIndex uint32 // Use an atomic counter for thread-safety.
    instances    []*Service
}

func (rr *RoundRobinStrategy) Pick() *Service {
    if len(rr.instances) == 0 {
        return nil
    }
    // Atomically increment the index and wrap around using modulo.
    index := atomic.AddUint32(&rr.currentIndex, 1) % uint32(len(rr.instances))
    return rr.instances[index]
}

But what if some instances are more powerful than others? Or what if some are already handling many requests? We might want a "least connections" strategy.

type LeastConnectionsStrategy struct{}

func (lc *LeastConnectionsStrategy) Pick(instances []*Service) *Service {
    if len(instances) == 0 {
        return nil
    }
    var selected *Service
    minConns := int32(1 << 30) // Start with a very large number.

    for _, instance := range instances {
        // Assume each Service has an atomic counter for active connections.
        conns := atomic.LoadInt32(&instance.Connections)
        if conns < minConns {
            minConns = conns
            selected = instance
        }
    }
    return selected
}

Sometimes, the network is the bottleneck. A server might be healthy but slow because it's far away. A latency-aware strategy can track response times and prefer faster instances.

We need a way to route actual user requests. In an HTTP server, we can use a middleware. This middleware intercepts each request, figures out which service it's for, uses the load balancer to pick an instance, and then forwards the request.

func DiscoveryMiddleware(registry *ServiceRegistry, lb *LoadBalancer) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            // Extract the target service name from the request path or a header.
            // For example, a path like /api/users/ might route to the "user-service".
            serviceName := "user-service" // Simplified extraction.

            // Get a list of healthy instances for this service.
            instances, err := registry.Discover(serviceName, true) // true = healthy only.
            if err != nil {
                http.Error(w, "Service not available", http.StatusServiceUnavailable)
                return
            }

            // Use the load balancer strategy to pick one.
            instance := lb.Pick(instances)
            if instance == nil {
                http.Error(w, "Service not available", http.StatusServiceUnavailable)
                return
            }

            // Forward the request to the chosen instance.
            // This is a simple reverse proxy setup.
            proxy := httputil.NewSingleHostReverseProxy(&url.URL{
                Scheme: "http",
                Host:   fmt.Sprintf("%s:%d", instance.Address, instance.Port),
            })
            proxy.ServeHTTP(w, r)
        })
    }
}

This is the basic flow. But a production system needs more. It needs to watch for changes. If a health check fails, the registry should immediately notify the load balancer so it stops using that instance. We can use a watcher pattern.

type RegistryEvent struct {
    Type    string // "REGISTER", "DEREGISTER", "HEALTH_CHANGE"
    Service *Service
}

type RegistryWatcher interface {
    Notify(event RegistryEvent)
}

// The LoadBalancer can implement this interface.
func (lb *LoadBalancer) Notify(event RegistryEvent) {
    lb.mu.Lock()
    defer lb.mu.Unlock()

    switch event.Type {
    case "HEALTH_CHANGE":
        if event.Service.Status == StatusUnhealthy {
            // Remove this instance from the active pool.
            lb.removeInstance(event.Service.ID)
        } else {
            // Add it back.
            lb.addInstance(event.Service)
        }
    }
}

Dynamic routing means your system automatically adjusts to failures and new capacity. When you deploy a new version of a service, you can register the new instances, let them pass health checks, and the load balancer will start sending them traffic. You can then gracefully deregister the old ones.

Let's look at a more complete example, putting the registry, health checker, and a simple load balancer together.

func main() {
    // Create the core components.
    registry := NewServiceRegistry()
    healthChecker := NewHealthChecker(registry)
    lb := NewLoadBalancer(registry, "round_robin")

    // Start the health checker in the background.
    go healthChecker.Run()

    // Simulate a service registering.
    registry.Register("cart-service", "10.0.0.1", 8080, []string{"v1"})

    // Define its health check.
    healthChecker.AddCheck("cart-service-10.0.0.1:8080", &HealthCheck{
        Type:         HealthCheckHTTP,
        HTTPEndpoint: "http://10.0.0.1:8080/health",
        Interval:     10 * time.Second,
        Timeout:      2 * time.Second,
    })

    // Set up an HTTP server that uses our discovery middleware.
    router := http.NewServeMux()
    // Your actual application routes would go here on `router`.

    // Wrap the router with our discovery middleware.
    app := DiscoveryMiddleware(registry, lb)(router)

    fmt.Println("Gateway listening on :8080")
    http.ListenAndServe(":8080", app)
}

In this code, the gateway listens on port 8080. A request comes in for /api/cart. The middleware asks the registry for a healthy "cart-service". The registry checks its list, which is being updated by the health checker. The load balancer picks one instance, say 10.0.0.1:8080. The request is forwarded there.

What about more complex rules? Maybe you want to send 1% of traffic to a new version for testing. This is where routing policies come in. You can tag your services with versions and have the load balancer read those tags.

type RoutingPolicy struct {
    ServiceName string
    MatchTags   map[string]string // e.g., {"version": "canary"}
    Weight      int               // e.g., 1 for 1% traffic.
}

func (lb *LoadBalancer) PickWithPolicy(serviceName string, policies []RoutingPolicy) *Service {
    // First, get all healthy instances.
    instances := lb.registry.GetHealthyInstances(serviceName)

    // Filter instances based on the policy tags.
    var filteredInstances []*Service
    for _, instance := range instances {
        if lb.matchesPolicy(instance, policies) {
            filteredInstances = append(filteredInstances, instance)
        }
    }

    // If the filtered list has instances, pick from it based on weight.
    // Otherwise, fall back to the general pool.
    if len(filteredInstances) > 0 {
        // Implement weighted logic here.
        return lb.weightedPick(filteredInstances, policies)
    }
    return lb.Pick(instances) // Default strategy.
}

Building this yourself teaches you the mechanics, but for a real, large-scale system, you'd likely use existing tools like Consul, etcd, or Kubernetes' built-in service discovery. These provide the distributed, consistent storage we glossed over. Our in-memory map works for a single registry node, but what if that node crashes? You need multiple registry nodes that agree on the state. That's a problem solved by consensus algorithms like Raft, which is used by Consul and etcd.

The code patterns, however, remain similar. You register services, you check health, you balance load. The difference is in the durability and fault-tolerance of the registry itself.

The system I've walked you through is like the nervous system for your microservices. It knows where everything is, it feels when something is wrong, and it directs traffic smoothly. It turns a collection of independent, fragile processes into a resilient, adaptable application.

Start simple. Build a registry that keeps a list. Add a health checker that updates it. Make a load balancer that reads from it. You'll learn more from getting these three pieces talking to each other than from any diagram or lecture. Once you have that basic loop working—registration, health, routing—you've built the heart of the system. Everything after that is making it faster, more reliable, and easier to manage.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Reverse Engineer Any Database Structure Using Python

Nithin Bharadwaj — Mon, 20 Apr 2026 23:11:21 +0000

Let's talk about databases. They're everywhere, holding the information that makes our digital world run. But sometimes, you encounter a database you didn't build. There's no manual, no diagram, just raw tables and columns. Your job is to figure out how it all fits together. This process of figuring out an existing database's structure is what we call reverse engineering. Python is an excellent tool for this job, and I want to share some practical ways to use it.

I'll walk you through several techniques, from the simple to the more complex, with plenty of code you can try yourself. Think of it like being a digital archaeologist, carefully brushing away the dust to reveal the blueprint of an ancient system.

The first and most fundamental step is connecting to the database and asking it directly about its structure. Every major database system—like SQLite, PostgreSQL, or MySQL—has a special set of system tables that store this blueprint. In Python, we use libraries to talk to these systems.

Here's a basic starter. We create a class that can connect to different types of databases and fetch a list of all the tables.

import sqlite3
import psycopg2
import mysql.connector

class DatabaseExplorer:
    def __init__(self, db_type='sqlite', **connection_params):
        self.db_type = db_type
        self.connection_params = connection_params
        self.connection = None

    def connect(self):
        """Establishes a connection to the database."""
        if self.db_type == 'sqlite':
            self.connection = sqlite3.connect(**self.connection_params)
        elif self.db_type == 'postgresql':
            self.connection = psycopg2.connect(**self.connection_params)
        elif self.db_type == 'mysql':
            self.connection = mysql.connector.connect(**self.connection_params)
        else:
            raise ValueError(f"This database type isn't supported yet: {self.db_type}")
        print(f"Connected to {self.db_type} database.")
        return self.connection

# Let's try it with a SQLite file.
explorer = DatabaseExplorer('sqlite', database='mystery_database.db')
explorer.connect()

Once you're connected, you need to know what you're looking at. You query the database's internal catalog. For a SQLite database, you ask the sqlite_master table. For PostgreSQL or MySQL, you query the information_schema. The code below gets the table names.

    def get_table_names(self):
        """Fetches all non-system table names from the database."""
        cursor = self.connection.cursor()
        tables = []

        if self.db_type == 'sqlite':
            cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
            for row in cursor.fetchall():
                table_name = row[0]
                # Filter out internal SQLite tables
                if not table_name.startswith('sqlite_'):
                    tables.append(table_name)

        elif self.db_type == 'postgresql':
            cursor.execute("""
                SELECT table_name
                FROM information_schema.tables
                WHERE table_schema = 'public';
            """)
            tables = [row[0] for row in cursor.fetchall()]

        elif self.db_type == 'mysql':
            cursor.execute("""
                SELECT table_name
                FROM information_schema.tables
                WHERE table_schema = DATABASE();
            """)
            tables = [row[0] for row in cursor.fetchall()]

        cursor.close()
        return tables

# Using our explorer
table_list = explorer.get_table_names()
print(f"Found tables: {table_list}")

Knowing table names is just the beginning. The real details are in the columns. What's each column called? What type of data does it hold? Is it a primary key? We can write a method to inspect each table closely. Let's look at a SQLite example.

    def inspect_table(self, table_name):
        """Gets detailed column information for a specific table."""
        cursor = self.connection.cursor()
        columns_info = []

        if self.db_type == 'sqlite':
            # PRAGMA is SQLite's special command for internal info
            cursor.execute(f"PRAGMA table_info('{table_name}');")
            for row in cursor.fetchall():
                # row format: (cid, name, type, notnull, default_value, pk)
                col_id, col_name, col_type, not_null, default_val, is_primary = row
                columns_info.append({
                    'name': col_name,
                    'type': col_type,
                    'can_be_null': not_null == 0,  # 0 means NULL is allowed
                    'default_value': default_val,
                    'is_primary_key': is_primary == 1
                })

        cursor.close()
        return columns_info

# Inspect a table called 'users'
user_columns = explorer.inspect_table('users')
for col in user_columns:
    pk_flag = " (Primary Key)" if col['is_primary_key'] else ""
    print(f"  - {col['name']}: {col['type']}{pk_flag}")

Tables rarely live in isolation. They are connected through foreign keys, which are crucial for understanding the database's logic. A foreign key in one table points to the primary key in another, creating a relationship. Finding these links is our next technique.

    def find_foreign_keys(self, table_name):
        """Identifies relationships where this table references others."""
        cursor = self.connection.cursor()
        relationships = []

        if self.db_type == 'sqlite':
            cursor.execute(f"PRAGMA foreign_key_list('{table_name}');")
            for row in cursor.fetchall():
                # row: (id, seq, table, from, to, on_update, on_delete, match)
                _, _, target_table, from_column, to_column, _, _, _ = row
                relationships.append({
                    'from_column': from_column,
                    'to_table': target_table,
                    'to_column': to_column
                })

        cursor.close()
        return relationships

# Find what the 'orders' table connects to
order_links = explorer.find_foreign_keys('orders')
for link in order_links:
    print(f"The 'orders.{link['from_column']}' column points to '{link['to_table']}.{link['to_column']}'.")

Sometimes you don't have a live database connection. All you have is a giant .sql file—a dump of the database structure. This is where SQL parsing comes in. Instead of querying a live system, you read the text file and look for patterns. It's like reading the architect's original notes instead of touring the building.

We can use regular expressions, or more powerful tools like the sqlparse library, to break down the SQL script.

import re
import sqlparse

def parse_sql_dump(file_path):
    """Reads an SQL file and extracts CREATE TABLE statements."""
    with open(file_path, 'r') as f:
        sql_content = f.read()

    # Use sqlparse to split into individual statements and normalize them
    statements = sqlparse.parse(sql_content)
    table_defs = {}

    for statement in statements:
        # Normalize: lowercase, strip extra spaces
        normalized = str(statement).lower().strip()

        # Look for the start of a CREATE TABLE command
        if normalized.startswith('create table'):
            # A very simple regex to get the table name.
            # Warning: This is fragile for complex SQL.
            match = re.search(r'create table (?:if not exists )?[\`"]?(\w+)[\`"]?', normalized)
            if match:
                table_name = match.group(1)
                table_defs[table_name] = normalized  # Store the full statement

    return table_defs

# Parse a file
definitions = parse_sql_dump('legacy_dump.sql')
print(f"Found definitions for tables: {list(definitions.keys())}")

With the raw CREATE TABLE text, we can do more precise extraction. We want to pull out column definitions, constraints, and keys. Let's write a more focused parser for a single statement.

def analyze_create_statement(create_sql):
    """Takes a single CREATE TABLE SQL string and breaks it down."""
    # First, let's clean it up a bit with sqlparse
    parsed = sqlparse.parse(create_sql)[0]
    # sqlparse can identify tokens like keywords, identifiers, punctuation
    # We'll do a simpler string-based approach for clarity.

    # Find the part between the first parentheses
    start = create_sql.find('(')
    end = create_sql.rfind(')')
    if start == -1 or end == -1:
        return {}

    inner_content = create_sql[start+1:end]
    # Split by commas, but be careful of commas inside parentheses (like in CHECK constraints)
    # This is a naive split for demonstration.
    lines = [line.strip() for line in inner_content.split(',')]

    columns = []
    primary_key = None
    foreign_keys = []

    for line in lines:
        line_upper = line.upper()
        if line.startswith('`') or line[0].isalpha():  # Heuristic for a column definition
            parts = line.split()
            col_name = parts[0].strip('`"')
            col_type = parts[1] if len(parts) > 1 else 'unknown'
            columns.append({'name': col_name, 'type': col_type})
        elif 'PRIMARY KEY' in line_upper:
            # Find column(s) inside parentheses
            match = re.search(r'PRIMARY KEY\s*\(([^)]+)\)', line_upper)
            if match:
                primary_key = [col.strip(' `"') for col in match.group(1).split(',')]
        elif 'FOREIGN KEY' in line_upper:
            # Match pattern: FOREIGN KEY (col1) REFERENCES other_table(other_col)
            match = re.search(r'FOREIGN KEY\s*\(([^)]+)\)\s*REFERENCES\s*(\w+)\s*\(([^)]+)\)', line_upper)
            if match:
                from_col, to_table, to_col = match.groups()
                foreign_keys.append({
                    'from_column': from_col.strip(' `"'),
                    'to_table': to_table.strip(' `"'),
                    'to_column': to_col.strip(' `"')
                })

    return {
        'columns': columns,
        'primary_key': primary_key,
        'foreign_keys': foreign_keys
    }

# Test it
sample_sql = """
CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    amount DECIMAL(10,2),
    FOREIGN KEY (customer_id) REFERENCES customers(cust_id)
);
"""
table_info = analyze_create_statement(sample_sql)
print(f"Columns: {[c['name'] for c in table_info['columns']]}")
print(f"Primary Key: {table_info['primary_key']}")
print(f"Foreign Keys: {table_info['foreign_keys']}")

After gathering all this raw data—tables, columns, keys—you have a collection of facts. The next step is to build a model that shows how they connect. This is about moving from a list to a map. I like to build a dictionary where each table knows about its connections.

class SchemaModel:
    def __init__(self):
        self.tables = {}  # Format: { 'table_name': { 'columns': [], 'links_to': [], 'linked_from': [] } }

    def add_table(self, name, columns, primary_key, foreign_keys):
        """Adds a table and its immediate foreign key connections."""
        self.tables[name] = {
            'columns': columns,
            'primary_key': primary_key,
            'outgoing_links': foreign_keys,  # Links FROM this table
            'incoming_links': []             # Links TO this table (will be filled later)
        }

    def build_relationships(self):
        """Processes all tables to find bidirectional relationships."""
        # First, reset incoming links
        for table_info in self.tables.values():
            table_info['incoming_links'] = []

        # For each foreign key in a table, tell the target table about it.
        for source_table, info in self.tables.items():
            for fk in info['outgoing_links']:
                target_table = fk['to_table']
                if target_table in self.tables:
                    # Add a note to the target table that someone points to it.
                    self.tables[target_table]['incoming_links'].append({
                        'from_table': source_table,
                        'from_column': fk['from_column'],
                        'to_column': fk['to_column']
                    })

    def find_orphaned_tables(self):
        """Identifies tables with no relationships to others."""
        orphaned = []
        for name, info in self.tables.items():
            if not info['outgoing_links'] and not info['incoming_links']:
                orphaned.append(name)
        return orphaned

# Build a simple model
model = SchemaModel()
model.add_table('customers', [{'name':'cust_id', 'type':'INT'}], ['cust_id'], [])
model.add_table('orders', [{'name':'order_id', 'type':'INT'}, {'name':'cust_id', 'type':'INT'}],
                ['order_id'],
                [{'from_column':'cust_id', 'to_table':'customers', 'to_column':'cust_id'}])
model.build_relationships()

print(f"Customers is linked from: {model.tables['customers']['incoming_links']}")

A picture is worth a thousand table definitions. Once you have a model, visualizing it helps you and others understand the design instantly. We can use graphviz via the diagrams library or matplotlib to draw a simple graph.

Here’s a basic example using networkx and matplotlib to plot tables as nodes and foreign keys as edges.

import matplotlib.pyplot as plt
import networkx as nx

def visualize_schema(model):
    """Creates a simple directed graph of the database schema."""
    G = nx.DiGraph()  # Directed Graph

    # Add nodes (tables)
    for table_name in model.tables:
        G.add_node(table_name)

    # Add edges (foreign key relationships)
    for source_table, info in model.tables.items():
        for fk in info['outgoing_links']:
            target_table = fk['to_table']
            if target_table in model.tables:
                # Label the edge with the column name
                G.add_edge(source_table, target_table, label=fk['from_column'])

    # Draw the graph
    pos = nx.spring_layout(G, seed=42)  # Layout for consistent placement
    plt.figure(figsize=(10, 8))

    nx.draw_networkx_nodes(G, pos, node_color='lightblue', node_size=2000)
    nx.draw_networkx_labels(G, pos, font_weight='bold')

    # Draw edges with labels
    edge_labels = nx.get_edge_attributes(G, 'label')
    nx.draw_networkx_edges(G, pos, arrowstyle='-|>', arrowsize=20)
    nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels)

    plt.title("Database Schema Relationships")
    plt.axis('off')  # Turn off the axis
    plt.tight_layout()
    plt.show()

# Visualize our small model
visualize_schema(model)

The final technique is about putting it all together into a useful report. You've discovered the structure; now document it. Generate a Markdown file, a JSON schema, or even SQL CREATE statements that reflect your understanding. This creates a living document for future developers.

def generate_markdown_report(model, filename='schema_report.md'):
    """Produces a human-readable Markdown document of the schema."""
    with open(filename, 'w') as f:
        f.write("# Database Schema Analysis Report\n\n")
        f.write(f"*Generated on: {datetime.now().strftime('%Y-%m-%d %H:%M')}*\n")
        f.write(f"*Total Tables Analyzed: {len(model.tables)}*\n\n")

        for table_name, info in model.tables.items():
            f.write(f"## Table: `{table_name}`\n\n")

            # Columns
            f.write("### Columns\n")
            f.write("| Name | Type | Primary Key |\n")
            f.write("|------|------|-------------|\n")
            for col in info['columns']:
                is_pk = "Yes" if col['name'] in (info['primary_key'] or []) else "No"
                f.write(f"| {col['name']} | {col['type']} | {is_pk} |\n")
            f.write("\n")

            # Relationships
            f.write("### Relationships\n")
            if info['outgoing_links']:
                f.write("**This table references:**\n")
                for fk in info['outgoing_links']:
                    f.write(f"- `{table_name}.{fk['from_column']}` -> `{fk['to_table']}.{fk['to_column']}`\n")
            if info['incoming_links']:
                f.write("\n**Referenced by:**\n")
                for link in info['incoming_links']:
                    f.write(f"- `{link['from_table']}.{link['from_column']}` -> `{table_name}.{link['to_column']}`\n")
            if not info['outgoing_links'] and not info['incoming_links']:
                f.write("*No foreign key relationships found.*\n")
            f.write("\n---\n\n")

    print(f"Report written to {filename}")

# Generate the report
generate_markdown_report(model)

Working through an unknown database can feel daunting. I remember once facing a database with over 200 tables, named with cryptic abbreviations, and no documentation. By applying these Python techniques systematically—first mapping the tables, then drawing the connections—I slowly pieced together the business logic it represented. The moment the visualization graph appeared, showing a clear central hub of customer data, the entire design intention became obvious. Python didn't just give me answers; it gave me a way to ask the right questions. These methods turn a opaque data store into a clear, manageable structure you can understand, document, and work with effectively.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Build a Bulletproof Database Migration System in JavaScript

Nithin Bharadwaj — Mon, 20 Apr 2026 22:57:30 +0000

Let me tell you about something that used to keep me awake at night. Picture this: you've built an application that people actually use. Real users, real data. Then you need to change how you store that data. Maybe you need to add a new field, split a table, or change how relationships work. The thought of running SQL commands directly on your production database should make your palms sweat. I've been there. That's where database migrations come in.

Think of migrations as version control for your database structure. Just like Git tracks changes to your code, migrations track changes to your database. You write small, incremental scripts that move your database from one state to another. You can also go backward if something goes wrong. This approach has saved me from disaster more times than I can count.

Let me show you how I handle migrations in my JavaScript projects. I'll start with the foundation - a migration management system. This class keeps track of what's been applied and what needs to run next.

class MigrationManager {
  constructor(database, options = {}) {
    this.database = database;
    this.options = {
      migrationsTable: options.migrationsTable || 'schema_migrations',
      migrationDirectory: options.migrationDirectory || './migrations',
      validateChecksums: options.validateChecksums !== false,
      ...options
    };

    this.migrations = new Map();
    this.appliedMigrations = new Set();
    this.setupMigrationSystem();
  }

  async setupMigrationSystem() {
    // Create the table that tracks migrations if it doesn't exist
    await this.createMigrationsTable();

    // Load which migrations have already been applied
    await this.loadAppliedMigrations();

    // Find all migration files in the directory
    await this.discoverMigrations();
  }
}

This manager class is the brains of the operation. It knows about your database connection, where to find migration files, and which migrations have already been applied. The setupMigrationSystem method gets everything ready to work. First, it makes sure we have a table to track migrations. Then it loads what's already been done. Finally, it discovers new migration files.

Versioning is crucial. Each migration needs a unique identifier that shows its order. I use timestamps because they're naturally ordered and avoid conflicts between team members. Every migration file starts with a timestamp, like 1698765432100_create_users_table.js. This ensures they run in the right order.

async function createMigration(name) {
  const timestamp = Date.now();
  const version = `${timestamp}_${Math.random().toString(36).substr(2, 6)}`;

  const template = `
module.exports = {
  version: '${version}',
  name: '${name}',

  async migrate(db) {
    // Your migration code here
  },

  async rollback(db) {
    // Your rollback code here
  }
};`;

  const filename = `${version}_${name}.js`;
  await writeFile(`./migrations/${filename}`, template);

  return { version, filename };
}

Creating a migration should be easy. This function generates a new migration file with a unique version and basic structure. The version combines a timestamp with a random string. This approach means two developers can create migrations at the same time without conflicts.

Now let's look at what goes inside a migration. A migration has two main parts: the migrate function that applies changes, and the rollback function that reverses them. Always write both. You might think you'll never need to roll back, but trust me, you will.

// A simple migration example
module.exports = {
  version: '1698765432100_abc123',
  name: 'create_users_table',

  async migrate(db) {
    await db.query(`
      CREATE TABLE users (
        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
        email VARCHAR(255) UNIQUE NOT NULL,
        username VARCHAR(100) UNIQUE NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
      )
    `);
  },

  async rollback(db) {
    await db.query('DROP TABLE IF EXISTS users');
  }
};

This migration creates a users table. The migrate function runs the SQL to create the table. The rollback function drops the table if we need to undo the change. Notice how simple and focused it is. Each migration should do one logical change.

Running migrations happens in order. The manager looks at all migration files, sorts them by version, and applies any that haven't been run yet. It keeps track in the database which ones have been applied successfully.

async function runMigrations() {
  const manager = new MigrationManager(database);
  const pending = manager.getPendingMigrations();

  for (const migration of pending) {
    console.log(`Applying ${migration.name}...`);
    await migration.migrate(database);

    // Record that this migration was applied
    await recordMigration(migration);
  }

  console.log(`Applied ${pending.length} migrations`);
}

The key here is that migrations run in a specific order and each is recorded after successful completion. If a migration fails partway through, we don't mark it as completed. This means we can fix the issue and try again.

Rolling back is just as important as moving forward. Sometimes a migration has unexpected consequences. Maybe it causes performance issues, or there's a bug in the logic. Being able to revert safely is non-negotiable.

async function rollbackMigrations(targetVersion) {
  const manager = new MigrationManager(database);
  const applied = manager.getAppliedMigrations();

  // Find which migrations to roll back
  let toRollback = applied;
  if (targetVersion) {
    const index = applied.findIndex(m => m.version === targetVersion);
    toRollback = applied.slice(0, index + 1);
  }

  // Roll back in reverse order
  toRollback.reverse();

  for (const migration of toRollback) {
    console.log(`Rolling back ${migration.name}...`);
    await migration.rollback(database);

    // Remove the migration record
    await removeMigrationRecord(migration);
  }
}

Rollback happens in reverse order. If we applied migrations A, B, then C, we roll back C, then B, then A. This ensures we return to a consistent state. The targetVersion parameter lets us roll back to a specific point in history.

Not all migrations are simple table creations. Real-world scenarios often involve data transformation. Let me show you a more complex example where we change how user roles are stored.

module.exports = {
  version: '1698765432200_migrate_roles',
  name: 'convert_roles_to_enum',

  async migrate(db) {
    // Create a new enum type for roles
    await db.query(`
      CREATE TYPE user_role AS ENUM ('user', 'admin', 'moderator')
    `);

    // Add the new column
    await db.query(`
      ALTER TABLE users 
      ADD COLUMN role user_role DEFAULT 'user'
    `);

    // Convert existing data
    const users = await db.query('SELECT id, old_role FROM users');

    for (const user of users) {
      let newRole = 'user';
      if (user.old_role === 'administrator') newRole = 'admin';
      if (user.old_role === 'mod') newRole = 'moderator';

      await db.query(
        'UPDATE users SET role = $1 WHERE id = $2',
        [newRole, user.id]
      );
    }

    // Remove the old column
    await db.query('ALTER TABLE users DROP COLUMN old_role');
  },

  async rollback(db) {
    // Add back the old column
    await db.query(`
      ALTER TABLE users 
      ADD COLUMN old_role VARCHAR(50)
    `);

    // Convert data back
    const users = await db.query('SELECT id, role FROM users');

    for (const user of users) {
      let oldRole = 'user';
      if (user.role === 'admin') oldRole = 'administrator';
      if (user.role === 'moderator') oldRole = 'mod';

      await db.query(
        'UPDATE users SET old_role = $1 WHERE id = $2',
        [oldRole, user.id]
      );
    }

    // Remove the new column
    await db.query('ALTER TABLE users DROP COLUMN role');

    // Remove the enum type
    await db.query('DROP TYPE IF EXISTS user_role');
  }
};

This migration converts string-based roles to an enum type. Notice how the rollback carefully reverses every step. We add back the old column, convert the data back, then remove the new column and enum type. Data migrations like this require extra care because you're changing existing information.

Sometimes you need to handle large datasets. Running updates on millions of rows at once can lock tables and cause downtime. Batch processing solves this by working in small chunks.

async function migrateLargeDataset(db) {
  const batchSize = 1000;
  let offset = 0;
  let processed = 0;

  do {
    // Get a batch of records
    const users = await db.query(
      `SELECT id, data FROM users 
       WHERE processed = false 
       ORDER BY id 
       LIMIT $1 OFFSET $2`,
      [batchSize, offset]
    );

    if (users.length === 0) break;

    // Process this batch
    for (const user of users) {
      // Transform data
      const newData = transformData(user.data);

      await db.query(
        'UPDATE users SET data = $1, processed = true WHERE id = $2',
        [newData, user.id]
      );
    }

    processed += users.length;
    offset += batchSize;

    console.log(`Processed ${processed} records`);

  } while (true);

  console.log(`Total processed: ${processed}`);
}

Batch processing keeps each transaction small and manageable. We process records in groups of 1000 (or whatever size makes sense for your database). This approach prevents timeouts and reduces locking. The processed flag ensures we don't reprocess the same records if the migration needs to restart.

Not all changes are additions. Sometimes you need to remove things. But deleting data or columns can be risky. A safer approach is to deprecate first, then remove later.

async function safeColumnRemoval(db) {
  // First migration: mark column as deprecated
  await db.query(`
    COMMENT ON COLUMN users.old_column 
    IS 'DEPRECATED: Will be removed after 2024-01-01'
  `);

  // Update application code to stop using this column

  // Second migration (weeks or months later): actually remove
  await db.query('ALTER TABLE users DROP COLUMN old_column');
}

This two-step process gives you time to update your application code. First, you mark the column as deprecated. Your application should stop using it. After sufficient time has passed (and you're sure nothing depends on it), you can safely remove it.

Testing migrations is critical. I never run a migration in production without testing it first. Here's how I structure my migration tests.

describe('User table migration', () => {
  let testDb;

  beforeEach(async () => {
    // Set up a test database
    testDb = await createTestDatabase();
  });

  afterEach(async () => {
    // Clean up test database
    await testDb.close();
  });

  test('migration applies correctly', async () => {
    // Run the migration
    const migration = require('./migrations/1698765432100_create_users_table');
    await migration.migrate(testDb);

    // Verify the table was created
    const tables = await testDb.query(`
      SELECT table_name 
      FROM information_schema.tables 
      WHERE table_name = 'users'
    `);

    expect(tables).toHaveLength(1);
  });

  test('rollback works correctly', async () => {
    // Apply then roll back
    const migration = require('./migrations/1698765432100_create_users_table');
    await migration.migrate(testDb);
    await migration.rollback(testDb);

    // Verify table was removed
    const tables = await testDb.query(`
      SELECT table_name 
      FROM information_schema.tables 
      WHERE table_name = 'users'
    `);

    expect(tables).toHaveLength(0);
  });
});

Each migration gets its own test file. We test both the migrate and rollback functions. The test database is isolated from development and production databases. This setup catches issues before they reach real data.

Sometimes you need to run custom validation during a migration. Maybe you need to check data quality or enforce business rules.

async function validateDataDuringMigration(db) {
  // Check for invalid data before making changes
  const invalidUsers = await db.query(`
    SELECT id, email 
    FROM users 
    WHERE email NOT LIKE '%@%'
  `);

  if (invalidUsers.length > 0) {
    throw new Error(
      `Found ${invalidUsers.length} users with invalid emails. ` +
      'Fix data before running migration.'
    );
  }

  // If validation passes, proceed with migration
  await db.query(`
    ALTER TABLE users 
    ADD CONSTRAINT valid_email 
    CHECK (email ~* '^[A-Z0-9._%+-]+@[A-Z0-9.-]+\\.[A-Z]{2,}$')
  `);
}

Validation migrations check that data meets certain criteria before applying changes. If the data isn't ready, the migration fails with a clear error message. This prevents applying changes to invalid data.

Database constraints are another important aspect. Adding constraints helps maintain data integrity, but they need to be added carefully.

async function addConstraintsSafely(db) {
  // First, check if any existing data violates the constraint
  const violatingRows = await db.query(`
    SELECT id 
    FROM users 
    WHERE username IS NULL 
    OR LENGTH(username) < 3
  `);

  if (violatingRows.length > 0) {
    // Fix the data first
    await db.query(`
      UPDATE users 
      SET username = 'user_' || id 
      WHERE username IS NULL OR LENGTH(username) < 3
    `);
  }

  // Now it's safe to add the constraint
  await db.query(`
    ALTER TABLE users 
    ADD CONSTRAINT username_not_null 
    CHECK (username IS NOT NULL AND LENGTH(username) >= 3)
  `);
}

This migration first identifies rows that would violate the new constraint. It fixes those rows, then adds the constraint. This approach prevents the constraint from failing due to existing bad data.

Index management is another common migration task. Adding indexes can improve performance, but they come with costs during creation and maintenance.

async function manageIndexes(db) {
  // Add index for common query patterns
  await db.query(`
    CREATE INDEX idx_users_email_status 
    ON users(email, status) 
    WHERE status = 'active'
  `);

  // Remove unused indexes
  const unusedIndexes = await identifyUnusedIndexes(db);

  for (const index of unusedIndexes) {
    await db.query(`DROP INDEX IF EXISTS ${index.name}`);
  }
}

This migration adds a partial index for a common query pattern while removing indexes that aren't being used. Partial indexes (with WHERE clauses) are smaller and faster than full table indexes.

Sometimes you need to coordinate migrations across multiple services. This requires careful planning and communication.

async function coordinatedMigration(db, serviceClient) {
  // Step 1: Prepare database changes
  await db.query('ALTER TABLE orders ADD COLUMN new_field VARCHAR(255)');

  // Step 2: Notify other services
  await serviceClient.notify({
    event: 'schema_change',
    change: 'orders.new_field_added',
    timestamp: Date.now()
  });

  // Step 3: Wait for acknowledgment
  await waitForAcknowledgements(['inventory-service', 'billing-service']);

  // Step 4: Backfill data
  await backfillNewField(db);
}

When multiple services use the same database, changes need coordination. This pattern adds a field, notifies dependent services, waits for them to acknowledge, then proceeds with data population.

Finally, let me share my personal checklist for every migration. This has prevented many late-night debugging sessions.

const migrationChecklist = {
  before: [
    'Test migration on development database',
    'Verify rollback works correctly',
    'Check for data inconsistencies',
    'Estimate execution time on production data size',
    'Schedule during low-traffic period',
    'Create database backup',
    'Notify team members',
  ],

  during: [
    'Monitor performance metrics',
    'Watch for errors or timeouts',
    'Log progress at regular intervals',
    'Have rollback script ready',
  ],

  after: [
    'Verify application works correctly',
    'Check database constraints',
    'Monitor for performance regressions',
    'Update documentation',
    'Remove deprecated code that used old schema',
  ]
};

This checklist covers the entire migration process. The "before" items are about preparation and safety. The "during" items focus on monitoring and having an escape plan. The "after" items ensure everything works correctly and stays maintainable.

Database migrations don't have to be scary. With the right approach, they become a routine part of application development. Start simple, test thoroughly, and always have a way back. The techniques I've shown here have evolved from years of trial and error. They've handled everything from small startups to large-scale applications with millions of users.

Remember that migrations are about more than just changing database structure. They're about managing risk, preserving data, and enabling your application to evolve. Each migration is a small, safe step forward. Taken together, they allow your application to grow and adapt without breaking what already works.

The key insight I've gained is this: treat your database schema like you treat your code. Version it, test it, review changes carefully, and deploy incrementally. This mindset shift transforms database changes from a source of anxiety into just another part of your development workflow.

Start with simple table creations and work up to complex data transformations. Build your migration system incrementally too. Add features as you need them. The most important thing is to begin tracking your schema changes systematically. Once you have that foundation, everything else becomes much easier.

I still get nervous before big migrations, but now I have confidence in the process. The system handles the details, and I can focus on the actual changes needed. That peace of mind is worth every line of migration code I've written.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Build a Fault-Tolerant Stream Processing System in Go With Exactly-Once Guarantees

Nithin Bharadwaj — Sat, 18 Apr 2026 11:32:12 +0000

Building a system that handles a continuous flow of data, like clicks on a website or sensor readings, is a common challenge. You need to group this data into time windows, calculate running totals or averages, and guarantee that every piece of data is processed once, and only once, even if parts of the system fail. Let me walk you through how to build such a system in Go, piece by piece.

Think of a data stream as a never-ending conveyor belt. Items arrive at different times, sometimes out of order. Our job is to sort them, put them into time-based buckets (like "last 5 minutes"), do math on each bucket, and send the results onward, all while keeping perfect track of what we've already done.

The heart of our system is the StreamProcessor. It coordinates everything. When you create one, you give it some settings: how big its internal channels should be, how long to wait for late data, and where to save its progress.

config := ProcessorConfig{
    BufferSize:        10000,
    WatermarkDelay:    5 * time.Second,
    CheckpointInterval: 30 * time.Second,
}
processor := NewStreamProcessor(config)

Events flow into the system through the EventIngestor. This is the front door. Its first job is to deal with a messy reality: events don't always arrive in the order they were created. A network delay might hold up an earlier event, while a later one zips through.

To handle this, we use a concept called a watermark. It's like a moving finish line. We watch the timestamps on incoming events. The watermark is set to the highest timestamp we've seen, minus a waiting period (like 5 seconds). This waiting period is our allowance for late data. We promise not to finalize any calculations for a time period until we're reasonably sure no more data from that period will arrive. Events with a timestamp older than the current watermark are safe to process. Newer ones are held in a buffer.

func (wm *WatermarkManager) Update(eventTime int64) {
    wm.mu.Lock()
    defer wm.mu.Unlock()
    // ... find the maximum timestamp seen ...
    wm.currentWatermark = maxTs - int64(wm.maxDelay/time.Millisecond)
}

The WindowManager is where the core logic lives. A window is just a time range. A "tumbling window" might be every minute, non-overlapping. A "sliding window" might be a five-minute average, updated every minute. When an event is deemed ready (its timestamp is past the watermark), the manager figures out which windows it belongs to.

For each relevant window, we need an Aggregator. This is a simple object that knows how to update a calculation. If we're counting events, the aggregator holds a number and adds one. If we're summing a value, it holds a running total.

type CountAggregator struct {
    count int64
}

func (ca *CountAggregator) Aggregate(event StreamEvent) (interface{}, error) {
    atomic.AddInt64(&ca.count, 1)
    return ca.count, nil
}

The magic of "exactly-once" processing hinges on two things: remembering what you've done and being able to restart from a known good point. This is the job of the StateStore and the Checkpointer.

Every event has a unique ID. Before we even look at an event's data, we ask the StateStore: "Have I seen this ID before?" The store checks a fast cache and, if needed, a persistent database. If the answer is yes, we skip the event. This handles duplicates that might be sent if a producer retries.

func (sp *StreamProcessor) processEvents(ctx context.Context) error {
    for {
        select {
        case event := <-sp.ingestor.orderedChan:
            // The critical check
            if processed := sp.stateStore.IsProcessed(event.ID); processed {
                continue // Skip duplicate
            }
            // ... process the event ...
            // Mark it as done
            sp.stateStore.MarkProcessed(event.ID, event.Timestamp)
        }
    }
}

But what if our entire program crashes? We don't want to lose the results of all the windows we've been calculating. This is where checkpoints come in. On a regular schedule (e.g., every 30 seconds), the Checkpointer tells every component to pause and take a snapshot.

It captures everything: the state of every aggregator in every window, the current watermark, and the list of processed event IDs. It writes this snapshot durably—to disk or a database. This snapshot is a checkpoint.

func (c *Checkpointer) createCheckpoint(store *StateStore) {
    checkpoint := Checkpoint{
        ID:        ksuid.New().String(), // A unique ID
        Timestamp: time.Now().UnixNano(),
        State:     store.Snapshot(), // Grab all state
    }
    store.checkpoints[checkpoint.ID] = checkpoint
    c.commitCheckpoint(store, checkpoint)
}

When our processor starts up, its first step is to look for the latest successful checkpoint. If it finds one, it loads all the saved state. The aggregators are restored with their previous counts, the watermark is reset, and the list of processed IDs is reloaded. Then, it starts reading the event stream from the point after that checkpoint. It will re-process some events, but the duplicate detection in the StateStore will filter them out. The result is that the system's output is exactly as if it had run continuously without failing.

Let's look at what happens to a single event, step-by-step.

Arrival: An event {ID: "abc123", Timestamp: 1625097600000, Data: {value: 42}} is sent to processor.Process().
Buffering: The EventIngestor receives it, updates the watermark, and puts it in a buffer if its timestamp is too new.
Ordering: Once the watermark advances past the event's timestamp, it's pushed into the ordered channel.
Deduplication: The main loop receives it and asks the StateStore if "abc123" was processed. It hasn't been, so we continue.
Window Assignment: The WindowManager checks all active windows. Does a 1-minute tumbling window for [1625097600000, 1625097660000) exist? It creates one if needed.
Aggregation: The manager finds the SumAggregator for that window and calls agg.Aggregate(event). The aggregator adds 42 to its internal total.
State Save: The new state of the aggregator (the new sum) is saved to the StateStore with a key like "window_sum_1min:1625097600000".
Completion: The event's ID "abc123" is marked as processed in the state store.
Emission: If this event caused the window to be complete (e.g., the watermark passed the window's end time), the window triggers. It takes the final result from the aggregator and sends it out as output.

All of this happens concurrently. The ingestor, the watermark updater, the window manager, and the checkpointer all run in their own goroutines, communicating through channels. This is where Go's concurrency model shines. We use mutexes (sync.RWMutex) sparingly, only to protect specific maps or slices when multiple goroutines might access them.

A real implementation needs a solid backend for the StateStore. In-memory maps work for testing, but for production, you need persistence. You could use embedded databases like RocksDB or Badger. They store data locally on disk and are very fast for key-value lookups, which is most of what we do.

// A simplified RocksDB backend
type RocksDBBackend struct {
    db *gorocksdb.DB
}

func (rb *RocksDBBackend) Put(key string, value []byte) error {
    wb := gorocksdb.NewWriteBatch()
    defer wb.Destroy()
    wb.Put([]byte(key), value)
    return rb.db.Write(defaultWriteOptions, wb)
}

Performance tuning is crucial. The size of the input channel buffer affects backpressure. If it fills up, producers will block. The watermark delay is a trade-off: a longer delay means better handling of late events, but it also increases the latency of your results. The checkpoint interval is a trade-off between recovery time (shorter interval) and performance overhead (longer interval).

You'll want to add metrics to every part. How many events per second are you receiving? What's the average processing latency? How far behind the current time is your watermark? This data is vital for operating the system in production.

type StreamMetrics struct {
    EventsReceived      uint64
    EventsProcessed     uint64
    DuplicateEvents     uint64
    WatermarkLag        int64 // in milliseconds
}

In a distributed setup, you would run multiple instances of this processor. You'd need a way to partition your event stream, perhaps by a key in the data. Each instance would be responsible for a subset of the keys. The checkpointing becomes more complex, requiring a distributed coordination system like Apache ZooKeeper or etcd to ensure all instances snapshot their state consistently.

Building this might seem daunting, but by breaking it down into these components—ingestion with watermarks, windowed aggregation, stateful deduplication, and periodic checkpointing—you create a system that is both powerful and understandable. It handles the inherent disorder of real-world data and provides the strong guarantees necessary for accurate, reliable results. You start with events on a conveyor belt and end with timely, correct insights, no matter what happens along the way.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How JavaScript Signals and Fine-Grained Reactivity Make Your Apps Faster

Nithin Bharadwaj — Sat, 18 Apr 2026 11:18:44 +0000

Let’s talk about state in JavaScript applications. If you’ve ever built something interactive, you know the challenge: when a piece of data changes, you need to update the right parts of your interface, and you need to do it efficiently. For a long time, this meant a lot of manual work or using systems that often re-rendered more than necessary, which can slow things down.

There’s a different way to think about this problem. Imagine if every individual piece of your application’s state could announce its own changes, and only the parts of your app that specifically care about that piece would listen and react. Nothing else would be disturbed. This is the core idea behind signals and fine-grained reactivity. It’s a shift from telling the entire UI to “check for changes” to having a direct, one-to-one conversation between data and its consumers.

I want to walk you through how this works, not just in theory but in practice. We’ll build up the concepts from the ground up, with code you can use and adapt. Think of it as learning a new set of tools that make your applications faster and your code simpler to reason about.

At the heart of this system is the signal. A signal is a container for a value with a simple superpower: it knows what other parts of the code are interested in its current value. When that value changes, it can notify just those interested parties.

Here is a basic way to create one. We start with a class that holds a value and keeps track of subscribers.

class SignalCore {
  constructor(value) {
    this._value = value;
    this._subscribers = new Set();
  }

  get value() {
    // Dependency tracking happens here
    return this._value;
  }

  set value(newValue) {
    if (Object.is(this._value, newValue)) return;
    this._value = newValue;
    this._notifySubscribers();
  }

  _notifySubscribers() {
    this._subscribers.forEach(subscriber => subscriber());
  }

  subscribe(fn) {
    this._subscribers.add(fn);
    return () => this._subscribers.delete(fn);
  }
}

This is the foundation. You create a signal with an initial value. When you get its .value, you’re reading it. When you set its .value, it compares the new value with the old one. If they’re different, it notifies everyone in its subscriber list by calling their functions.

This alone is useful, but the real magic starts when we add dependency tracking. We need a way for the signal to know who is reading it at any given moment, so it can add that reader to its subscriber list automatically. We do this by setting a global pointer to the currently running “computation” or effect.

Let’s modify our getter to use this system.

get value() {
  // If something is currently 'listening', track this signal as a dependency
  if (SignalCore._activeComputation) {
    this._subscribers.add(SignalCore._activeComputation);
  }
  return this._value;
}

We need a place to store that active computation. We can use a static property on the class.

SignalCore._activeComputation = null;

Now we can create our first reactive primitive: an effect. An effect is a function that runs and automatically re-runs whenever any signal it reads inside changes.

function createEffect(effectFn) {
  const execute = () => {
    // Set this effect as the active computation
    SignalCore._activeComputation = execute;
    try {
      effectFn();
    } finally {
      // Clean up after running
      SignalCore._activeComputation = null;
    }
  };
  // Run it once to set up initial dependencies
  execute();
}

Let’s see it in action with two signals.

const firstName = new SignalCore('John');
const lastName = new SignalCore('Doe');

createEffect(() => {
  console.log(`Hello, ${firstName.value} ${lastName.value}!`);
});
// Logs: Hello, John Doe!

firstName.value = 'Jane';
// Automatically logs: Hello, Jane Doe!

The effect ran once. It read both signals, so it became a subscriber to both. When firstName changed, the effect’s function was called again. This is fine-grained reactivity. The effect didn’t need to know what changed; the signals told it.

Our next technique is the computed signal. This is a signal whose value is derived from other signals. It should only re-calculate when one of its dependencies changes.

To build this, we need a more robust tracking system. Our SignalCore needs to track not just subscriber functions, but also which computed signals depend on it. Let’s update the class.

class SignalCore {
  constructor(value) {
    this._value = value;
    this._subscribers = new Set();
    this._dependencies = new Set(); // For computed signals that depend on this
  }

  get value() {
    if (SignalCore._activeComputation) {
      this._dependencies.add(SignalCore._activeComputation);
      SignalCore._activeComputation._dependencies.add(this);
    }
    return this._value;
  }

  set value(newValue) {
    if (Object.is(this._value, newValue)) return;
    this._value = newValue;
    this._notify();
  }

  _notify() {
    // Notify direct subscriber functions
    this._subscribers.forEach(sub => sub());
    // Notify any dependent computed signals
    this._dependencies.forEach(dep => dep._update());
  }

  subscribe(fn) {
    this._subscribers.add(fn);
    return () => this._subscribers.delete(fn);
  }
}

Now we can create a ComputedSignal class that extends this core. It won’t have a setter. Its value comes from a computation function.

class ComputedSignal extends SignalCore {
  constructor(computeFn) {
    super(undefined); // Start with no value
    this._computeFn = computeFn;
    this._isDirty = true; // Needs first calculation
    this._update(); // Calculate initial value
  }

  get value() {
    if (this._isDirty) {
      this._update();
    }
    // Use parent's getter to track dependencies of this computed signal
    return super.value;
  }

  _update() {
    // Clear old dependencies
    this._dependencies.forEach(dep => dep._dependencies.delete(this));
    this._dependencies.clear();

    // Run computation with this as the active context
    SignalCore._activeComputation = this;
    try {
      const newValue = this._computeFn();
      if (!Object.is(this._value, newValue)) {
        this._value = newValue;
        this._isDirty = false;
        // Notify anyone depending on this computed signal
        this._notify();
      }
    } finally {
      SignalCore._activeComputation = null;
    }
  }

  // Called when a dependency notifies us
  _markDirty() {
    if (!this._isDirty) {
      this._isDirty = true;
      // We don't recalculate yet, just notify our dependents
      this._notify();
    }
  }
}

Let’s create some helper functions to make this easier to use.

function createSignal(value) {
  return new SignalCore(value);
}

function createComputed(fn) {
  return new ComputedSignal(fn);
}

function createEffect(fn) {
  const effect = {
    _execute() {
      SignalCore._activeComputation = effect;
      try {
        fn();
      } finally {
        SignalCore._activeComputation = null;
      }
    }
  };
  effect._dependencies = new Set();
  effect._execute();
  // For simplicity, we don't handle cleanup here yet
}

Now, here’s a practical example. Imagine a shopping cart.

const itemCount = createSignal(2);
const itemPrice = createSignal(25.99);

const subtotal = createComputed(() => itemCount.value * itemPrice.value);
const tax = createComputed(() => subtotal.value * 0.08);
const total = createComputed(() => subtotal.value + tax.value);

createEffect(() => {
  console.log(`Total to pay: $${total.value.toFixed(2)}`);
});
// Logs: Total to pay: $56.14

itemCount.value = 3;
// Automatically re-computes and logs: Total to pay: $84.21

Only the necessary computations happen. When itemCount changes, subtotal, tax, and total are recalculated in the right order. Our effect runs once at the end.

A common problem in UI updates is something called a "glitch." This happens when you read a computed signal that is in a temporarily inconsistent state because its dependencies are mid-update. Our current setup might have this issue if we update multiple signals at once.

We can solve this with a fourth technique: batching updates. The goal is to postpone all notifications until a batch of changes is complete, then notify once with all final values.

We’ll create a simple batch manager.

const batchStack = [];

function batch(callback) {
  batchStack.push(new Set());
  try {
    callback();
  } finally {
    const updates = batchStack.pop();
    // If we're not in a nested batch, flush updates
    if (batchStack.length === 0) {
      updates.forEach(signal => signal._notify());
    } else {
      // Otherwise, add to parent batch
      updates.forEach(signal => batchStack[batchStack.length - 1].add(signal));
    }
  }
}

Our signals need to cooperate. When a signal’s value is set inside a batch, it should schedule its notification instead of doing it immediately.

class SignalCore {
  // ... previous code ...

  set value(newValue) {
    if (Object.is(this._value, newValue)) return;
    this._value = newValue;

    if (batchStack.length > 0) {
      // Schedule for later
      batchStack[batchStack.length - 1].add(this);
    } else {
      this._notify();
    }
  }
}

Now you can group changes.

batch(() => {
  itemCount.value = 10;
  itemPrice.value = 19.99;
});
// The effect runs only once, after both updates.

This prevents intermediate states from causing unnecessary work or visual flicker.

The fifth technique is about structure. In a real app, you don’t have just a few loose signals. You have a state tree. We can create a store that organizes signals and provides methods to update them predictably.

Here’s a simple store pattern.

class SignalStore {
  constructor(initialState) {
    this._state = {};
    this._signals = {};

    Object.keys(initialState).forEach(key => {
      this._signals[key] = createSignal(initialState[key]);
      // Define a getter for easy access
      Object.defineProperty(this._state, key, {
        get: () => this._signals[key].value,
        enumerable: true
      });
    });
  }

  get(key) {
    return this._state[key];
  }

  getSignal(key) {
    return this._signals[key];
  }

  set(key, value) {
    if (this._signals[key]) {
      this._signals[key].value = value;
    }
  }

  batchUpdate(updater) {
    batch(() => updater(this));
  }
}

Use it like this.

const userStore = new SignalStore({
  name: 'Alice',
  age: 30,
  preferences: { theme: 'light' }
});

const userName = userStore.getSignal('name');
createEffect(() => {
  document.title = `Profile: ${userName.value}`;
});

userStore.batchUpdate((store) => {
  store.set('name', 'Bob');
  store.set('age', 31);
});

This gives you a central place for state that is still built from fine-grained signals.

The sixth technique integrates this with the DOM. We want our UI to update when signals change. We can create a small library to bind signals to DOM elements.

class ReactiveDOM {
  bindText(element, signal) {
    const update = () => element.textContent = signal.value;
    signal.subscribe(update);
    update(); // Set initial value
  }

  bindAttribute(element, attrName, signal) {
    const update = () => element.setAttribute(attrName, signal.value);
    signal.subscribe(update);
    update();
  }

  createReactiveElement(tag, props, ...children) {
    const el = document.createElement(tag);
    for (const [key, value] of Object.entries(props || {})) {
      if (value && typeof value === 'object' && 'value' in value) {
        // It's a signal
        this.bindAttribute(el, key, value);
      } else {
        el.setAttribute(key, value);
      }
    }
    children.forEach(child => {
      if (typeof child === 'string') {
        el.appendChild(document.createTextNode(child));
      } else if (child && typeof child === 'object' && 'value' in child) {
        // It's a signal for text content
        this.bindText(el, child);
      } else if (child instanceof HTMLElement) {
        el.appendChild(child);
      }
    });
    return el;
  }
}

Let’s build a counter UI.

const dom = new ReactiveDOM();
const count = createSignal(0);

const button = dom.createReactiveElement('button', {
  onclick: () => count.value++
}, 'Count: ', count);

document.body.appendChild(button);

Every time you click, the count signal updates, and the button’s text updates automatically. Only that specific text node is touched by the DOM.

The seventh technique deals with a common need: managing side effects and cleanup. Our simple createEffect needs improvement. When a dependency changes, we should clean up the old effect before running the new one, and we should clean up when the effect is no longer needed.

Let’s build a more robust effect manager.

function createEffect(effectFn) {
  let cleanupFn;
  let isDisposed = false;

  const execute = () => {
    // Clean up previous effect
    if (cleanupFn) {
      cleanupFn();
      cleanupFn = null;
    }

    // Set as active computation
    SignalCore._activeComputation = execute;
    try {
      // The effect can return a cleanup function
      cleanupFn = effectFn() || null;
    } finally {
      SignalCore._activeComputation = null;
    }
  };

  // Run once to start
  execute();

  // Return a dispose function
  return () => {
    if (isDisposed) return;
    isDisposed = true;
    if (cleanupFn) cleanupFn();
    // Remove this effect from all its dependencies
    execute._dependencies.forEach(signal => {
      signal._dependencies.delete(execute);
      signal._subscribers.delete(execute);
    });
  };
}

This pattern is useful for setting up and tearing down event listeners, subscriptions, or intervals.

const timer = createSignal(0);

const disposeEffect = createEffect(() => {
  const interval = setInterval(() => {
    timer.value++;
  }, 1000);

  // Cleanup function
  return () => clearInterval(interval);
});

// Later, to stop everything
// disposeEffect();

The eighth technique is about scalability and developer experience. As your app grows, you’ll want tools for debugging and inspection. We can add a simple devtools hook.

We can instrument our SignalCore to log changes.

class SignalCore {
  constructor(value, name) {
    this._value = value;
    this._subscribers = new Set();
    this._dependencies = new Set();
    this._name = name; // For debugging

    if (typeof window !== 'undefined' && window.__SIGNAL_DEVTOOLS__) {
      window.__SIGNAL_DEVTOOLS__.registerSignal(this);
    }
  }

  set value(newValue) {
    const oldValue = this._value;
    if (Object.is(oldValue, newValue)) return;
    this._value = newValue;

    // Notify devtools
    if (window.__SIGNAL_DEVTOOLS__) {
      window.__SIGNAL_DEVTOOLS__.logUpdate(this, oldValue, newValue);
    }

    // ... rest of setter logic ...
  }
}

You could then implement a simple devtools panel that listens to this global hook and displays a log of signal changes, helping you trace the flow of data.

Bringing it all together, these eight techniques—basic signals, dependency tracking, effects, computed values, batching, structured stores, DOM binding, and managed effects—form a complete foundation for fine-grained reactivity.

This approach might seem different at first, but it directly addresses the performance and complexity issues of older models. Instead of diffing a large virtual tree to guess what changed, you know exactly what changed because the signals tell you.

The code examples here are building blocks. You can start small, with just a few signals and effects, and gradually incorporate more patterns as your application demands. The key insight is that by making state itself observable and reactive, you shift the burden of update coordination from the developer to the system, leading to code that is often simpler and applications that are consistently faster.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Use Python to Control Real-World Hardware: Sensors, Motors, and IoT Devices

Nithin Bharadwaj — Sat, 18 Apr 2026 11:03:43 +0000

Python has become my secret weapon for making computers interact with the real world. It started for me with a simple goal: make a light blink. What seemed like a basic task opened a door to controlling robots, reading environmental data, and building devices that respond to physical touch. If you've ever wondered how software can affect hardware—like telling a motor to spin or reading a temperature sensor—Python is one of the most approachable ways to start.

I think of Python as a translator. My computer speaks in high-level, abstract code, but hardware components like sensors and motors speak in electrical signals—voltages, currents, and precise timing. Python, with its vast collection of libraries, acts as the middle layer that converts my simple commands into the low-level signals the hardware understands. This lets me focus on what I want the system to do, rather than getting lost in complex electronic details.

Let's begin with the most fundamental concept: talking directly to the pins on a board. Many small computers, like the Raspberry Pi, have a set of physical pins called GPIO, or General Purpose Input/Output pins. Each pin is like a tiny switch or a sensor port I can control from my code. I can set a pin to be an output to send power (to light an LED) or an input to listen for a signal (like a button being pressed).

Here is a practical class I often use as a foundation. It wraps the basic operations so I can manage multiple pins without repeating setup code.

import RPi.GPIO as GPIO
import time
from typing import List, Optional

class GPIOManager:
    def __init__(self, mode=GPIO.BCM):
        GPIO.setmode(mode)
        GPIO.setwarnings(False)
        self.pin_modes = {}

    def setup_pin(self, pin: int, mode: str, pull_up_down: Optional[str] = None):
        """Set a pin as either input or output."""
        if mode.upper() == 'OUT':
            GPIO.setup(pin, GPIO.OUT)
            self.pin_modes[pin] = 'OUT'
        elif mode.upper() == 'IN':
            if pull_up_down:
                if pull_up_down.upper() == 'UP':
                    GPIO.setup(pin, GPIO.IN, pull_up_down=GPIO.PUD_UP)
                elif pull_up_down.upper() == 'DOWN':
                    GPIO.setup(pin, GPIO.IN, pull_up_down=GPIO.PUD_DOWN)
            else:
                GPIO.setup(pin, GPIO.IN)
            self.pin_modes[pin] = 'IN'
        else:
            raise ValueError(f"Mode must be 'IN' or 'OUT'. Got: {mode}")

    def write_pin(self, pin: int, state: bool):
        """Send a HIGH (True) or LOW (False) signal to an output pin."""
        if self.pin_modes.get(pin) != 'OUT':
            raise ValueError(f"Pin {pin} is not an output.")
        GPIO.output(pin, GPIO.HIGH if state else GPIO.LOW)

    def read_pin(self, pin: int) -> bool:
        """Check if an input pin is receiving a HIGH (True) or LOW (False) signal."""
        if self.pin_modes.get(pin) != 'IN':
            raise ValueError(f"Pin {pin} is not an input.")
        return GPIO.input(pin) == GPIO.HIGH

    def pwm_control(self, pin: int, frequency: float, duty_cycle: float):
        """Create a Pulse Width Modulation signal. Great for dimming LEDs or controlling motor speed."""
        self.setup_pin(pin, 'OUT')
        pwm = GPIO.PWM(pin, frequency)
        pwm.start(duty_cycle)
        return pwm

    def monitor_button(self, pin: int, callback, bounce_time: int = 200):
        """Watch a button pin. The 'bounce_time' stops false triggers from physical switch jitter."""
        self.setup_pin(pin, 'IN', pull_up_down='UP')
        GPIO.add_event_detect(pin, GPIO.FALLING, callback=callback, bouncetime=bounce_time)

    def cleanup(self):
        """Always run this when your program ends to reset the pins safely."""
        GPIO.cleanup()

# Here's how I would use it in a simple project:
try:
    gpio = GPIOManager()

    # Setup an LED on pin 17
    led_pin = 17
    gpio.setup_pin(led_pin, 'OUT')

    print("Blinking the LED 5 times...")
    for i in range(5):
        gpio.write_pin(led_pin, True)  # Turn on
        time.sleep(0.5)
        gpio.write_pin(led_pin, False) # Turn off
        time.sleep(0.5)

    # Use PWM to fade an LED
    print("Now fading an LED with PWM...")
    pwm = gpio.pwm_control(18, 100, 50)  # Pin 18, 100Hz, start at 50% brightness
    time.sleep(2)
    pwm.ChangeDutyCycle(25)  # Dim to 25%
    time.sleep(2)
    pwm.stop()

finally:
    gpio.cleanup()  # This is crucial!

This basic control is powerful, but many interesting components, like temperature sensors or screen displays, need a more structured conversation. They use special communication protocols. I see these as languages for chips to talk to each other. Two of the most common are I2C and SPI.

I2C is like a small office meeting. There's a manager (the main computer) and several employees (sensor chips) all connected to the same two wires. The manager calls out an address to speak to one employee at a time. It's great for connecting many simple devices over short distances. SPI is faster and more like a direct phone line. It uses more wires but allows for full-speed, simultaneous data exchange, which is ideal for things like high-resolution analog sensors or memory chips.

Let me show you how I handle these protocols. The following code manages I2C communication and even includes a driver for a common temperature and pressure sensor, the BMP280.

import smbus2
import struct
import time

class I2CManager:
    def __init__(self, bus_number: int = 1):
        self.bus = smbus2.SMBus(bus_number)

    def scan_devices(self):
        """Find all I2C devices connected to the bus."""
        found = []
        for address in range(0x03, 0x78):
            try:
                self.bus.read_byte(address)
                found.append(hex(address))
            except:
                pass
        return found

    def read_register(self, address: int, register: int) -> int:
        """Read one byte of data from a specific register on a device."""
        return self.bus.read_byte_data(address, register)

    def write_register(self, address: int, register: int, value: int):
        """Write one byte of data to a specific register."""
        self.bus.write_byte_data(address, register, value)

class BMP280_Sensor:
    """A driver for the BMP280 temperature and barometric pressure sensor."""

    def __init__(self, i2c_bus: I2CManager, address: int = 0x76):
        self.i2c = i2c_bus
        self.address = address
        self.calibration = {}
        self._setup_sensor()

    def _setup_sensor(self):
        """Read the unique calibration data from the chip and configure it."""
        # First, check we're talking to the right chip
        chip_id = self.i2c.read_register(self.address, 0xD0)
        if chip_id != 0x58:
            raise RuntimeError(f"Wrong chip. Expected BMP280 (ID 0x58), got {hex(chip_id)}")

        # The chip stores unique calibration numbers. We must read them.
        cal_data = self.i2c.read_block(self.address, 0x88, 24)

        # Convert bytes into usable numbers (this is specific to the BMP280)
        self.calibration['dig_T1'] = cal_data[0] | (cal_data[1] << 8)
        self.calibration['dig_T2'] = self._to_signed(cal_data[2] | (cal_data[3] << 8))
        self.calibration['dig_T3'] = self._to_signed(cal_data[4] | (cal_data[5] << 8))
        # ... (similar for pressure calibration values P1-P9)

        # Tell the sensor to start taking normal measurements
        self.i2c.write_register(self.address, 0xF4, 0x27)
        time.sleep(0.005)

    def _to_signed(self, value: int) -> int:
        """Helper to convert raw bytes to a signed integer."""
        if value & 0x8000:
            return value - 0x10000
        return value

    def read(self) -> dict:
        """Trigger a measurement and return calculated temperature and pressure."""
        # Read 6 bytes of raw data
        data = self.i2c.read_block(self.address, 0xF7, 6)
        temp_raw = (data[3] << 12) | (data[4] << 4) | (data[5] >> 4)
        press_raw = (data[0] << 12) | (data[1] << 4) | (data[2] >> 4)

        # Use the calibration data to convert raw numbers to real values
        temperature = self._calculate_temperature(temp_raw)
        pressure = self._calculate_pressure(press_raw, temperature)

        return {'temperature_c': temperature, 'pressure_hPa': pressure}

    def _calculate_temperature(self, raw_temp: int) -> float:
        """Apply the calibration formula from the BMP280 datasheet."""
        dig = self.calibration
        var1 = ((raw_temp / 16384.0) - (dig['dig_T1'] / 1024.0)) * dig['dig_T2']
        var2 = ((raw_temp / 131072.0) - (dig['dig_T1'] / 8192.0)) * dig['dig_T3']
        t_fine = var1 + var2
        return t_fine / 5120.0

    def _calculate_pressure(self, raw_press: int, t_fine: float) -> float:
        """Apply the more complex pressure calibration formula."""
        dig = self.calibration
        # ... (detailed compensation algorithm)
        return pressure_value  # in hPa

# Using the sensor
print("I2C Sensor Example")
print("=" * 40)
try:
    i2c = I2CManager(1)
    print(f"Devices on bus: {i2c.scan_devices()}")

    sensor = BMP280_Sensor(i2c)
    reading = sensor.read()
    print(f"Temperature: {reading['temperature_c']:.2f} °C")
    print(f"Pressure: {reading['pressure_hPa']:.2f} hPa")

except Exception as e:
    print(f"Error: {e}")

Sometimes, I need to connect to devices that act like old-school terminals, sending streams of text data. This is serial communication, often over a USB cable or direct wires called UART. It's how I talk to GPS modules, old printers, or even another computer. The pyserial library makes this straightforward.

A common task is reading from a GPS receiver, which sends data in a standard text format called NMEA sentences. Here’s how I structure code to handle that continuous stream of data.

import serial
import serial.tools.list_ports
import threading
import time

class SerialMonitor:
    def __init__(self):
        self.connection = None
        self.monitor_thread = None
        self.stop_signal = threading.Event()

    def find_ports(self):
        """Show all serial ports (like USB adapters) available on the computer."""
        return [p.device for p in serial.tools.list_ports.comports()]

    def connect(self, port_name: str, baud_rate: int = 9600):
        """Open a connection to a serial device."""
        self.connection = serial.Serial(
            port=port_name,
            baudrate=baud_rate,
            timeout=2.0
        )
        print(f"Connected to {port_name} at {baud_rate} baud.")

    def start_listening(self, data_handler):
        """Start a background thread that constantly listens for incoming data."""
        if not self.connection:
            raise RuntimeError("Not connected to a port.")

        self.stop_signal.clear()

        def listen_loop():
            buffer = ""
            while not self.stop_signal.is_set():
                if self.connection.in_waiting:
                    # Read bytes and decode to text
                    text = self.connection.read(self.connection.in_waiting).decode('ascii', errors='ignore')
                    buffer += text
                    # Split by newline to get complete sentences
                    while '\\n' in buffer:
                        line, buffer = buffer.split('\\n', 1)
                        data_handler(line.strip())  # Send the line to our handler function
                time.sleep(0.01)

        self.monitor_thread = threading.Thread(target=listen_loop)
        self.monitor_thread.start()

    def stop_listening(self):
        """Safely stop the background listening thread."""
        self.stop_signal.set()
        if self.monitor_thread:
            self.monitor_thread.join(timeout=1.0)

    def send_command(self, command: str):
        """Send a text command to the connected device."""
        if self.connection:
            self.connection.write((command + '\\r\\n').encode())

    def close(self):
        """Close the serial port."""
        self.stop_listening()
        if self.connection:
            self.connection.close()

# Example: Parsing GPS NMEA sentences
def parse_gps_sentence(sentence: str):
    """A simple parser for common GPS data sentences."""
    if not sentence.startswith('$'):
        return None
    parts = sentence.split(',')

    if sentence.startswith('$GPGGA'):  # Basic location and fix data
        if len(parts) > 9:
            return {
                'type': 'location',
                'time': parts[1],
                'lat': parts[2] + parts[3],  # Latitude and direction (N/S)
                'lon': parts[4] + parts[5],  # Longitude and direction (E/W)
                'altitude': parts[9] + parts[10]
            }
    elif sentence.startswith('$GPRMC'):  # Recommended minimum data
        if len(parts) > 7:
            return {
                'type': 'movement',
                'speed_knots': parts[7],
                'course': parts[8]
            }
    return None

# Using the monitor
print("Serial GPS Monitor Example")
print("=" * 40)

monitor = SerialMonitor()
ports = monitor.find_ports()
print(f"Available ports: {ports}")

if ports:
    # In a real scenario, you would connect to the correct port, e.g., ports[0]
    # monitor.connect(ports[0], 9600)

    def handle_incoming_data(data_line):
        parsed = parse_gps_sentence(data_line)
        if parsed:
            print(f"GPS Data: {parsed}")

    # monitor.start_listening(handle_incoming_data)
    # time.sleep(30)  # Listen for 30 seconds
    # monitor.stop_listening()
    # monitor.close()

    print("(Serial simulation skipped for example.)")

Moving beyond simple on/off signals, I often need to control the position of something, like a robot arm or a camera panning mechanism. This is where servo motors come in. They don't just spin; they move to a specific angle based on the signal I send. The signal is a special type of PWM pulse. Here's a practical servo controller class.

import RPi.GPIO as GPIO
import time

class ServoController:
    def __init__(self, control_pin: int):
        self.pin = control_pin
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(self.pin, GPIO.OUT)

        # Standard servo PWM frequency is 50Hz (20ms period)
        self.pwm = GPIO.PWM(self.pin, 50)
        self.pwm.start(0)  # Start with 0 duty cycle
        self._current_angle = 90  # Assume starting at center

    def set_angle(self, angle: float):
        """Move the servo to a specific angle (typically 0 to 180 degrees)."""
        if angle < 0 or angle > 180:
            raise ValueError("Angle must be between 0 and 180 degrees.")

        # The magic formula: Duty Cycle = 2.5 + (angle / 18)
        # This gives a pulse width between 0.5ms (0°) and 2.5ms (180°)
        duty_cycle = 2.5 + (angle / 18.0)
        self.pwm.ChangeDutyCycle(duty_cycle)
        self._current_angle = angle
        time.sleep(0.5)  # Give the servo time to move
        self.pwm.ChangeDutyCycle(0)  # Stop sending pulse to avoid jitter

    def sweep(self, start_angle=0, end_angle=180, step=1, delay=0.02):
        """Smoothly sweep the servo between two angles."""
        for angle in range(start_angle, end_angle + step, step):
            self.set_angle(angle)
            time.sleep(delay)

    def cleanup(self):
        """Stop PWM and cleanup GPIO."""
        self.pwm.stop()
        GPIO.cleanup()

# Example usage
print("Servo Motor Control")
print("=" * 40)
try:
    servo = ServoController(control_pin=12)  # Assuming servo signal wire on GPIO12

    print("Moving to center (90°)")
    servo.set_angle(90)
    time.sleep(1)

    print("Moving to left (0°)")
    servo.set_angle(0)
    time.sleep(1)

    print("Sweeping from left to right...")
    servo.sweep(0, 180, 5, 0.05)

finally:
    servo.cleanup()

Many modern devices connect via USB, appearing as serial ports, storage drives, or Human Interface Devices (HID) like keyboards. Python can interact with these, too. For example, I can use the pyusb library to communicate with a custom USB device that doesn't have a standard driver.

import usb.core
import usb.util

class USBDeviceFinder:
    """Finds and interacts with a USB device based on its Vendor ID and Product ID."""

    def __init__(self, vendor_id: int, product_id: int):
        self.vid = vendor_id
        self.pid = product_id
        self.device = None

    def find(self):
        """Locate the USB device."""
        self.device = usb.core.find(idVendor=self.vid, idProduct=self.pid)
        if self.device is None:
            raise ValueError("Device not found. Check connections.")
        print(f"Found device: {self.device}")
        return self.device

    def detach_kernel_driver(self, interface=0):
        """Sometimes the OS claims the device. This tells the OS to let go."""
        if self.device.is_kernel_driver_active(interface):
            try:
                self.device.detach_kernel_driver(interface)
                print("Detached kernel driver.")
            except usb.core.USBError as e:
                raise RuntimeError(f"Could not detach kernel driver: {e}")

    def write_data(self, endpoint: int, data: bytes):
        """Send data to a specific endpoint on the device."""
        # Endpoint 0x01 is typically a "bulk out" endpoint for sending data.
        bytes_written = self.device.write(endpoint, data)
        print(f"Sent {bytes_written} bytes.")
        return bytes_written

    def read_data(self, endpoint: int, size: int):
        """Read data from a specific endpoint."""
        # Endpoint 0x81 is typically a "bulk in" endpoint for receiving data.
        data = self.device.read(endpoint, size)
        print(f"Received {len(data)} bytes.")
        return data.tobytes()

# Note: This requires running with appropriate permissions (often as root/sudo).
print("USB Device Interaction Concept")
print("=" * 40)
print("This code shows the structure for talking to a custom USB device.")
print("Actual Vendor and Product IDs are needed to run it.")
# Example: finder = USBDeviceFinder(0x1234, 0xabcd)
# finder.find()

For projects that connect to the internet, like a weather station that posts data online, I use IoT protocols. MQTT is a popular, lightweight standard for this. It works on a publish/subscribe model: my device "publishes" temperature readings to a topic, and another program "subscribes" to that topic to receive them. Here's a basic setup using the paho-mqtt library.

import paho.mqtt.client as mqtt
import time
import json

class IoT_Reporter:
    def __init__(self, broker_address="test.mosquitto.org", client_id=""):
        self.broker = broker_address
        self.client = mqtt.Client(client_id if client_id else mqtt.base62())
        self.client.on_connect = self._on_connect_callback
        self.client.on_message = self._on_message_callback

    def _on_connect_callback(self, client, userdata, flags, rc):
        """This runs when the connection to the broker is established."""
        if rc == 0:
            print("Connected successfully to MQTT broker!")
        else:
            print(f"Failed to connect, return code {rc}")

    def _on_message_callback(self, client, userdata, message):
        """This runs when a message is received on a subscribed topic."""
        print(f"Message on [{message.topic}]: {message.payload.decode()}")

    def connect(self):
        """Start the connection to the broker."""
        self.client.connect(self.broker, 1883, 60)
        self.client.loop_start()  # Start a background network thread

    def publish_sensor_data(self, topic: str, sensor_name: str, value: float, unit: str):
        """Format and send sensor data as a JSON message."""
        payload = {
            "sensor": sensor_name,
            "value": value,
            "unit": unit,
            "timestamp": time.time()
        }
        result = self.client.publish(topic, json.dumps(payload))
        if result.rc == mqtt.MQTT_ERR_SUCCESS:
            print(f"Published to {topic}")
        else:
            print(f"Publish failed: {result}")

    def subscribe(self, topic: str):
        """Listen for messages on a specific topic."""
        self.client.subscribe(topic)
        print(f"Subscribed to {topic}")

    def disconnect(self):
        """Cleanly disconnect from the broker."""
        self.client.loop_stop()
        self.client.disconnect()

# Example usage
print("MQTT IoT Reporting Example")
print("=" * 40)

reporter = IoT_Reporter("test.mosquitto.org")
reporter.connect()
time.sleep(1)  # Wait for connection

# Simulate publishing temperature every 5 seconds
try:
    for i in range(3):
        simulated_temp = 20.5 + i
        reporter.publish_sensor_data("home/livingroom/temperature", "bmp280", simulated_temp, "C")
        time.sleep(5)
finally:
    reporter.disconnect()

One of the most exciting areas is combining hardware control with computer vision. Using a library like OpenCV, I can make a camera feed control physical outputs. A classic beginner project is a simple color tracker that moves a servo to follow a colored object.

import cv2
import numpy as np
from ServoController import ServoController  # Using the class from earlier

class ColorTracker:
    def __init__(self, servo_pan_pin: int, servo_tilt_pin: int):
        # Initialize two servos: one for left-right (pan), one for up-down (tilt)
        self.servo_pan = ServoController(servo_pan_pin)
        self.servo_tilt = ServoController(servo_tilt_pin)
        self.cap = cv2.VideoCapture(0)

        # Define range of blue color in HSV (Hue, Saturation, Value)
        self.lower_blue = np.array([100, 150, 50])
        self.upper_blue = np.array([140, 255, 255])

        # Center position for the servos
        self.pan_angle = 90
        self.tilt_angle = 90
        self.servo_pan.set_angle(self.pan_angle)
        self.servo_tilt.set_angle(self.tilt_angle)

    def process_frame(self, frame):
        """Find the largest blue object in the frame and return its center."""
        # Convert from BGR to HSV color space
        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)

        # Create a mask: white for blue pixels, black for everything else
        mask = cv2.inRange(hsv, self.lower_blue, self.upper_blue)

        # Find contours (outlines) of the white blobs in the mask
        contours, _ = cv2.findContours(mask, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)

        if contours:
            # Get the largest contour
            largest_contour = max(contours, key=cv2.contourArea)
            # Calculate the center of this contour
            M = cv2.moments(largest_contour)
            if M["m00"] != 0:
                center_x = int(M["m10"] / M["m00"])
                center_y = int(M["m01"] / M["m00"])
                return (center_x, center_y), largest_contour
        return None, None

    def update_servos(self, center_x, center_y, frame_width, frame_height):
        """Adjust servo angles based on object position."""
        # Convert pixel position to angle (0-180 range)
        new_pan = 180 - (center_x / frame_width) * 180  # Invert so object left moves servo left
        new_tilt = (center_y / frame_height) * 180

        # Smooth the movement: only move if the change is significant
        if abs(new_pan - self.pan_angle) > 2:
            self.pan_angle = max(0, min(180, new_pan))
            self.servo_pan.set_angle(self.pan_angle)
        if abs(new_tilt - self.tilt_angle) > 2:
            self.tilt_angle = max(0, min(180, new_tilt))
            self.servo_tilt.set_angle(self.tilt_angle)

    def run(self):
        """Main loop to capture video and track."""
        print("Starting color tracker. Press 'q' to quit.")
        while True:
            ret, frame = self.cap.read()
            if not ret:
                break

            frame = cv2.flip(frame, 1)  # Mirror the image
            height, width, _ = frame.shape

            center, contour = self.process_frame(frame)

            if center:
                # Draw a circle and rectangle on the object
                cv2.circle(frame, center, 10, (0, 255, 0), -1)
                x, y, w, h = cv2.boundingRect(contour)
                cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 255, 0), 2)

                # Move the servos to follow it
                self.update_servos(center[0], center[1], width, height)

            # Show the video feed
            cv2.imshow('Color Tracker', frame)

            if cv2.waitKey(1) & 0xFF == ord('q'):
                break

        self.cap.release()
        cv2.destroyAllWindows()
        self.servo_pan.cleanup()
        self.servo_tilt.cleanup()

# To run this:
# tracker = ColorTracker(servo_pan_pin=12, servo_tilt_pin=13)
# tracker.run()
print("Computer Vision Servo Tracker")
print("=" * 40)
print("This code combines OpenCV for seeing a blue object with servos to follow it.")

A critical thing I've learned is that Python is not always the fastest. For tasks that require microsecond precision, like generating a very specific digital signal, a compiled language like C might be better. However, for the vast majority of my projects—where events happen on a scale of milliseconds or seconds—Python is perfectly adequate. The key is to structure your code to avoid delays. Use threading for long tasks (like waiting for a sensor reading) so your main loop can keep checking buttons or updating displays.

Finally, safety is paramount. Always disconnect power when wiring. Use resistors with LEDs. Double-check pin assignments. A small mistake can let the magic smoke out of a component, ending its life. Start with simple examples, understand the flow of electricity and data, and build up to more complex systems. The goal is to create a dialog between the digital world of code and the physical world of materials and forces. Python provides a clear and powerful voice for that conversation.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

How to Integrate Rust FFI With C Libraries Without Rewriting Your Entire Codebase

Nithin Bharadwaj — Fri, 17 Apr 2026 08:56:54 +0000

Think of your code as a house. For years, you've built additions, rooms, and plumbing using a fast but sometimes risky method—let's call it the "C" method. It works, but occasionally, pipes leak or a floorboard gives way because of a small mistake. Now, you've discovered a new, stricter building code called Rust that prevents these accidents by design. You don't want to rebuild the entire house from scratch. Instead, you want to build a new, safer room and connect it securely to the old structure. That's what Rust's Foreign Function Interface, or FFI, is for. It's the secure doorway between your new, safe Rust code and the vast, existing world of C and C++ libraries.

I need to talk to the old house. In programming, languages like C expose their functions through something called an Application Binary Interface, or ABI. It's a rulebook for how functions are named, where arguments go in memory, and how to get a result back. Rust can speak this rulebook fluently. The key is the extern block. When I write extern "C", I'm telling Rust, "The function I'm about to describe is written in C, so use its rules to call it."

Let's start with the simplest door. Imagine a C library has a function that adds two numbers. To call it from Rust, I first declare its shape.

// Tell Rust we're using a type that matches C's 'int'
use std::os::raw::c_int;

// Declare the external C function. This is a promise; the actual code is elsewhere.
extern "C" {
    fn c_add(a: c_int, b: c_int) -> c_int;
}

fn main() {
    let result: c_int;

    // Calling a C function is inherently 'unsafe' because Rust can't check its rules.
    unsafe {
        result = c_add(10, 20);
    }

    println!("Result from C: {}", result); // Prints: Result from C: 30
}

See the unsafe block? This is Rust's most important safety feature in this context. It draws a bright red line. Everything inside that block is a promise from me, the programmer. I am promising Rust that I have read the C library's manual, I know how c_add behaves, and I am calling it correctly. Rust can't verify what happens in the C code, so it requires me to take explicit responsibility. This forces me to think carefully every time I cross this boundary.

Of course, we rarely just add numbers. We deal with strings, which are a famous source of problems. In C, a string is just a pointer to a sequence of characters ending with a zero byte (a 'null terminator'). Rust's strings are more complex and safe. To talk to C, we need a translator. That's where std::ffi::CString comes in.

Let's say we want to call the C standard library's strlen function, which counts the characters in a string.

use std::os::raw::c_int;
use std::ffi::CString;

extern "C" {
    // The 'strlen' function from the C standard library.
    // It takes a *const i8 (a raw pointer to a signed 8-bit integer, like a C 'char')
    // and returns a c_int (a C integer).
    fn strlen(s: *const i8) -> c_int;
}

fn main() {
    // Create a Rust String.
    let rust_string = String::from("Hello from Rust!");

    // Convert it to a CString. This allocates memory and adds the null terminator.
    // It can fail if our string has internal null bytes, hence the '.unwrap()'.
    let c_string = CString::new(rust_string).unwrap();

    let length: usize;

    unsafe {
        // Call the C function. We get the raw pointer from the CString with `.as_ptr()`.
        // We then cast the c_int result to Rust's usize.
        length = strlen(c_string.as_ptr()) as usize;
    }

    println!("The C function says the length is: {}", length);
}

This works, but we're still writing unsafe in our main function. A better pattern is to build a safe Rust wrapper around the unsafe C call. I try to hide the danger inside a well-lit, safe room.

use std::ffi::CString;
use std::os::raw::c_int;

extern "C" {
    fn strlen(s: *const i8) -> c_int;
}

// A safe wrapper function. It takes a normal &str, handles the conversion,
// performs the unsafe call, and returns a plain usize.
pub fn safe_strlen(input: &str) -> usize {
    let c_string = CString::new(input).expect("Failed to create CString");

    unsafe {
        strlen(c_string.as_ptr()) as usize
    }
}

fn main() {
    // Now, my main code is completely safe. No 'unsafe' keyword in sight.
    let my_text = "This is much nicer.";
    let len = safe_strlen(my_text);
    println!("Length: {}", len);
}

This is the core idea. The unsafe block is small, contained, and easy to review. The rest of my program uses the friendly safe_strlen function without a care in the world. I've built a safe bridge over the dangerous creek.

Data structures are trickier. If a C function expects a pointer to a struct, we need to make sure our Rust struct is laid out in memory exactly the same way. We use #[repr(C)] for this. It tells Rust's compiler to use the same layout rules as a C compiler.

Imagine a C library for shapes that has this struct:

// C Code
struct Point {
    int x;
    int y;
};

Here is how we mirror it in Rust and use it with a hypothetical C function print_point.

use std::os::raw::c_int;

// This attribute is crucial. It ensures the struct fields are in the same order
// and with the same alignment as the C compiler would use.
#[repr(C)]
pub struct Point {
    pub x: c_int,
    pub y: c_int,
}

extern "C" {
    // A C function that takes a pointer to a Point.
    fn print_point(p: *const Point);
}

fn main() {
    // We can create a Point just like any Rust struct.
    let my_point = Point { x: 42, y: 17 };

    unsafe {
        // We pass a raw pointer to our struct to the C function.
        print_point(&my_point as *const Point);
    }
    // The C code will receive the exact bytes it expects for 'x' and 'y'.
}

Memory management is where things get serious. Who owns the memory? Who frees it? This must be crystal clear. A common pattern is for a C function to return a pointer to heap-allocated memory. In Rust, we need to take that pointer, wrap it in a type that Rust understands, and ensure it gets freed with the correct allocator.

Suppose a C function create_greeting allocates and returns a new string.

use std::ffi::CStr;
use std::os::raw::c_char;

extern "C" {
    // This function returns a pointer to a new C string.
    // The documentation MUST say: "Caller must free with free_greeting."
    fn create_greeting(name: *const c_char) -> *mut c_char;
    fn free_greeting(s: *mut c_char);
}

// We create a Rust struct to own this C-allocated string.
pub struct Greeting {
    // This raw pointer is our link to the C-allocated memory.
    ptr: *mut c_char,
}

impl Greeting {
    pub fn new(name: &str) -> Option<Self> {
        let c_name = CString::new(name).ok()?;

        let ptr = unsafe { create_greeting(c_name.as_ptr()) };

        // Check if the C function returned a null pointer (indicating failure).
        if ptr.is_null() {
            None
        } else {
            Some(Greeting { ptr })
        }
    }

    // A method to read the greeting as a Rust string slice.
    pub fn as_str(&self) -> &str {
        unsafe {
            // Convert the raw pointer to a CStr (a view of a null-terminated string).
            let c_str = CStr::from_ptr(self.ptr);
            // Convert the CStr to a &str. This may fail if not valid UTF-8.
            c_str.to_str().unwrap_or("[Invalid UTF-8]")
        }
    }
}

// The critical part: implementing Drop.
// When a Greeting goes out of scope, this runs to free the C memory.
impl Drop for Greeting {
    fn drop(&mut self) {
        if !self.ptr.is_null() {
            unsafe { free_greeting(self.ptr) };
        }
    }
}

fn main() {
    // Now we can use it safely.
    let greeting = Greeting::new("Alice").expect("Failed to create greeting");
    println!("{}", greeting.as_str());
    // When 'greeting' goes out of scope at the end of main, 'drop' is called automatically,
    // and the C memory is freed. No memory leak.
}

This pattern is powerful. It uses Rust's ownership system to enforce the memory management rules of the C library. The Drop trait is our guarantee that cleanup will happen.

But what if the C code wants to call us? This is for callbacks. Many C libraries let you register a function pointer, which they will call later (think of an event handler). We can write a Rust function and pass it to C.

The rule is: the Rust function must be marked extern "C" so it follows C's calling rules. We also often use #[no_mangle] to tell Rust not to change the function's name, so the C code can find it.

use std::os::raw::c_int;

// This type alias defines a C function pointer type.
type Callback = extern "C" fn(data: c_int);

extern "C" {
    // A C function that registers our callback.
    fn register_callback(cb: Callback);
}

// Our Rust function that will be called from C.
// '#[no_mangle]' prevents name mangling.
// 'extern "C"' makes it use the C ABI.
#[no_mangle]
pub extern "C" fn my_rust_callback(value: c_int) {
    println!("C just called back into Rust with value: {}", value);
}

fn main() {
    // We pass the function pointer to the C library.
    unsafe {
        register_callback(my_rust_callback);
    }

    // Later, when the C library decides to, it will call 'my_rust_callback'.
}

In practice, you rarely write all these bindings by hand. It's tedious and error-prone. This is where the Rust ecosystem shines. The bindgen tool is a lifesaver. You give it a C header file, and it automatically generates the extern blocks, structs, and type definitions for you.

First, you add bindgen to your Cargo.toml as a build dependency. Then, you create a build.rs file. Cargo runs this script before compiling your main code.

Cargo.toml:

[package]
name = "my_ffi_project"
version = "0.1.0"

[build-dependencies]
bindgen = "0.68"

[dependencies]
libc = "0.2"

build.rs:

extern crate bindgen;

use std::env;
use std::path::PathBuf;

fn main() {
    // Tell Cargo to link against the C library `mylib`.
    println!("cargo:rustc-link-lib=mylib");

    // Set up bindgen.
    let bindings = bindgen::Builder::default()
        // The header file of the C library we're wrapping.
        .header("wrapper.h")
        // Tell bindgen to generate code for everything in the header.
        .generate()
        .expect("Unable to generate bindings");

    // Write the generated Rust bindings to a file in the output directory.
    let out_path = PathBuf::from(env::var("OUT_DIR").unwrap());
    bindings
        .write_to_file(out_path.join("bindings.rs"))
        .expect("Couldn't write bindings!");
}

Then, in your main Rust code, you include the generated file:

// Include the bindings generated during the build.
include!(concat!(env!("OUT_DIR"), "/bindings.rs"));

fn main() {
    // You can now directly use the functions and types from `mylib`!
    unsafe {
        // For example, if `wrapper.h` defined a function `c_start_engine`:
        // c_start_engine();
    }
}

This automation is a game-changer. It keeps your bindings perfectly synchronized with the C library's header. If the C library updates, you just re-run the build, and your Rust interface updates with it.

Let's walk through a more complete, practical example. Let's pretend we're integrating a simple C library for a lightbulb. The library is called libbulb.

C Library (bulb.h):

#ifndef BULB_H
#define BULB_H

typedef struct {
    int brightness;
    int is_on;
} Bulb;

Bulb* bulb_create(int initial_brightness);
void bulb_turn_on(Bulb* bulb);
void bulb_turn_off(Bulb* bulb);
void bulb_set_brightness(Bulb* bulb, int level);
int bulb_get_brightness(const Bulb* bulb);
void bulb_destroy(Bulb* bulb);

#endif

Our goal is to create a safe Rust API for this. We'll use bindgen to generate the raw bindings, then write our own safe wrapper.

After running bindgen (as shown above), we get a bindings.rs file. We don't need to look at its ugliness. Instead, we create a lib.rs that builds the safe bridge.

// lib.rs

// Include the raw, unsafe bindings.
mod ffi {
    include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
}

use std::ptr::NonNull;

// Our safe Rust wrapper for the Bulb.
pub struct Bulb {
    // We use NonNull to indicate this pointer is never null,
    // which is an invariant we must uphold.
    raw: NonNull<ffi::Bulb>,
}

impl Bulb {
    pub fn new(initial_brightness: i32) -> Option<Self> {
        // Call the C constructor.
        let raw_ptr = unsafe { ffi::bulb_create(initial_brightness) };

        // Convert to NonNull, checking for null (which indicates allocation failure).
        NonNull::new(raw_ptr).map(|ptr| Bulb { raw: ptr })
    }

    pub fn turn_on(&mut self) {
        unsafe { ffi::bulb_turn_on(self.raw.as_ptr()) }
    }

    pub fn turn_off(&mut self) {
        unsafe { ffi::bulb_turn_off(self.raw.as_ptr()) }
    }

    pub fn set_brightness(&mut self, level: i32) {
        unsafe { ffi::bulb_set_brightness(self.raw.as_ptr(), level) }
    }

    pub fn get_brightness(&self) -> i32 {
        unsafe { ffi::bulb_get_brightness(self.raw.as_ptr()) }
    }
}

// Implement Drop to ensure the C memory is freed.
impl Drop for Bulb {
    fn drop(&mut self) {
        unsafe { ffi::bulb_destroy(self.raw.as_ptr()) }
    }
}

// Main function to demonstrate usage.
fn main() {
    let mut my_bulb = Bulb::new(50).expect("Failed to create bulb");

    println!("Brightness: {}", my_bulb.get_brightness()); // 50

    my_bulb.turn_on();
    my_bulb.set_brightness(75);

    println!("Brightness now: {}", my_bulb.get_brightness()); // 75

    my_bulb.turn_off();
    // When `my_bulb` goes out of scope, its `drop` method is called,
    // and `bulb_destroy` is invoked on the C side.
}

This is the complete picture. We started with a raw, dangerous C interface. We used bindgen to get a raw Rust equivalent. Then, we constructed a neat, safe Rust struct that manages the underlying C object, following Rust's ownership and lifetime rules. The unsafe code is confined to a few lines in the wrapper's implementation. The user of our Bulb struct enjoys all the safety of Rust.

This approach is how large projects like Firefox integrate Rust. They don't rewrite the entire browser. They pick a component—a CSS parser, a video decoder—and write a new, safer version in Rust. They use FFI to slot this new component into the existing C++ architecture. Over time, the island of safe Rust grows, making the whole system more reliable without a risky, all-at-once rewrite.

It feels less like replacing a foundation and more like installing steel reinforcements, one beam at a time. You get the safety benefits where they matter most, without losing access to decades of tested, performant code. Rust's FFI isn't just a feature; it's a pragmatic strategy for building a safer future, one secure doorway at a time.

📘 Checkout my latest ebook for free on my channel!

Be sure to like, share, comment, and subscribe to the channel!

101 Books

Check out our book Golang Clean Code available on Amazon.

Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!

Our Creations

Be sure to check out our creations:

We are on Medium

Forem: Nithin Bharadwaj

**WebAssembly Is Turning Your Browser Into a Full Desktop Application Runtime**

101 Books

Our Creations

We are on Medium

How to Build a High-Performance GraphQL Gateway That Handles Real Production Traffic

Take these concepts, implement them in your language of choice, and watch your GraphQL performance transform. Your users will thank you, your database will thank you, and you'll sleep better knowing your API can handle whatever queries come its way.

101 Books

Our Creations

We are on Medium

Natural Language Processing with Python: 8 Text Analysis Techniques You Need to Know

101 Books

Our Creations

We are on Medium

# Rust and Formal Verification: How to Prove Your Code Actually Works

101 Books

Our Creations

We are on Medium

Building a Time-Series Storage Engine in Go That Handles Billions of Data Points

101 Books

Our Creations

We are on Medium

How to Build Service Discovery and Load Balancing for Distributed Systems in Go

101 Books

Our Creations

We are on Medium

How to Reverse Engineer Any Database Structure Using Python

101 Books

Our Creations

We are on Medium

How to Build a Bulletproof Database Migration System in JavaScript

I still get nervous before big migrations, but now I have confidence in the process. The system handles the details, and I can focus on the actual changes needed. That peace of mind is worth every line of migration code I've written.

101 Books

Our Creations

We are on Medium

How to Build a Fault-Tolerant Stream Processing System in Go With Exactly-Once Guarantees

101 Books

Our Creations

We are on Medium

How JavaScript Signals and Fine-Grained Reactivity Make Your Apps Faster

101 Books

Our Creations

We are on Medium

How to Use Python to Control Real-World Hardware: Sensors, Motors, and IoT Devices

101 Books

Our Creations

We are on Medium

How to Integrate Rust FFI With C Libraries Without Rewriting Your Entire Codebase

101 Books

Our Creations

We are on Medium

WebAssembly Is Turning Your Browser Into a Full Desktop Application Runtime