Forem: Rose Wabere

REST vs GraphQL vs WebSockets vs Webhooks: A Real-World Decision Guide (With Code)

Rose Wabere — Tue, 31 Mar 2026 20:34:03 +0000

You have used all of these. But when someone asks you, maybe in an interview, or in a system design meeting, why you chose WebSockets over polling, or webhooks over a queue, can you answer precisely?

This isn't another definitions post. This article is about knowing which tool to reach for and why, with code you can actually use.

Quick mental model before we start:

Communication patterns  →  REST | GraphQL | WebSockets | Webhooks
Code execution model    →  async/await

These live at different layers. Conflating them is the most common source of confusion.

async/await: The Foundation, Not the Feature

Let's kill one myth immediately: async/await is not a communication pattern. It's how your server handles waiting.

Every I/O operation — database queries, HTTP calls, file reads — makes your code wait. async/await ensures that waiting doesn't freeze every other user's request.

# BAD: blocks the event loop - all other requests stall for 40ms
@app.get("/order/{id}")
def get_order(id: int):
    order = db.execute("SELECT * FROM orders WHERE id = %s", id)  # blocking
    return order

# GOOD: yields control during the wait - other requests run while DB responds
@app.get("/order/{id}")
async def get_order(id: int):
    order = await db.fetch_one("SELECT * FROM orders WHERE id = $1", id)
    return order

When it matters most: High-concurrency services. A delivery platform handling 500 simultaneous drivers checking order status. An NGO dashboard pulling live survey data.

The trap: Calling a synchronous library inside an async function. This blocks the entire event loop.

import requests  # synchronous - don't use this inside async functions
import httpx     # async-capable - use this instead

# WRONG
async def fetch_rate():
    r = requests.get("https://api.exchangerate.host/latest")  # blocks event loop
    return r.json()

# CORRECT
async def fetch_rate():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.exchangerate.host/latest")
        return r.json()

REST: Default Choice for Good Reason

Use REST when:

The client initiates all interactions
Data doesn't change faster than the user refreshes
You're building standard CRUD

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Order(BaseModel):
    customer_id: int
    items: list[str]
    total: float

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    order = await db.fetch_one("SELECT * FROM orders WHERE id = $1", order_id)
    return order

@app.post("/orders")
async def create_order(order: Order):
    result = await db.execute(
        "INSERT INTO orders (customer_id, items, total) VALUES ($1, $2, $3) RETURNING id",
        order.customer_id, order.items, order.total
    )
    return {"order_id": result["id"]}

@app.delete("/orders/{order_id}")
async def cancel_order(order_id: int):
    await db.execute("UPDATE orders SET status = 'cancelled' WHERE id = $1", order_id)
    return {"status": "cancelled"}

Real-world scenario: A SACCO member portal. Members log in, check their loan balance, submit a loan application. All request/response. No data is changing while they're looking at a page. REST is perfect.

Production considerations:

Add proper HTTP caching headers (Cache-Control, ETag) - REST can be highly cacheable
Version your APIs (/v1/orders) from day one
Return proper status codes (201 for created, 404 for not found, 422 for validation errors - FastAPI handles this automatically)

GraphQL: REST With Client-Controlled Queries

Use GraphQL when:

Multiple clients (mobile, web, third-party integrations) need different shapes of the same data
You're constantly over-fetching or under-fetching with REST
You have deeply nested, relational data

# Using Strawberry (best GraphQL library for FastAPI)
import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

@strawberry.type
class Order:
    id: int
    total: float
    status: str

@strawberry.type
class User:
    id: int
    name: str
    email: str
    orders: list[Order]

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: int) -> User:
        return await get_user_with_orders(id)

schema = strawberry.Schema(query=Query)
graphql_router = GraphQLRouter(schema)

app = FastAPI()
app.include_router(graphql_router, prefix="/graphql")

Now a mobile app can request exactly what it needs:

# Mobile - bandwidth-conscious, needs minimal data
query {
  user(id: 1) {
    name
    orders(last: 3) {
      id
      status
    }
  }
}

# Web dashboard - needs full details
query {
  user(id: 1) {
    name
    email
    orders {
      id
      total
      status
      items { name price }
    }
  }
}

Real-world scenario: An NGO platform serving both field officers on 2G mobile and HQ analysts on desktops. Field officers need lightweight data. Analysts need full datasets. GraphQL lets one API serve both without maintaining separate endpoints.

Production considerations:

Implement depth limiting to prevent abusive nested queries
Add query complexity analysis: prevent user → orders → user → orders recursion
GraphQL doesn't cache well at HTTP layer: use DataLoader for N+1 query prevention
Don't default to GraphQL for simple services: it adds real overhead

WebSockets: When the Server Needs to Talk First

Use WebSockets when:

Data changes continuously, and the user needs to see updates immediately
Polling would generate unacceptable load or latency
Both client and server need to send messages freely

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import dict

app = FastAPI()

# Connection manager for multiple clients
class ConnectionManager:
    def __init__(self):
        self.active: dict[str, WebSocket] = {}

    async def connect(self, user_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active[user_id] = websocket

    def disconnect(self, user_id: str):
        self.active.pop(user_id, None)

    async def send_to_user(self, user_id: str, message: dict):
        ws = self.active.get(user_id)
        if ws:
            await ws.send_json(message)

    async def broadcast(self, message: dict):
        for ws in self.active.values():
            await ws.send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/track/{driver_id}")
async def track_driver(websocket: WebSocket, driver_id: str):
    await manager.connect(driver_id, websocket)
    try:
        while True:
            data = await websocket.receive_json()
            # Driver sent a location update - broadcast to assigned rider
            if data["type"] == "location_update":
                await manager.send_to_user(
                    data["rider_id"],
                    {"type": "driver_location", "lat": data["lat"], "lng": data["lng"]}
                )
    except WebSocketDisconnect:
        manager.disconnect(driver_id)

Real-world scenario: A ride-hailing app showing a driver moving on the map in real time. The driver's app sends GPS coordinates every 3 seconds over a persistent WebSocket. The rider's app receives them without polling. This would require: 1,000 REST requests per ride if you used polling at 3-second intervals.

Production considerations:

Persistent connections consume server resources — plan for horizontal scaling early
Use Redis Pub/Sub to share WebSocket state across multiple server instances
Always handle WebSocketDisconnect — clients drop off constantly (network, battery, background app)
Heartbeats keep connections alive through load balancers that close idle connections after 60 seconds

# Heartbeat to keep connection alive
@app.websocket("/ws/live")
async def live_feed(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            try:
                # Wait for message with 30s timeout
                data = await asyncio.wait_for(websocket.receive_json(), timeout=30.0)
                await handle_message(data)
            except asyncio.TimeoutError:
                # Send ping to keep connection alive
                await websocket.send_json({"type": "ping"})
    except WebSocketDisconnect:
        pass

Webhooks: Event Notification Between Services

Use webhooks when:

An external system needs to notify your system that something happened
You don't control the other system's push mechanism
You want event-driven integration without maintaining a persistent connection

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

WEBHOOK_SECRET = "your_flutterwave_webhook_secret"

def verify_flutterwave_signature(payload: bytes, signature: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.post("/webhooks/payment")
async def payment_webhook(request: Request):
    # 1. Verify the request is actually from Flutterwave
    signature = request.headers.get("verif-hash", "")
    body = await request.body()

    if not verify_flutterwave_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = await request.json()
    event_id = payload["data"]["id"]

    # 2. Idempotency - Flutterwave WILL retry on 500s
    if await db.webhook_event_exists(event_id):
        return {"status": "already_processed"}

    # 3. Acknowledge immediately, process async
    # Don't do heavy work here -- return 200 fast, process in background
    await background_tasks.add_task(process_payment_event, payload)
    await db.mark_webhook_received(event_id)

    return {"status": "received"}

async def process_payment_event(payload: dict):
    order_id = payload["data"]["meta"]["order_id"]
    await db.execute(
        "UPDATE orders SET status = 'paid', paid_at = NOW() WHERE id = $1",
        order_id
    )
    await send_confirmation_email(order_id)

Real-world scenario: Your Kenyan e-commerce platform integrates Mpesa via Daraja API. When a customer pays, Safaricom calls your /webhooks/mpesa endpoint with the transaction details. You mark the order paid and send a confirmation SMS. No polling. No persistent connection.

Production considerations:

Always return 200 fast; the webhook caller will retry if you're slow or error
Never trust webhook data without verifying the signature
Log every webhook payload for debugging; payment disputes will happen
Queue heavy processing (emails, SMS, inventory updates) with a background worker

Real-World Architecture: All Four Together

Here's how a production fintech app in Kenya uses all of these together:

┌──────────────────────────────────────────────────────────┐
│               Mobile/Web Client                          │
└───────┬───────────────────┬─────────────────┬────────────┘
        │                   │                 │
   REST (CRUD)         WebSocket         GraphQL
   POST /loans         /ws/notifications   /graphql
   GET /balance        (real-time alerts)  (analytics)
        │                   │                 │
┌───────▼───────────────────▼─────────────────▼────────────-┐
│                  FastAPI Backend                          │
│              (all async/await internally)                 │
└───────┬───────────────────────────────────────────────────┘
        │
   Webhook receiver
   POST /webhooks/mpesa   ← Safaricom calls this
   POST /webhooks/credit  ← Credit bureau calls this

Each pattern handles exactly what it's good at. The async/await inside FastAPI makes sure none of them block each other.

When to Use What: Decision Framework

Signal	Use
Client requests data on demand	REST
Multiple clients need different data shapes	GraphQL
User needs to see live updates without refreshing	WebSocket
External service needs to notify your backend of events	Webhook
Your code waits on database, HTTP, or file I/O	async/await

Hard rules:

If polling interval < 5 seconds and data changes frequently → switch to WebSocket
If you have > 3 different client types with different data needs → consider GraphQL
If you're integrating a payment provider, shipping tracker, or auth service → expect webhooks
If you're running FastAPI → use async def everywhere there's I/O, no exceptions

Common Mistakes (Those That Hurt in Production)

1. requests inside async def: blocks the entire event loop. Use httpx with await.

2. No idempotency on webhook handler: payment events get retried. Without idempotency checks, you'll charge customers twice.

3. WebSocket without reconnection logic: mobile networks drop. Your client-side WebSocket needs exponential backoff reconnection, or users see frozen data silently.

4. Assuming GraphQL is real-time: GraphQL subscriptions require a separate WebSocket-based setup. Standard queries/mutations are still request/response.

5. No signature verification on webhooks: your endpoint is public. Anyone can POST to it. Always verify HMAC signatures.

6. Keeping heavy processing in the webhook handler: the caller expects a fast response. Queue everything with a task worker (Celery, ARQ) and return 200 immediately.

If you found this helpful, please share it with others who it might help. And if you have questions, kindly drop them in the comments below!

Rose Wabere - Data & Analytics Engineer, Nairobi. Building real-world data systems with Python, FastAPI, and whatever tool the problem actually needs.

Building a Production-Grade Async Backend with FastAPI, SQLAlchemy, PostgreSQL, and Alembic

Rose Wabere — Sun, 22 Mar 2026 22:47:50 +0000

A complete technical breakdown of how FastAPI, SQLAlchemy, asyncpg, PostgreSQL, Declarative Base, and Alembic work together as a production system - covering sessions, transactions, rollbacks, relationships, migrations, and architecture.

Most tutorials show you how to get something running. The focus of this article is on how it actually works and why each decision matters when you are building something that must survive real traffic, real failures, and real schema changes over time.

What You Are Actually Building

A backend system is a coordinated pipeline. Every request travels through several layers before a response comes back.

Client → API Layer → Business Logic → Database Layer → Storage Engine → Response

Each tool in the stack owns one layer of this pipeline:

Tool	Responsibility
FastAPI	Handles HTTP requests and routing
SQLAlchemy	Translates Python objects to SQL
asyncpg	Async driver for PostgreSQL communication
PostgreSQL	Persistent storage
Declarative Base	Defines schema as Python classes
Alembic	Manages schema changes over time

The reason most backend bugs are hard to fix is not incorrect code. It is incorrect mental models of what each layer is doing.

Sync vs Async: Why the Model Matters

Your system can handle requests in two ways: blocking one at a time, or handling many concurrently.

Synchronous means each request holds a worker thread until the database responds:

def get_users(db):
    return db.query(User).all()

Asynchronous means the event loop handles other work while waiting for I/O:

async def get_users(db):
    result = await db.execute(select(User))
    return result.scalars().all()

Key insight: async does not make individual queries faster. It increases throughput by allowing hundreds of concurrent requests without spinning up hundreds of threads.

FastAPI runs on Python's asyncio event loop. Everything else in this stack is chosen to stay compatible with that.

Declarative Base: Defining Schema in Python

from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

This class is three things at once: a Python object, a table definition, and a mapping layer. Alembic reads this to generate migrations.

Async Engine and asyncpg

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"

engine = create_async_engine(DATABASE_URL)

Common mistake: using postgresql:// instead of postgresql+asyncpg://. It will not error immediately, but it will block your event loop under load.

Sessions: The Most Critical Concept

A session is not a database connection. It is a unit of work manager. It tracks modified objects, manages the transaction lifecycle, and handles cleanup.

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False
)

expire_on_commit=False keeps object data accessible after commit. Without it, accessing attributes after commit in an async context can raise errors because the session may already be closed.

Dependency Injection: Session Management at Scale

from typing import AsyncGenerator

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

FastAPI calls get_db(), creates a session, injects it into your route, and closes it after the response. Even on exceptions.

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

Without this pattern, connection leaks are almost inevitable as the codebase grows.

Transactions and Rollbacks

Every write operation in production must be treated as potentially failing.

try:
    db.add(user)
    db.add(audit_log)
    await db.commit()
except Exception:
    await db.rollback()
    raise

Without rollback: partial writes, inconsistent state, very hard to debug.
With rollback: database stays in its previous clean state.

Relationships: Foreign Keys and Loading Strategies

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="posts")

Loading relationships efficiently matters at scale. Use selectinload to avoid the N+1 problem:

result = await db.execute(
    select(User).options(selectinload(User.posts))
)

selectinload uses a second SELECT. joinedload uses a JOIN. Understanding the difference affects query performance directly.

Alembic: Schema Evolution

# Generate migration after changing a model
alembic revision --autogenerate -m "add email to users"

# Apply it
alembic upgrade head

# Reverse it
alembic downgrade -1

Critical setup: Alembic must import all your models in env.py or it will not detect their changes.

# alembic/env.py
from app.db.base import Base
from app.db.models import user, post
target_metadata = Base.metadata

Production Project Structure

app/
├── main.py
├── api/
│   └── v1/
│       └── routes/
├── core/
│   └── config.py
├── db/
│   ├── base.py
│   ├── session.py
│   └── models/
├── schemas/
├── services/
└── repositories/

Routes stay thin. Services hold business logic. Repositories handle data access. This separation keeps large codebases testable and maintainable.

Common Mistakes That Break Production Systems

# Wrong: sync query in async context
session.query(User).all()

# Wrong: missing await
db.execute(select(User))

# Wrong: manual session (no lifecycle guarantee)
db = SessionLocal()

# Wrong: sync URL with async engine
DATABASE_URL = "postgresql://..."

# Wrong: no rollback on failure
db.add(user)
await db.commit()

Every one of these mistakes has a production impact. None of them cause immediate errors in development.

Conclusion

A modern Python backend is a layered system where each component has a specific responsibility. The most important things to get right:

Use async throughout. One sync database call can block your event loop and degrade performance for all concurrent requests.
Use dependency injection for sessions. Manual session management leads to leaks.
Always handle rollbacks on writes. Partial state is worse than no state.
Use Alembic from day one. Manually altering production databases is how data gets corrupted.
Structure your project into layers. Routes, services, repositories, and schemas each have a job.

Understanding why this stack is designed the way it is separates engineers who can debug it from engineers who can only copy it.

I build real-world data systems and write about what I learn. Follow for more backend engineering and data content. GitHub: https://github.com/RoseWabere

Fix “invalid literal for int() with base 10: 'None'” in FastAPI (Asyncpg + Docker)

Rose Wabere — Wed, 18 Mar 2026 03:59:00 +0000

You're using Docker to deploy your FastAPI application with Aiven PostgreSQL, when all of a sudden you notice this traceback:

ValueError: invalid literal for int() with base 10: 'None'

It refers to SQLAlchemy's URL parser and complains that the string "-None" is the port portion of your database URL. Your URL appears as postgresql+asyncpg://user:pass@host:None/dbname. What caused that to occur? And above all, how do you fix it?

In this tutorial, I'll walk you through the cause of this error and provide a clean solution. We'll also discuss best practices for managing environment variables in containerized FastAPI applications.

Understanding the Error

The error occurs when SQLAlchemy's make_url() function tries to parse the database URL and convert the port component to an integer, but receives the literal string "None". This happens because somewhere in your code, you're building the URL with a variable that evaluates to None, and when interpolated into an f-string, None becomes the string "None".

The Code That Breaks

Consider this common pattern in database.py:

import os

DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOST = os.getenv("DB_HOST")
DB_PORT = os.getenv("DB_PORT")
DB_NAME = os.getenv("DB_NAME")

ASYNC_DB_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
ASYNC_DB_URL = os.getenv("DATABASE_URL")  # attempt to override with full URL

engine = create_async_engine(ASYNC_DB_URL)

When DB_PORT is not set in your environment (such as a Docker container), it becomes None. A URL with :None in the port position is generated by the f-string. The engine creation still uses the final value even if you later reassign ASYNC_DB_URL with the correct DATABASE_URL. Why, then, does it still have "None"? Because the assignment order is deceptive or because DATABASE_URL may also be missing.

In my instance, DATABASE_URL was set, but a string containing "-None" was already created in the first line and used. No, the engine actually makes use of the final value. Since the second line sets ASYNC_DB_URL to None but the engine attempts to use None, the issue must be that DATABASE_URL is not set. That would result in a different error. The port "None" error we observe indicates that the f-string URL is being used.

This only occurs in the event that DATABASE_URL is not set, in which case the code reverts to the constructed URL (possibly through a or clause). For example:

ASYNC_DB_URL = os.getenv("DATABASE_URL") or f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

If DATABASE_URL is empty, it uses the constructed URL, which has None for port if any component is missing.

Step-by-Step Debugging

Check your environment variables inside the container:

docker exec <container_name> env

Verify that DATABASE_URL is present and correct, and that individual component variables are not set (unless you intentionally set them).

Inspect the code in database.py. Look for any fallback logic that builds the URL from components. If you see something like the above, that's your culprit.
Simplify: Remove the component-based construction entirely. Use only the full DATABASE_URL:

import os

DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
    raise ValueError("DATABASE_URL environment variable is not set")

engine = create_async_engine(DATABASE_URL)

This ensures you're not accidentally creating a malformed URL.

Test by running your app again. The error should disappear.

Why the Single-Env-Var Approach is Better

Consistency: Most cloud database providers (Aiven, Heroku, etc.) give you a single DATABASE_URL. Using it directly avoids mismatches.
Simplicity: Fewer variables to manage in your .env and Docker configuration.
Portability: Your app can be deployed anywhere without changing how the URL is assembled.
Security: You avoid accidentally logging or exposing partial credentials.

Handling Async with asyncpg

If you're using asyncpg with SQLAlchemy, your DATABASE_URL must use the postgresql+asyncpg:// scheme. Aiven provides a URL like:

postgres://avnadmin:password@host:port/defaultdb?sslmode=require

You need to change the scheme to postgresql+asyncpg://. You can do this programmatically:

raw_url = os.getenv("DATABASE_URL")
if raw_url and raw_url.startswith("postgres://"):
    raw_url = raw_url.replace("postgres://", "postgresql+asyncpg://", 1)
engine = create_async_engine(raw_url)

Full Working Example

Here's a robust database.py for FastAPI with Aiven:

import os
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
    raise ValueError("DATABASE_URL environment variable is required")

# Ensure asyncpg scheme
if DATABASE_URL.startswith("postgres://"):
    DATABASE_URL = DATABASE_URL.replace("postgres://", "postgresql+asyncpg://", 1)

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

Environment Variables in Docker Compose

In your docker-compose.yml, pass the DATABASE_URL via environment or an env_file:

services:
  app:
    build: .
    env_file:
      - .env

And in your .env file:

DATABASE_URL=postgresql+asyncpg://avnadmin:password@host:port/defaultdb?sslmode=require

Conclusion

The ValueError: invalid literal for int() with base 10: 'None' is a symptom of a configuration mismatch. By adopting a single DATABASE_URL environment variable and eliminating component-based construction, you can avoid this and similar pitfalls. Remember to always validate required environment variables early and use defensive programming.

If you found this helpful, please share it with others who might be struggling with the same issue. And if you have questions, kindly drop them in the comments below!

Deploying FastAPI with Aiven PostgreSQL and Docker (Production-Ready Guide)

Rose Wabere — Tue, 17 Mar 2026 18:59:16 +0000

PostgreSQL is a reliable database, and FastAPI has emerged as one of the most popular frameworks for creating high-performance APIs. However, maintaining your own Postgres instance can be a distraction when it comes time to deploy. This is where the managed cloud database service Aiven comes into play. It takes away infrastructure issues and gives you complete access to PostgreSQL, making operations much less painful.

In this tutorial, I'll walk you through setting up a FastAPI project, connecting it to Aiven PostgreSQL with asyncpg, and containerizing everything with Docker for both development and production. Along the way, you'll discover how each component fits into a contemporary backend stack and the reasons these tools are frequently used together.

Requirements:

Python 3.9+
Docker & Docker Compose (for reproducible environments and easy deployment)
An Aiven account (free tier works and is enough for testing)
Basic familiarity with FastAPI and async/await (understanding non-blocking I/O will help)

Step 1: FastAPI Project Setup

Create a new project folder and set up a virtual environment:

mkdir myfastapi && cd myfastapi
python -m venv venv
source venv/bin/activate

Install dependencies:

# requirements.txt
fastapi
uvicorn[standard]
asyncpg
python-dotenv
gunicorn
email-validator
python-multipart

Here’s what these packages do:

FastAPI - the web framework
uvicorn - ASGI server for running FastAPI
asyncpg - high-performance async PostgreSQL driver
python-dotenv - loads environment variables from .env
gunicorn - production-grade process manager
others → useful utilities for forms and validation

Now, a minimal FastAPI app with a health check and a database test endpoint:

# app/main.py
from fastapi import FastAPI
import asyncpg
import os

app = FastAPI()

DATABASE_URL = os.getenv("DATABASE_URL")

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/db-check")
async def db_check():
    conn = await asyncpg.connect(DATABASE_URL)
    version = await conn.fetchval("SELECT version()")
    await conn.close()
    return {"postgres_version": version}

This example demonstrates:

A basic API endpoint (/)
A real database connection (/db-check)
Async database access using await, which allows FastAPI to handle many requests efficiently without blocking

Step 2: Configure Aiven PostgreSQL

Log in to Aiven Console, create a new PostgreSQL service (choose your cloud provider and region).
Once the service is running, go to the Overview tab and copy the Service URI. It will look like:

postgres://avnadmin:password@host:port/defaultdb?sslmode=require

Important: For asyncpg, the URI must use the postgresql+asyncpg:// scheme. Adjust it to:

postgresql+asyncpg://avnadmin:password@host:port/defaultdb?sslmode=require

This tells your app to use the asyncpg driver instead of the default PostgreSQL driver.

Save this as DATABASE_URL in a .env file (but remember, never commit it!).

Step 3: Dockerize the App

Create a Dockerfile:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Docker lets you package your app with all its dependencies into a container, ensuring:

consistent behavior across environments
easy deployment to cloud platforms
no “works on my machine” issues

Build the image: docker build -t myfastapi .

Step 4: Docker Compose for Development

docker-compose.yml (dev version):

version: "3.9"
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: ${DATABASE_URL}
    volumes:
      - ./app:/app/app   # live reload for code changes
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

Key ideas here:

Volumes allow live code updates without rebuilding the container
--reload enables hot-reloading for faster development
Environment variables keep secrets out of your code

Run with: docker compose --env-file .env up

Step 5: Running SQL Scripts on Aiven

Since we are not using a local database container, we execute scripts directly against Aiven using psql:

psql "$DATABASE_URL" < scripts/01_create_tables.sql

What this approach does:

treats Aiven as your single source of truth
avoids environment drift between local and production
is closer to how real production systems are managed

Make sure your DATABASE_URL in the shell uses the postgresql://scheme (without +asyncpg) for psql.

Step 6: Docker Compose for Production

Create docker-compose.prod.yml:

version: "3.9"
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: ${DATABASE_URL}
    command: gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000
    restart: always

Differences from development:

No volume mounts (immutable containers)
Uses Gunicorn for better process management
Multiple workers for handling concurrent traffic

Run in detached mode:

docker compose --env-file .env.production -f docker-compose.prod.yml up -d

Step 7: Environment Variables & Security

Never commit .env files. Use .env.example as a template.

In production, inject secrets via your CI/CD or orchestration tool.

Always use sslmode=require with Aiven to enforce encrypted connections.

Conclusion

You now have a FastAPI app running with a managed PostgreSQL database, fully containerized and ready for production. This setup lets you focus on writing code while Aiven handles the database heavy lifting.

If you found this helpful, give it a ❤️ and share it with your network.

Nairobi Property Listings Scraping: 500+ Property Listings with Smart Size Extraction

Rose Wabere — Wed, 18 Feb 2026 18:40:08 +0000

The Challenge

Building a house price prediction model for Nairobi requires a dataset that simply doesn’t exist in the open. You have to build it yourself by scraping property portals. But real estate data is very messy, listings are inconsistent, poorly structured, and is often contradictory.

Taking the size property for instance, one listing gives size as "4,350 sq. ft." Another says "Approx. 350 – 400 sqm." A third buries three villa sizes inside a paragraph.

So, how do you extract clean, usable data from this chaos?

In this post, I’ll walk you through the architecture of a scraper that collected 528 listings with 10 structured fields, focusing on the size extraction logic that handles ranges, commas, and mixed units.

Project Structure

nairobi-house-price-prediction/
├── data/
│   └── raw_listings.csv          # raw csv output
├── notebooks/
│   └── extraction.ipynb          # scraping logic
├── src/
│   ├── scraper.py                 # Core scraping functions
│   └── utils.py                    # Helpers (parsers, cleaners)
├── data_dictionary.json            # Schema definition
└── requirements.txt

Core Scraping Logic

The scraper iterates over pages, extracts needed data from each listing card, and then fetches more data from the detail page for richer information. I used requests with retries and BeautifulSoup for parsing.

The Size Extraction Problem

Property size appears in many forms on the site:

Example	Format
250 m²	Single, metric
4,350 sq. ft.	Imperial with comma
Approx. 350 – 400 sqm	Range, metric
465 to 476 square meters	Range with “to”
7,755 sq ft	Imperial, no dot
240 SQM	Uppercase

Some listings even contain multiple size mentions (built‑up area, terrace, garden). We need to extract the first meaningful built‑up size.

With all that chaos, the extract_size_from_text() function handles it all.

import re

def extract_size_from_text(text):
    """
    Extract built-up/property size from messy real estate descriptions.
    Returns original string of the most plausible size, or "N/A".
    """
    if not text:
        return "N/A"

    text = text.replace(",", "")
    candidates = []  # (size_in_sqm, original_text)

    # 1. Ranges in sqm
    range_matches = re.findall(
        r'(\d+(\.\d+)?)\s*(?:–|-|to)\s*(\d+(\.\d+)?)\s*(sqm|m²|square meters?)',
        text, re.IGNORECASE
    )
    for match in range_matches:
        low = float(match[0])
        high = float(match[2])
        if high >= 30:
            candidates.append((high, f"{match[0]}–{match[2]} sqm"))

    # 2. Single sqm
    sqm_matches = re.findall(
        r'(\d+(\.\d+)?)\s*(sqm|m²|square meters?)',
        text, re.IGNORECASE
    )
    for match in sqm_matches:
        val = float(match[0])
        if val >= 30:
            candidates.append((val, f"{match[0]} sqm"))

    # 3. Square feet (convert to sqm)
    sqft_matches = re.findall(
        r'(\d+(\.\d+)?)\s*(sq\.?\s*ft\.?|sqft)',
        text, re.IGNORECASE
    )
    for match in sqft_matches:
        sqft = float(match[0])
        if sqft >= 300:
            sqm = sqft * 0.092903
            candidates.append((sqm, f"{match[0]} sq ft"))

    # 4. Acres (including fractions)
    acre_matches = re.findall(
        r'(\d+/\d+|\d+(\.\d+)?)\s*-?\s*(acre)',
        text, re.IGNORECASE
    )
    for match in acre_matches:
        raw = match[0]
        if '/' in raw:
            num, den = raw.split('/')
            acres = float(num) / float(den)
        else:
            acres = float(raw)
        if acres >= 0.05:
            sqm = acres * 4046.86
            candidates.append((sqm, f"{raw} acre"))

    if candidates:
        plausible = [c for c in candidates if c[0] >= 30]
        if plausible:
            # Return the largest (most likely built-up area)
            return max(plausible, key=lambda x: x[0])[1]

    return "N/A"

Why This Approach?

Unit normalization allows fair comparison across different measurement systems.

Thresholds (30 sqm, 300 sqft, 0.05 acre) eliminate noise from tiny areas that are clearly not the main house or are clearly not logical.

Max selection handles listings that describe multiple areas (e.g., terraces, gardens). This means that we take the largest, which typically corresponds to the built‑up area.

Regex patterns are flexible enough to catch variations like "sqm", "m²", "square meters", "sq. ft.", "sqft", and acre fractions.

Integration with the Scraper

The scraper collects size from two places:

Listing card – quick size from swiper slides or truncated description.

Detail page – full description that contains a more accurate size. If found, it overrides the card size.

def extract_bedrooms_bathrooms_size(listing):
    bedrooms = bathrooms = "N/A"
    size_from_swiper = "N/A"

    # ... extract from swiper slides ...

    # Extract from description on the card
    desc_div = listing.find('div', id='truncatedDescription')
    if desc_div:
        desc_text = desc_div.get_text(" ", strip=True)
        size_from_desc = extract_size_from_text(desc_text)

    size = size_from_desc if size_from_desc != "N/A" else size_from_swiper
    return bedrooms, bathrooms, size

The detail page

def extract_bedrooms_bathrooms_size(listing):
    bedrooms = bathrooms = "N/A"
    size_from_swiper = "N/A"

    # ... extract from swiper slides ...

    # Extract from description on the card
    desc_div = listing.find('div', id='truncatedDescription')
    if desc_div:
        desc_text = desc_div.get_text(" ", strip=True)
        size_from_desc = extract_size_from_text(desc_text)

    size = size_from_desc if size_from_desc != "N/A" else size_from_swiper
    return bedrooms, bathrooms, size

Then in scrape_listing:

if size_from_detail != "N/A":
    size = size_from_detail  # override card size

Error Handling & Resilience

Retry logic with fetch_page – requests up to 3 times with a 2‑second delay between attempts.

Fallback data– if detail page fails, return basic info (size from card).

Checkpointing – main loop stops when max_listings stops at max_listings (800) is reached to avoid over‑scraping.

Polite delays – 1s between detail requests, 2s between pages to prevent IP blocking.

Data Dictionary: Schema as Code

I defined the schema upfront to ensure consistency:

data_dictionary = [
    {"Column": "Title", "Type": "String", "Description": "Property name"},
    {"Column": "Property Type", "Type": "String", "Description": "Apartment, Townhouse, etc."},
    # ... etc.
]

This is saved as data_dictionary.json, serving as documentation for collaborators and future‑me.

Results after running pages 1–40:

START_PAGE = 1
END_PAGE = 40      
MAX_LISTINGS = 800     # max

# Scrape
df = scrape_pages(START_PAGE, END_PAGE, MAX_LISTINGS)

print(len(df))  # 800

print(df.columns)
# ['Title', 'Property Type', 'Price', 'Location', 'Bedrooms', 'Bathrooms', 'Size', 'Amenities', 'Surroundings', 'Created At']

800 listings from pages 1–40
10 fields per listing
Size captured in listings if available
0 critical failures – thanks to retries and fallbacks
Raw CSV saved to data/raw_listings.csv

Lessons for Production Scrapers

Always normalise units – you can't compare apples and oranges.

Filter implausible values – they corrupt your model.

Have a fallback – if detail page fails, keep the card data.

Document your schema – you'll thank yourself later.

Respect the source – rate limiting isn't optional.

Next Steps: Data Cleaning

Load raw_listings.csv
Remove duplicates
Standardize location strings
Convert price to integer (remove “KSh”, commas)
Convert size to numeric (extract first number from ranges, convert acres to sqm)
Create features: price_per_sqft, amenity_score (count of amenities), month from Created At
Basic EDA and save as clean_listings.csv

Get the Code

All code is available on GitHub: github

Contributions welcome! Feel free to open issues or PRs.

Connect with Me

LinkedIn: linkedin

Python #WebScraping #BeautifulSoup #DataScience #MachineLearning #Kenya #OpenSource #Regex

Fixing Kali Linux's "Failed to Set Beacon Parameters" Hotspot Error: A Troubleshooting Tale

Rose Wabere — Sat, 07 Feb 2026 20:22:38 +0000

Have you ever needed to turn your Kali Linux machine into a Wi-Fi hotspot, only to be stopped cold by a cryptic error? You install hostapd, write your config, and run the command, brimming with confidence. Then, the terminal spits back:

Failed to set beacon parameters
Interface initialization failed

Your hotspot is dead on arrival. This isn't just a minor bug—it's a fundamental hardware limitation that most tutorials gloss over. I hit this wall myself while setting up a penetration testing lab. Through trial, error, and deep diving, I found the solution isn't just another apt install command. It's about understanding the single-channel limitation of your wireless card and using the right tool to bypass it.

This is the story of that troubleshooting journey and the definitive guide to creating a reliable Wi-Fi-to-Wi-Fi hotspot on Kali Linux.

The Scene of the Crime: A Standard Setup That Should Have Worked

My goal was simple: share my laptop's existing Wi-Fi internet connection with another device through a new hotspot, all using Kali. The standard approach for this is using hostapd (the host access point daemon). I followed the typical steps:

Installed the required packages: hostapd and dnsmasq.
Created a clean /etc/hostapd/hostapd.conf file, setting my SSID, password, and channel.
Configured dnsmasq for DHCP and set up IP forwarding with iptables.

I ran sudo hostapd /etc/hostapd.conf, expecting to see AP-ENABLED. Instead, I was greeted with the "Failed to set beacon parameters" error. The interface would flip between enabled and disabled states before failing completely. A search online showed I wasn't alone—this is a common, frustrating roadblock.

The Investigation: Why Your Wi-Fi Card Refuses to Cooperate

The core issue is physics, not software. Most internal Wi-Fi cards are designed to operate on one radio frequency at a time.

The Problem: If your Kali laptop is connected to a router on Channel 6, the card's radio is tuned to that frequency. You cannot force the same physical radio to simultaneously broadcast a hotspot on Channel 11. It's like asking a musician to play two different notes on the same instrument at the exact same moment—it's impossible for many consumer-grade cards.
The Error: This hardware conflict manifests as the Failed to set beacon parameters error. The driver receives conflicting commands and fails.
The Realization: The standard hostapd method works perfectly for creating a hotspot from a wired Ethernet connection, because the Wi-Fi card has only one job: broadcasting. The "Wi-Fi to Wi-Fi" scenario is where it falls apart.

The Solution: Introducing `linux-wifi-hotspot`

The breakthrough came when I discovered the linux-wifi-hotspot tool. Its key advantage is right in the description on its GitHub page: it is "able to create a hotspot using the same wifi card which is connected to an AP already". It does this by creating a virtual interface and handling the complex routing and channel coordination automatically.

Here’s a comparison of the two primary methods:

Feature	The Manual `hostapd` Method	The `linux-wifi-hotspot` Method
Core Purpose	Creates a standard access point (AP).	Creates an AP specifically while connected to another network.
"Wi-Fi to Wi-Fi"	Usually fails due to single-channel limitation.	Designed to solve this problem.
Ease of Use	Requires manual config of `hostapd`, `dnsmasq`, `iptables`.	Simple GUI or one-line terminal command.
Best For	Sharing a wired Ethernet connection or advanced custom setups.	Sharing your current Wi-Fi connection quickly and reliably.

Step-by-Step: Building and Using Your Reliable Hotspot

The tool isn't always in the default Kali repositories, so we build it from source. Here’s the process that finally worked:

1. Install the Dependencies:
First, we need the libraries for building the software and the engines that power the hotspot.

sudo apt update
sudo apt install -y libgtk-3-dev build-essential gcc g++ pkg-config make hostapd dnsmasq libqrencode-dev libpng-dev libglib2.0-dev

2. Build and Install linux-wifi-hotspot from Source:
Building from source ensures we get the latest version with all features intact.

git clone https://github.com/lakinduakash/linux-wifi-hotspot
cd linux-wifi-hotspot
make
sudo make install

3. Launch the Hotspot (The Correct Way):
You can use the GUI (wihotspot), but the command line gives us the precise control needed to solve the channel conflict.

First, identify your current Wi-Fi channel:

sudo iwlist wlan0 channel | grep Current

Note the channel number (e.g., Current Frequency:2.437 GHz (Channel 6)).

Now, launch the hotspot, forcing it to the SAME channel:

sudo create_ap -c 6 wlan0 wlan0 MySecureHotspot MyStrongPassword123

The -c 6 flag is the magic key. It forces the hotspot to broadcast on Channel 6, eliminating the channel conflict. Replace 6 with your identified channel, and wlan0 with your interface name if different.

Expected Success Output:

Config dir: /tmp/create_ap.wlan0.conf.xyz
PID: 12345
...
AP-Enabled

Keep that terminal window open—closing it will turn off the hotspot. To run it in the background, you can add the -m flag to the command.

The Takeaway: More Than Just a Fixed Hotspot

Solving this problem taught me a crucial lesson in troubleshooting: understand the tool's purpose. I was trying to use a hammer (hostapd) to drive a screw (Wi-Fi-to-Wi-Fi sharing). The real fix was finding the right screwdriver (linux-wifi-hotspot).

This process highlights essential skills for any security professional or Linux user:

Diagnosing Hardware Limitations: Learning to use iw list to check for AP mode support and understanding channel conflicts is foundational.
Systematic Troubleshooting: Moving from a generic error message to identifying the single-channel limitation mirrors real-world diagnostic workflows.
Leveraging the Right Tool: The open-source community often has specialized tools for niche problems. Finding and using them is a key competency.

By matching channels and using create_ap, you're not just running a command; you're applying a deeper understanding of your system to bypass a hardware constraint. This transforms your Kali machine from a device stumped by a basic error into a powerful, flexible networking hub.

Have you encountered other stubborn Kali Linux networking errors? Share your story in the comments—let's build a repository of solutions for the community.

Tags: #KaliLinux #CyberSecurity #LinuxNetworking #WiFiHotspot #FailedToSetBeaconParameters #Troubleshooting #LinuxTips #InfoSec

Building Production-Grade Data Analytics Pipelines: A Real-World Case Study in Government Data

Rose Wabere — Wed, 21 Jan 2026 11:43:44 +0000

How I transformed suppressed government data into actionable intelligence for policy makers—and what it taught me about data engineering, stakeholder communication, and building systems that survive real-world scrutiny.

The Problem

A government agency tasked me with analyzing a multi-year, multi-dimensional dataset covering hierarchical performance metrics across thousands of locations and demographic subgroups. Think: sprawling Excel files with cryptic codes, privacy-protected values, and conflicting data quality requirements.

The catch? About 30-40% of the data was suppressed using privacy-protection codes:

* = Small populations (suppressed for confidentiality)
< or > = Boundary conditions (threshold effects)
Numeric values = Reportable data
Blank = Missing or unreported

They needed actionable insights from incomplete data. Fast. But I knew that garbage in = garbage out. I couldn't just drop this into a model and hope for the best.

This is the story of how I built a production-grade analytics pipeline that turns suppressed, incomplete data into trustworthy intelligence—and what it taught me about working with real-world datasets.

The Challenge: Data as a Governance Problem

Most tutorials show you clean, happy datasets. Real life? Real life shows up with:

# Real government data looks like this:
df.head()

# Output:
   Location Name         Subgroup         Metric Rate
0  Location A            All              65.5
1  Location A            Subgroup X       *
2  Location A            Subgroup Y       >95
3  Location A            Subgroup Z       <3
4  Location B            All              54.2

The core challenge wasn't technical—it was conceptual.

How do you analyze data when some values are intentionally hidden for privacy? How do you communicate with confidence when 30-40% of your dataset is suppressed? How do you avoid making bad decisions based on incomplete information?

Traditional approaches would either:

Ignore suppressed data (lose 30-40% of context)
Estimate suppressed values (introduce bias and legal risk)
Include them as zeros or arbitrary values (create false conclusions)
Give up (deliver nothing)
Classify and track systematically (the right way)

I needed approach #5.

The Solution: A Multi-Layer Data Architecture

Layer 1: Suppression Classification

First, I built a classifier that understood what each code meant:

def identify_suppression_type(value):
    """
    Classify suppression types in educational data.
    Returns classification for auditing and analysis purposes.
    """
    if pd.isna(value):
        return 'missing'

    str_val = str(value).strip()

    if str_val == '*':
        return 'complete_suppression'      # Privacy protection
    elif str_val.startswith('>'):
        return 'threshold_high'             # >98% (ceiling)
    elif str_val.startswith('<'):
        return 'threshold_low'              # <2% (floor)
    elif any(c.isdigit() for c in str_val.replace('.', '').replace('%', '')):
        return 'valid_numeric'              # Actual data
    else:
        return 'unknown_text'               # Data quality issue

Why this matters: This function became the "source of truth" for understanding data quality. Every record now had a classification, enabling downstream decision-making.

Layer 2: Dual-Column Architecture

Here's where I made a deliberate architectural choice that surprised stakeholders but saved countless hours of debugging:

def clean_government_data(df, target_column):
    """
    Preserve original values while creating analytics-ready columns.

    Architecture:
    - Original column: Unchanged (audit trail for compliance)
    - Numeric column: Converted values (analytics calculations)
    - Flag column: Suppression type (governance tracking)
    """
    df_clean = df.copy()

    # Create metadata column (governance and compliance)
    df_clean[f'{target_column}_flag'] = df_clean[target_column].apply(
        identify_suppression_type
    )

    # Create analytics column (separate from original)
    def convert_to_numeric(value):
        if pd.isna(value):
            return np.nan

        str_val = str(value).strip()

        # Suppress non-numeric indicators
        if str_val in ['*', ''] or str_val.startswith('>') or str_val.startswith('<'):
            return np.nan

        # Convert valid values
        try:
            return float(str_val.replace('%', ''))
        except ValueError:
            return np.nan

    df_clean[f'{target_column}_numeric'] = df_clean[target_column].apply(
        convert_to_numeric
    )

    return df_clean

Result: Three versions of every datapoint:

Original:        65.5  (or *, <, >)
Numeric:         65.5  (or NaN)
Flag:            'valid_numeric' (or 'suppression_type')

Why this architecture?

Audit trail (legal requirement for government work)
Recalculation possible (if methodology changes)
No information loss (original preserved)
Type safety (NaN, not corrupted values)
Compliance-ready (can demonstrate data handling integrity)

Layer 3: Intelligent Missing Data Handling

Most analyses would dropna() and move on. I went deeper:

def verify_data_coverage(df_cleaned, column_name):
    """
    Understand suppression patterns to assess analysis reliability.
    This determines confidence levels for any downstream insights.
    """
    numeric_col = f'{column_name}_numeric'
    flag_col = f'{column_name}_flag'

    total_rows = len(df_cleaned)
    valid_data = df_cleaned[numeric_col].notna().sum()
    suppressed = (df_cleaned[flag_col] == 'complete_suppression').sum()
    threshold_issues = (
        (df_cleaned[flag_col] == 'threshold_high') | 
        (df_cleaned[flag_col] == 'threshold_low')
    ).sum()

    coverage_pct = 100 * valid_data / total_rows

    print(f"Data Coverage for {column_name}:")
    print(f"  Valid data: {valid_data}/{total_rows} ({coverage_pct:.1f}%)")
    print(f"  Privacy suppression: {suppressed} ({100*suppressed/total_rows:.1f}%)")
    print(f"  Threshold issues: {threshold_issues} ({100*threshold_issues/total_rows:.1f}%)")

    return {
        'total': total_rows,
        'valid': valid_data,
        'suppressed': suppressed,
        'coverage_pct': coverage_pct
    }

Output:

Data Coverage for Performance Metric:
  Valid data: 15,234/20,700 (73.6%)
  Privacy suppression: 3,102 (15.0%)
  Threshold issues: 1,854 (8.9%)

This transparency gave stakeholders the confidence they needed: "We can analyze 73.6% of data with high confidence. The missing 26.4% is either privacy-protected or boundary conditions. Here's what that means for our conclusions..."

Key insight: By quantifying data quality upfront, I established credibility and set realistic expectations about what the analysis could deliver.

The Analysis: From Data to Decisions

With clean data, the analysis became straightforward. But the presentation? That's where the real work happened.

Discovery #1: Disparities Emerge When You Measure Correctly

subgroup_stats = location_df.groupby('Demographic_Subgroup')['Metric_numeric'].agg([
    'count', 'mean', 'median', 'std', 'min', 'max'
])

# When analyzed across valid data only:
# Subgroup_A: 52.1% (below system average)
# Subgroup_B: 77.1% (above system average)
# System Average: 65.5%
# Disparity: ~25 percentage points

Key insight for stakeholders: Once you properly handle suppressed data, disparities become visible and quantifiable. This is the foundation for evidence-based decision-making.

Discovery #2: Geography Is a Powerful Predictor

# Location-level analysis reveals dramatic variation
location_rates = location_df.groupby('Location_ID')['Metric_numeric'].mean()

print(f"Highest performing: {location_rates.max():.1f}%")
print(f"Lowest performing: {location_rates.min():.1f}%")
print(f"Range: {location_rates.max() - location_rates.min():.1f}%")

Individual locations vary dramatically. This variation tells you something important: context matters. Different resources, different strategies, different outcomes.

Discovery #3: Hierarchical Analysis Reveals System-Level Patterns

# Analyze at multiple levels:
location_level = location_df.groupby('Location').agg({'Metric_numeric': ['mean', 'count']})
district_level = location_df.groupby('District').agg({'Metric_numeric': ['mean', 'count']})
regional_level = location_df.groupby('Region').agg({'Metric_numeric': ['mean', 'count']})
system_level = location_df['Metric_numeric'].mean()

Insight: What looks like variation at the location level often has district-level explanations. System-level patterns suggest policy impacts.

The Visualization: Making Data Speak

Raw numbers don't move decision-makers. Color-coded, clearly labeled visualizations do.

import matplotlib.pyplot as plt
import seaborn as sns

def get_risk_color(metric_value):
    """Color-code by performance tier for immediate visual understanding."""
    if metric_value > 70:
        return '#388e3c'      # Green: Strong performance
    elif metric_value > 65:
        return '#fbc02d'      # Yellow: Acceptable
    elif metric_value > 60:
        return '#f57c00'      # Orange: Needs attention
    else:
        return '#d32f2f'      # Red: Critical

# Create multi-panel risk-stratified visualization
fig, axes = plt.subplots(2, 2, figsize=(16, 12))

# Panel 1: All subgroups ranked by performance
subgroup_stats = location_df.groupby('Demographic_Subgroup')['Metric_numeric'].mean().sort_values()
colors = [get_risk_color(val) for val in subgroup_stats.values]

axes[0, 0].barh(range(len(subgroup_stats)), subgroup_stats.values, color=colors)
axes[0, 0].set_yticks(range(len(subgroup_stats)))
axes[0, 0].set_yticklabels(subgroup_stats.index)
axes[0, 0].axvline(location_df['Metric_numeric'].mean(), 
                   color='purple', linestyle='--', label='System Average')
axes[0, 0].set_title('Performance by Demographic Group (Ranked)', 
                     fontweight='bold')
axes[0, 0].legend()

# Panel 2: Critical performance focus
critical = subgroup_stats[subgroup_stats < 60]
if len(critical) > 0:
    axes[0, 1].barh(range(len(critical)), critical.values, 
                    color=['#d32f2f']*len(critical))
    axes[0, 1].set_title(f'Groups Requiring Targeted Support ({len(critical)} groups)', 
                         fontweight='bold')

# Panel 3: Disparity visualization
highest = subgroup_stats.max()
lowest = subgroup_stats.min()
disparity = highest - lowest
axes[1, 0].bar(['Lowest Performing', 'System Average', 'Highest Performing'], 
              [lowest, location_df['Metric_numeric'].mean(), highest],
              color=['#d32f2f', '#9c27b0', '#388e3c'])
axes[1, 0].set_title(f'Performance Disparity Gap: {disparity:.1f}%', 
                     fontweight='bold')

# Panel 4: Performance distribution
performance_dist = pd.Series({
    'Strong (>70%)': (subgroup_stats > 70).sum(),
    'Acceptable (65-70%)': ((subgroup_stats >= 65) & (subgroup_stats <= 70)).sum(),
    'Needs Attention (60-65%)': ((subgroup_stats >= 60) & (subgroup_stats < 65)).sum(),
    'Critical (<60%)': (subgroup_stats < 60).sum()
})
axes[1, 1].bar(range(len(performance_dist)), performance_dist.values, 
              color=['#388e3c', '#fbc02d', '#f57c00', '#d32f2f'])
axes[1, 1].set_xticks(range(len(performance_dist)))
axes[1, 1].set_xticklabels(performance_dist.index, rotation=45, ha='right')
axes[1, 1].set_title('Performance Distribution Across All Groups', fontweight='bold')

plt.tight_layout()
plt.show()

Impact: These visualizations communicate what statistics can't—the scale of disparities and the distribution of challenges. Decision-makers understand immediately which groups need attention.

Lessons Learned: What Real Government Clients Actually Care About

Lesson #1: Data Quality > Data Quantity

The agency had 20,000+ records. But 30-40% was suppressed. That's not 20,000 usable datapoints—it's ~12,000-14,000.

Most analysts would hide this fact. I led with it:

"Here's what we can analyze with high confidence (73.6% of records), here's what's privacy-protected (15%), and here's what's boundary conditions (8.9%). Our recommendations are based on properly validated data across thousands of locations."

Result: They trusted me more, not less. Transparency about limitations builds credibility.

Lesson #2: Preserve the Audit Trail

When I showed them three columns (original → numeric → flag), they initially thought I was overcomplicating things.

Months later, when budget and policy questions arose: "How did you calculate that disparity? Can you show the original data?"

I could pull the exact original value, show the classification logic, show the calculation, and audit the entire pipeline.

That's when they realized the real value.

Code that can be audited survives. Government agencies demand auditability. It's not optional.

Lesson #3: Stakeholders Care About Impact, Not Methods

I spent weeks optimizing the data pipeline and architecture. The stakeholders spent 30 seconds looking at the visualizations.

They didn't care that I used pandas groupby operations or matplotlib color schemes. They cared that:

The data quality was clear
The disparities were quantifiable
They had actionable next steps
The methodology could be audited

Lesson: Show 30 seconds of impact, not 3 hours of methodology. (Save the methodology for technical documentation and code comments.)

Lesson #4: One Number Changes Everything

The single most powerful metric? The disparity magnitude.

Not "Group A is at 52%" and "Group B is at 77%"...

"We have a 25-point performance disparity across demographic groups."

That number led directly to:

Identifying priority populations
Budget allocation decisions
Program creation
Policy discussions

Find the one number that captures the essence of the problem.

Lesson #5: Government Work Requires Different Thinking

This wasn't a startup where you ship fast and iterate. This was government where:

Everything must be documented
All decisions need justification
All calculations must be verifiable
Privacy and compliance are non-negotiable
Stakeholders need to understand why, not just what

This requires a different approach than commercial data science. You're not optimizing for speed—you're optimizing for integrity, transparency, and defensibility.

Technical Skills Showcased

Here's what this project demonstrated to potential clients and recruiters:

Data Engineering

Multi-source data consolidation (spreadsheets → Pandas)
Schema design for compliance (original + numeric + flag columns)
Data quality assessment (suppression classification)
Hierarchical data processing (locations → districts → regions → system)

Data Analysis

Descriptive statistics with missing data handling
Comparative analysis across hierarchical groupings
Disparity quantification and gap analysis
Statistical confidence assessment based on coverage %

Data Visualization

Matplotlib + Seaborn for publication-quality charts
Performance-based color coding (green/yellow/orange/red)
Multi-panel dashboards for comprehensive understanding
Clear labeling and legend design for non-technical audiences

Problem-Solving

Handling suppressed/sensitive data (privacy protection)
Architectural decisions (dual-column approach for auditability)
Stakeholder communication (translating data → insights)
Government compliance thinking (documentation, verification)

Statistical Thinking

Sample size implications of data suppression
Understanding confidence in partial datasets
Appropriate analysis methods given data constraints
Avoiding false precision and overstatement

Python Proficiency

Pandas for data manipulation and groupby operations
NumPy for numeric handling
Matplotlib for visualization
Jupyter notebooks for reproducible analysis
Functions and classes for code organization

The Business Impact

What started as a data cleaning exercise became a comprehensive analytics solution:

4 hierarchical levels processed (Location, District, Region, System)
20,000+ records analyzed (with clear confidence levels)
Disparities quantified (specific gaps identified across demographic groups)
Resource allocation decisions informed (data-driven budget prioritization)
5 evidence-based recommendations delivered to leadership
Reproducible pipeline established (for ongoing monitoring)
Audit trail complete (every calculation traceable to source data)

What I'd Do Differently

If I could go back:

Involve stakeholders earlier - I built the analysis framework in isolation. More collaborative design upfront would've ensured alignment with actual decision-making needs.
Create interactive dashboards - Static visualizations are good; interactive tools are better. "What if we focus on X region?" dashboard would've been powerful.
Qualitative context - "Why" matters as much as "what." Interviews with field staff would've contextualized the numbers.
Longer-term monitoring - Set up quarterly reporting to track progress on recommendations and refine strategies.

The Key Takeaway

Real-world data work isn't about building the most sophisticated model. It's about:

Understanding the domain (government operations, compliance, policy constraints)
Handling messy reality (suppressed data, missing values, privacy protection)
Building trust (transparent methodology, auditable results, clear communication)
Communicating impact (visualizations, clear narratives, actionable next steps)
Creating reproducibility (clean code, documentation, maintainability)

If you can do these five things, you can work with any dataset for any client—whether it's government, nonprofit, or private sector.

For Recruiters & Potential Clients

If you're looking for someone who:

Transforms messy, suppressed data into actionable intelligence
Communicates insights to non-technical stakeholders and decision-makers
Handles data governance, privacy, and compliance requirements
Builds production-grade analytics systems that survive scrutiny
Thinks about impact and stakeholder value, not just technical elegance

...I'm interested in talking.

The tools matter less than the thinking. Python, R, SQL, Excel—they're all just ways to implement sound analytical thinking. That's what I bring to every project.

Specific strengths:

Government and public sector data projects
Handling sensitive/suppressed data responsibly
Complex hierarchical analysis
Stakeholder communication and visualization
Building trust through transparency

Code & Documentation

The technical approach used in this project is documented in detail at: data_handling.md

Key files demonstrate:

Suppression classification logic
Dual-column architecture pattern
Data quality assessment functions
Visualization frameworks
Best practices for government data projects

What data challenges are you facing? What's preventing you from getting clear answers from your data? What would change if you could confidently analyze your complete dataset?

Let me know in the comments—I'd love to help you think through it.

DataScience #DataAnalytics #Python #Government #DataQuality #RealWorldDataAnalysis #DataVisualization #DataEngineering #StakeholderCommunication

Building a Modern Data Platform to Track Kenya’s Food Prices — A Data Engineering Case Study

Rose Wabere — Thu, 13 Nov 2025 14:36:25 +0000

Food price volatility has always been a sensitive issue across Kenya. From urban households in Nairobi to smallholder farmers in Turkana, the ripple effects of fluctuating food prices are felt everywhere. In this article, I share my end-to-end journey designing and implementing a data platform that tracks food prices across Kenyan counties, detects inflation patterns, and visualizes trends for policymakers, consumers, and researchers.

This project is both a portfolio centerpiece and a real-world solution — demonstrating how modern data engineering practices can power impactful analytics for the public good.

Project Goals & Context

The main objective:

To design a scalable, automated pipeline that cleans, aggregates, and visualizes food price data from multiple Kenyan data sources — delivering real-time, actionable insights.

Key outcomes:

Build a unified data model to harmonize market, commodity, and regional data.
Automate the ETL process using PySpark.
Store processed data in PostgreSQL for analytics.
Build dashboards in Grafana for spatial and temporal insights.

High-Level Architecture

[Raw Data Sources]
    |-- WFP HDX Food Prices
    |-- KNBS & FAOSTAT Datasets
    |-- Government Market Surveys
        ↓
[Ingestion Layer]
    pandas 
        ↓
[Processing Layer]
    PySpark Data Cleaning + Normalization
        ↓
[Storage Layer]
    PostgreSQL (Star Schema)
        ↓
[Visualization Layer]
    Grafana Dashboards / Streamlit Web App

This modular setup ensures scalability — you can plug in additional data sources or visualization layers (like Power BI or Superset) without refactoring the pipeline.

Data Ingestion

I used publicly available data from the World Food Programme (WFP) via the Humanitarian Data Exchange (HDX).

Data was fetched dynamically from the remote CSV URL and processed in PySpark for scalability.

Code Example: Reading the Data in PySpark

import requests
from pyspark.sql import SparkSession

# Initialize Spark Session
spark = SparkSession.builder \
    .appName("KenyaFoodPrices") \
    .config("spark.jars.packages", "org.postgresql:postgresql:42.7.3") \
    .getOrCreate()

# Remote CSV from HDX
csv_url = "https://data.humdata.org/dataset/e0d3fba6-f9a2-45d7-b949-140c455197ff/resource/517ee1bf-2437-4f8c-aa1b-cb9925b9d437/download/wfp_food_prices_ken.csv"

def fetch_and_load_csv():
    response = requests.get(csv_url)
    data = response.content.decode('utf-8')

    # Split into header and data
    lines = data.splitlines()
    rdd = spark.sparkContext.parallelize(lines)
    df = spark.read.csv(rdd, header=True, inferSchema=True)
    return df

# Load into Spark DataFrame
df = fetch_and_load_csv()
df.show(5)

This approach avoids downloading files manually and scales well in scheduled jobs (for example, through Airflow or Databricks).

Data Transformation with PySpark

PySpark handled the heavy ETL logic — particularly useful for large datasets spanning multiple years and counties.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, lower

spark = SparkSession.builder \
    .appName("KenyaFoodPriceETL") \
    .config("spark.jars", "postgresql-42.6.0.jar") \
    .getOrCreate()

# Load data
raw_df = spark.read.csv("combined_raw.csv", header=True, inferSchema=True)

# Clean commodity and location names
clean_df = raw_df.withColumn("commodity", trim(lower(col("commodity")))) \
                 .withColumn("region", trim(lower(col("region")))) \
                 .filter(col("price").isNotNull())

# Normalize price units (example conversion)
from pyspark.sql.functions import when

clean_df = clean_df.withColumn(
    "price_per_kg",
    when(col("unit") == '90kg', col("price") / 90)
    .when(col("unit") == '1kg', col("price"))
    .otherwise(col("price"))
)

This logic ensures consistency — converting all prices to a comparable unit (per kilogram), filtering out nulls, and standardizing categorical columns.

Data Modeling: Star Schema Design

To make the data warehouse efficient for analytics, I implemented a star schema model in PostgreSQL.

Fact Table: fact_prices

Column	Description
date_id	Foreign key to `date_dim`
location_id	Foreign key to `location_dim`
product_id	Foreign key to `product_dim`
price_type_id	Retail/Wholesale category
price	Price per KG
quantity	Quantity sold

Dimension Tables:

product_dim: commodity, category
location_dim: region, district, market, latitude, longitude
date_dim: day, month, year
price_type_dim: retail/wholesale indicator

Table Creation via SQLAlchemy

Before loading the data from Spark, I pre-created the tables using SQLAlchemy — ensuring primary keys and constraints were properly set.

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, Float, Date, ForeignKey
from dotenv import load_dotenv
import os

load_dotenv()

engine = create_engine(os.getenv("POSTGRES_URL"))
metadata = MetaData()

product_dim = Table('product_dim', metadata,
    Column('product_id', Integer, primary_key=True),
    Column('commodity', String),
    Column('category', String)
)

fact_prices = Table('fact_prices', metadata,
    Column('price_id', Integer, primary_key=True),
    Column('product_id', Integer, ForeignKey('product_dim.product_id')),
    Column('location_id', Integer, ForeignKey('location_dim.location_id')),
    Column('price', Float),
    Column('date_id', Integer, ForeignKey('date_dim.date_id'))
)

metadata.create_all(engine)

This step ensures referential integrity and avoids schema drift.

Writing Data to PostgreSQL

Once tables were set, PySpark’s JDBC writer handled efficient parallel writes to PostgreSQL.

clean_df.write \
    .format("jdbc") \
    .option("url", os.getenv('POSTGRES_URL')) \
    .option("dbtable", "fact_prices") \
    .option("user", os.getenv('POSTGRES_USER')) \
    .option("password", os.getenv('POSTGRES_PASSWORD')) \
    .mode("append") \
    .save()

With proper partitioning, this approach efficiently handled millions of rows.

Visualization with Grafana

Grafana provided a clean, dynamic way to visualize the data — connected directly to PostgreSQL.

Key Dashboards:

Price Trend Over Time
- Monthly price lines by commodity
- Compare multiple items
Latest Prices by Commodity and Region
- Price comparison by location
- Can be used to detect inflation by region
National Average Price Dashboard
- Avg price per commodity by latest month
- Used for cost-of-living analysis

Grafana panels were linked to SQL queries like:

SELECT
    d.year,
    d.month,
    p.commodity,
    AVG(f.price) AS avg_price
FROM fact_prices f
JOIN date_dim d ON f.date_id = d.date_id
JOIN product_dim p ON f.product_id = p.product_id
GROUP BY d.year, d.month, p.commodity
ORDER BY d.year, d.month;

Challenges & Lessons Learned

Inconsistent Units: Prices were listed as 90kg bags, 1kg tins, or liters. Unit normalization required extensive mapping logic.
Duplicate Commodity Names: Variants like “Beans (Dry)” vs “Dry Beans” — solved with fuzzy matching.
Spatial Mapping: Used latitude/longitude data to create a geographic layer for Grafana maps.
Performance Tuning: Spark’s .repartition() and broadcast joins improved load times.

Real-World Impact

Enables WFP or local governments to monitor inflation regionally
Helps farmers and consumers understand price dynamics
Open to integrating other indicators (e.g., rainfall, NDVI)
Ready for real-time (Kafka + Debezium integration possible)

Next Steps

Integrate Apache Airflow for automated ETL scheduling.
Use Prophet or ARIMA models to forecast inflation.
Add Kafka + Debezium CDC for real-time updates.
Expand visualizations with weather and trade data overlays.

Takeaways

This project blends engineering rigor with social impact — using open data to empower informed decisions. It demonstrates proficiency in:

Distributed data processing (PySpark)
SQL and data modeling best practices
Workflow orchestration (Airflow-ready design)
Visualization and analytics integration (Grafana)

It’s a strong addition to any data engineering portfolio — especially for roles focusing on public data, infrastructure, or analytics engineering.

Links

GitHub: github.com/RoseWabere/Kenyan-Food-Price-Tracker-and-Inflation-Analysis

Keywords: Kenya food prices, PySpark ETL, PostgreSQL data warehouse, Grafana dashboard, data engineering portfolio, Airflow pipeline, public data analytics, inflation analysis Kenya

What is Data Engineering?

Rose Wabere — Thu, 01 May 2025 23:34:04 +0000

With data being a highly valuable asset in today’s data-driven world, roles requiring expertise and knowledge on data have been on the rise as institutions hire data experts to help make the most out of the data obtained.

This article looks into data engineering, its roles and responsibilities, differences between data engineers and data scientists, skills needed, why one would consider a data engineering career path, where data engineers can work, and why companies need to hire data engineers.

Defining data engineering

Data engineering is a field focused on designing and building systems used in the collection, organization, and processing of data. When the data is collected, it is often in a raw and unstructured format that requires the input of a data engineer to transform it so it can be used in analysis and operations.

In an era where data is considered to be a valuable asset, data engineers are needed to help ensure data is available whenever and wherever it is needed. With businesses generating a lot of data, these professionals are tasked with preparing and optimizing the data for analysis and operations.

What are the roles and responsibilities of data engineers?

Some of the roles and responsibilities of data engineers include:

Data pipeline management and designing: Data engineers are tasked with designing and building automated ETL pipelines, which help ensure that data flows smoothly from sources to data warehouses or lakes.

Managing data infrastructure: As organizations increasingly accumulate more data, engineers help select and manage storage architectures that can handle the growing datasets efficiently and maintain performance and scalability.

Data modelling: Involves creating data models and schemas that make storing and retrieving data efficient.

Optimizing performance: This is achieved by improving query speeds, cost management, and streamlining the infrastructure.

Ensuring data security and compliance: It is the responsibility of a data engineer to ensure that policies on data access and usage are implemented.

Data engineering vs data science

The roles of data engineers and data scientists are often confused. Although they work together, they have differences in aspects such as goals, tasks, skills, and the tools they use.

Before delving into the differences, let’s look at what data science is. Data science is a multi-disciplinary field that integrates maths, artificial intelligence, analytics, and machine learning.

Goals: While data engineers aim to build infrastructure for data collection and make data accessible and usable, data scientists aim to use data to extract insights and predict outcomes while utilizing machine learning, thus enhancing and improving the decision-making processes.
Tasks: Data engineers are mainly tasked with building ETL pipelines, warehousing, and modelling, while data scientists are tasked with analytics, machine learning, and visualization.
Skills: Data engineers need skills in programming, databases, ETL, and architecture, while data scientists need skills like mathematics, statistics, and machine learning.
Tools used: Data engineers mainly use tools such as Python, Apache Spark, Kafka, Airflow, and SQL, while the main tools for data scientists are Python, R, TensorFlow, scikit-learn, Jupyter, and Notebooks.

Data engineering is the backbone of data science, analytics, and machine learning, as it helps make life easier for other data specialists. Data engineers will build the infrastructure to handle data, and the data scientists and analysts will use that data to gain insights crucial for making informed decisions.

Data engineering skills

Programming skills. Knowledge of programming languages such as Python, Java, or Scala is needed.
Database knowledge. Data engineers need to understand relational databases like SQL for structured data and NOSQL for unstructured data, which are essential in storing and managing data.
Familiarity with ETL (extract, transform, load). Data engineers deal with different data sources from which they have to clean the data and load it to the required destinations, thus they need to know how to use tools such as Apache Airflow.
Cloud computing knowledge. The expansive use and adoption of cloud platforms like AWS and Azure to store data necessitates that data engineers have cloud computing expertise.
Big data technologies. To handle huge data sets, it is essential to be familiar with frameworks such as Apache Kafka, Hadoop, and Apache Spark, which largely help examine complex data sets.
Problem-solving and analytical skills. Data engineers often work with complicated systems where they encounter problems that require them to be quick and effective.

Why consider a data engineering career?

High demand: Opportunities for data engineering continue to be on the rise as organizations increasingly become data-driven.
Lucrative opportunities: Data engineers are among the highest-paid professionals in the tech field for their expertise and skills.
Diverse opportunities: Work opportunities are not limited since the profession is spread across fields such as finance, retail, technology, and health care.
Diverse and transferable skillset: since data engineering combines skillsets in database management, cloud technologies, and systems architecture, it opens doors across different industries such as data science, data analytics, and software engineering, among others.
Basis for AI: Data engineers create pipelines and the infrastructure that powers AI and predictive analysis, without which analytics and machine learning would not be possible

Where do data engineers work?

Data engineers work in different industries such as media, finance, technology, education, healthcare, retail, and government. Let’s dive into some of their responsibilities in these sectors:

E-commerce and retail - A data engineer can build recommendation systems by processing a client’s behaviour, transactions, and product data. Most importantly, the data engineer will help streamline operations by enabling real-time analytics for inventory management, logistics, and forecasting.
Healthcare - He/she may integrate and process patient data to support diagnostics and personalized treatment, which helps to improve patient care and research.
Finance - A data engineer may create pipelines that can help in fraud detection, risk assessment, and customer analytics. The engineer will also play the critical role of ensuring data quality and integrity are maintained, which is critical for performing audits, filings, and making strategic decisions.
Government - With the government, data engineers undertake responsibilities such as ensuring data privacy and compliance, and integrating census, tax, and public service data.
Media - The data engineer can optimize content delivery, analyze viewer engagement, personalize user experiences, and support recommendation systems.

Why should companies hire data engineers?

Companies and institutions certainly need data engineers, as they help enable businesses to have a strategic advantage as well as utilize the full potential of their data in the following ways:

Data security. With data being a valuable asset and in an age where data breaches and privacy concerns are dominant, a data engineer will implement secure and reliable data systems, ensuring data safety and accessibility.
Cost saving. Data engineers help reduce costs by providing solutions tailored to the business needs. By optimizing data storage and processing, they also help reduce cost and improve resource utilization.
Scalability. Companies increasingly face large data volumes, and data engineers help ensure the data infrastructure remains flexible with the business needs. Scalability is especially crucial for businesses facing high growth and needing data solutions to keep up with the growth.
Improved analytics and insights. Data engineers create well-structured data pipelines and storage solutions that help data scientists and analysts access and analyze data easily, leading to quicker insights and timely decision-making.
Data quality: High data quality is crucial since inaccuracies can easily lead to flawed analysis and misguided business decisions. Data engineers will validate and cleanse the data and monitor systems to identify and fix irregularities in the datasets.

Conclusion

Data engineers are beyond technical specialists; they are business enablers, as their expertise in data plays a crucial role. By ensuring the required data is available at the right time and in the right format, they help institutions move faster, make better decisions, and remain competitive.

Data volumes will undoubtedly continue to grow, and so will advancements in technologies like AI, cloud computing, and big data, which further fuels the need for data engineers. It is the companies that invest in building strong data engineering teams that will be positioned for long-term success.

Forem: Rose Wabere

REST vs GraphQL vs WebSockets vs Webhooks: A Real-World Decision Guide (With Code)

async/await: The Foundation, Not the Feature

REST: Default Choice for Good Reason

GraphQL: REST With Client-Controlled Queries

WebSockets: When the Server Needs to Talk First

Webhooks: Event Notification Between Services

Real-World Architecture: All Four Together

When to Use What: Decision Framework

Common Mistakes (Those That Hurt in Production)

Building a Production-Grade Async Backend with FastAPI, SQLAlchemy, PostgreSQL, and Alembic

What You Are Actually Building

Sync vs Async: Why the Model Matters

Declarative Base: Defining Schema in Python

Async Engine and asyncpg

Sessions: The Most Critical Concept

Dependency Injection: Session Management at Scale

Transactions and Rollbacks

Relationships: Foreign Keys and Loading Strategies

Alembic: Schema Evolution

Production Project Structure

Common Mistakes That Break Production Systems

Conclusion

Fix “invalid literal for int() with base 10: 'None'” in FastAPI (Asyncpg + Docker)

Understanding the Error

The Code That Breaks

Step-by-Step Debugging

Why the Single-Env-Var Approach is Better

Handling Async with asyncpg

Full Working Example

Environment Variables in Docker Compose

Conclusion

Deploying FastAPI with Aiven PostgreSQL and Docker (Production-Ready Guide)

Step 1: FastAPI Project Setup

Step 2: Configure Aiven PostgreSQL

Step 3: Dockerize the App

Step 4: Docker Compose for Development

Step 5: Running SQL Scripts on Aiven

Step 6: Docker Compose for Production

Step 7: Environment Variables & Security

Conclusion

Nairobi Property Listings Scraping: 500+ Property Listings with Smart Size Extraction

The Challenge

Core Scraping Logic

The Size Extraction Problem

Why This Approach?

Integration with the Scraper

Error Handling & Resilience

Data Dictionary: Schema as Code

Lessons for Production Scrapers

Next Steps: Data Cleaning

Python #WebScraping #BeautifulSoup #DataScience #MachineLearning #Kenya #OpenSource #Regex

Fixing Kali Linux's "Failed to Set Beacon Parameters" Hotspot Error: A Troubleshooting Tale

The Scene of the Crime: A Standard Setup That Should Have Worked

The Investigation: Why Your Wi-Fi Card Refuses to Cooperate

The Solution: Introducing linux-wifi-hotspot

Step-by-Step: Building and Using Your Reliable Hotspot

The Takeaway: More Than Just a Fixed Hotspot

Building Production-Grade Data Analytics Pipelines: A Real-World Case Study in Government Data

The Problem

The Challenge: Data as a Governance Problem

The Solution: A Multi-Layer Data Architecture

Layer 1: Suppression Classification

Layer 2: Dual-Column Architecture

Layer 3: Intelligent Missing Data Handling

The Analysis: From Data to Decisions

Discovery #1: Disparities Emerge When You Measure Correctly

Discovery #2: Geography Is a Powerful Predictor

Discovery #3: Hierarchical Analysis Reveals System-Level Patterns

The Visualization: Making Data Speak

Lessons Learned: What Real Government Clients Actually Care About

Lesson #1: Data Quality > Data Quantity

Lesson #2: Preserve the Audit Trail

Lesson #3: Stakeholders Care About Impact, Not Methods

Lesson #4: One Number Changes Everything

Lesson #5: Government Work Requires Different Thinking

Technical Skills Showcased

The Business Impact

What I'd Do Differently

The Key Takeaway

The Solution: Introducing `linux-wifi-hotspot`