DEV Community: Marce

Architectural Patterns for High-Throughput Async Backends in Python: A Comprehensive Enterprise Guide

Marce — Sun, 11 Oct 2026 21:51:43 +0000

When building modern enterprise microservices, handling high-volume concurrent I/O efficiently is the difference between a resilient, scalable system and a cascading failure that brings down your entire infrastructure. While standard synchronous frameworks struggle under heavy loads, Python's asyncio ecosystem—combined with modern asynchronous ORMs, non-blocking database drivers, and resilient connection pooling—provides robust capabilities for high-throughput backends.

In this comprehensive architectural guide, we will dissect production-grade patterns for building non-blocking database queries, managing asynchronous connection pools, implementing resilient outbound HTTP calls with circuit breakers, preventing event loop starvation, and orchestrating distributed tasks under heavy concurrent workloads.

1. The Anatomy of Concurrency: Understanding the Event Loop at Scale

Traditional WSGI servers (like Gunicorn running synchronous Flask or Django apps) execute requests sequentially or block OS threads while waiting for database queries, network sockets, or third-party HTTP responses.

The Thread Pool Trap: Under high concurrency, thread pools saturate rapidly, leading to thread contention, excessive context switching, memory exhaustion due to heavy OS thread stacks (typically 1MB to 2MB per thread), and skyrocketing latency percentiles (p95, p99).
The ASGI Paradigm: Migrating to an ASGI server (such as Uvicorn or Hypercorn) coupled with FastAPI or native asyncio allows a single-threaded event loop to switch execution contexts during I/O wait times. This maximizes CPU efficiency and allows handling thousands of concurrent client connections without spawning heavy OS threads.

However, moving to async introduces dangerous architectural traps: blocking the event loop with CPU-bound code, synchronous database calls, or unoptimized connection pools will instantly neutralize your concurrency gains.

2. Advanced Non-Binding Database Operations with Async SQLAlchemy

A frequent performance bottleneck in Python backends is improper ORM configuration that silently blocks the event loop during SQL execution. To achieve true asynchronous performance, we must leverage SQLAlchemy's AsyncSession alongside an asynchronous database driver (asyncpg for PostgreSQL) with carefully tuned connection pool parameters.

Here is a robust, enterprise-grade implementation pattern for a repository layer handling async database transactions, connection pooling, and automated error rollbacks:

python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.future import select
from contextlib import asynccontextmanager
import logging

# Configure structured enterprise logger
logger = logging.getLogger("backend_core")

# Database URL pointing directly to the asyncpg native driver
DATABASE_URL = "postgresql+asyncpg://enterprise_user:secure_pass@localhost:5432/enterprise_db"

# Engine configured with explicit connection pooling for heavy multi-worker concurrency
engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=40,               # Maximum number of permanent connections to keep
    max_overflow=20,            # Additional connections allowed during traffic spikes
    pool_timeout=30.0,          # Seconds to wait before timing out on pool exhaustion
    pool_recycle=1800,          # Recycle connections every 30 minutes to prevent stale sockets
    pool_pre_ping=True          # Validate connection health before handing it to a session
)

AsyncSessionLocal = async_sessionmaker(
    engine, 
    expire_on_commit=False, 
    class_=AsyncSession
)

@asynccontextmanager
async def get_db_session() -> AsyncSession:
    """Context manager ensuring safe transaction lifecycle, automatic rollbacks, and cleanup."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception as e:
            await session.rollback()
            logger.error("Database transaction rolled back due to execution failure", extra={"error": str(e)})
            raise
        finally:
            await session.close()

3. Managing Concurrent External API Calls, Rate Limiting, and Timeouts

When a backend endpoint needs to aggregate data from multiple downstream microservices or third-party APIs, making sequential requests multiplies your system latency linearly ($O(N)$).Instead, we use asyncio.gather to execute operations concurrently. However, in production systems, unbounded concurrency can overwhelm downstream services or trigger rate limits. We must incorporate concurrency limiters (asyncio.Semaphore), connection reuse limits via httpx.Limits, and strict execution timeouts:

import asyncio
import httpx
import logging

logger = logging.getLogger("backend_core")

async def fetch_external_resource(
    client: httpx.AsyncClient, 
    url: str, 
    semaphore: asyncio.Semaphore
) -> dict:
    """Fetches an external resource respecting concurrency limits and strict timeouts."""
    async with semaphore:
        try:
            response = await client.get(url, timeout=4.0)
            response.raise_for_status()
            return {"url": url, "status": "success", "data": response.json()}
        except httpx.TimeoutException:
            logger.warning("External service request timed out", extra={"url": url})
            return {"url": url, "status": "timeout", "data": None}
        except httpx.HTTPError as exc:
            logger.warning(f"External service failed: {exc}", extra={"url": url})
            return {"url": url, "status": "error", "message": str(exc)}

async def aggregate_dashboard_metrics(user_id: int):
    urls = [
        f"[https://api.internal/v1/users/](https://api.internal/v1/users/){user_id}/orders",
        f"[https://api.internal/v1/users/](https://api.internal/v1/users/){user_id}/analytics",
        f"[https://api.internal/v1/users/](https://api.internal/v1/users/){user_id}/notifications",
        f"[https://api.internal/v1/users/](https://api.internal/v1/users/){user_id}/billing",
        f"[https://api.internal/v1/users/](https://api.internal/v1/users/){user_id}/preferences"
    ]

    # Limit concurrent outgoing requests to 15 to protect downstream services
    semaphore = asyncio.Semaphore(15)

    # Configure explicit connection limits for connection reuse (HTTP/1.1 keep-alive)
    limits = httpx.Limits(max_keepalive_connections=30, max_connections=100)

    async with httpx.AsyncClient(limits=limits) as client:
        tasks = [fetch_external_resource(client, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    return results

4. Preventing the Silent Killer: Event Loop Starvation

One of the most dangerous anti-patterns in asynchronous Python is executing heavy CPU-bound operations (such as cryptographic hashing, complex JWT validation loops, large JSON parsing, or image manipulation) directly inside an async route handler.

Because Python's event loop operates on a single thread per process, a heavy synchronous operation will freeze the entire event loop, causing all concurrent requests currently managed by that loop to stall completely.

Solution: Offloading to Process Pools
For CPU-bound operations, developers must offload work to a separate process pool executor using loop.run_in_executor:

import asyncio
import hashlib
import json

def cpu_heavy_cryptographic_workload(payload: str) -> str:
    """Simulates a heavy cryptographic hashing workload that would block an event loop."""
    data = payload
    for _ in range(200000):
        data = hashlib.sha256(data.encode()).hexdigest()
    return data

async def handle_secure_operation(payload: str):
    loop = asyncio.get_running_loop()

    # Offloads CPU-bound work to a separate process pool, preventing event loop starvation
    result = await loop.run_in_executor(None, cpu_heavy_cryptographic_workload, payload)
    return {"status": "success", "hashed_signature": result}

5. Observability: Structured JSON Logging in Distributed Microservices

In distributed systems and microservice architectures, standard string-based prints or basic text logs are completely insufficient for debugging race conditions, tracking distributed trace IDs, or analyzing latency anomalies. Enterprise-grade backends require structured JSON logs that can be seamlessly ingested by log aggregators (such as Datadog, ELK, or AWS CloudWatch).

import sys
import json
import logging
from datetime import datetime, timezone

class JSONFormatter(logging.Formatter):
    """Custom formatter to output logs in strict JSON format for observability pipelines."""
    def format(self, record):
        log_record = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
            "process_id": record.process
        }
        if hasattr(record, "extra") and isinstance(record.extra, dict):
            log_record.update(record.extra)
        return json.dumps(log_record)

# Initialize production logging pipeline
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())

root_logger = logging.getLogger("backend_core")
root_logger.setLevel(logging.INFO)
root_logger.handlers = [handler]
root_logger.propagate = False

6. Resilience Patterns: Implementing a Circuit Breaker for Fragile Dependencies

When dealing with third-party APIs or legacy microservices, a failing dependency can cause cascading failures across your entire application due to thread or connection exhaustion. Implementing a Circuit Breaker pattern protects your system by failing fast when a dependency is unhealthy.

import time
import logging

logger = logging.getLogger("backend_core")

class CircuitBreakerOpenException(Exception):
    pass

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF-OPEN
        self.last_failure_time = 0.0

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"
            self.last_failure_time = time.time()
            logger.error("Circuit breaker tripped to OPEN state due to consecutive failures.")

    async def execute(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF-OPEN"
                logger.info("Circuit breaker entering HALF-OPEN state for test probe.")
            else:
                raise CircuitBreakerOpenException("Circuit breaker is OPEN. Fast failing request.")

        try:
            result = await func(*args, **kwargs)
            if self.state == "HALF-OPEN":
                self.record_success()
            return result
        except Exception as e:
            self.record_failure()
            raise e

Conclusion

Transitioning from junior patterns to senior backend architecture requires shifting focus from just making code work to ensuring concurrency efficiency, non-blocking I/O integrity, protection against event loop starvation, fault tolerance via circuit breakers, and production-grade observability.

By enforcing connection pool discipline, structured JSON logging, process pool offloading, and strict async boundaries, your Python backends will scale gracefully under heavy enterprise workloads.

Building an ETL Pipeline with Python, Docker, and PostgreSQL (And Debugging the Real Errors)

Marce — Tue, 29 Sep 2026 00:14:25 +0000

Most ETL tutorials show a perfect, frictionless flow. The reality? My pipeline turned into a festival of KeyError's, outdated schemas, and API payload typos.

In this article, I will walk you through a complete Extrac -> transform -> Load pipeline using Python, Docker, and PostgreSqul. But more importantly, I'll share the real errors I ran into anh how I debugged them. Because calmly reading tracebacks is the only realible way to fix data pipelines when they inevitably fail.

What we'ew building

GitHub API -> extract.py -> transform.py -> load.py -> PostgreSQL (Docker)

Extract: Paginated fetch of issues from a public repository via the GitHub REST API.
Transform: Normalizes each issue into a flat record and calculates the hours it took to close.
Load: Creates the database table if it doesn't exist ans upserts by ID, so re-running the pipeline updates rows instead of duplicating them.

The Stack

Python 3.14
psycopg (v3) to connect to Postgres
python-dotenv for configuration management
PostgreSQL running in Docker Compose

Windows + Python 3.14 Note: If you are on Windows, use psycopg[binary] in your requirements instead of psycopg2-binary. The latter often fails to compile cleanly on this combination.

1. Proyect Structure

.
├── extract.py
├── transform.py
├── load.py
├── main.py
├── docker-compose.yml
├── requirements.txt
└── .env.example

2. Spinning up Postgres with Docker

First let's get our database running. Using Docker Compose keeps our local enviroment clean.

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: etl_user
      POSTGRES_PASSWORD: etl_password
      POSTGRES_DB: github_analytics
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Start the database with:

docker compose up -d

3. Configuration via `.env`

Create your .env file with the necessary credentials:

GITHUB_TOKEN=your_github_personal_access_token
GITHUB_REPO=microsoft/vscode
DB_HOST=localhost
DB_PORT=5432
DB_NAME=github_analytics
DB_USER=etl_user
DB_PASSWORD=etl_password

The Common Trap: KeyError: GitHUB_REPO **The Lesson:** __If you forget to create.envfrom.env.example, or if you forget callload_dotenv()inmain.py`, the script blows up immediately. Always chechks your enviroment variables first when a pipeline fails at startup.__

4. Extract: Pulling the Issues

We use the request library to handle pagination from the GitHub API.

`python
extract.py
import os
import logging
import requests
from tenacity import retry, stop_after_attempt, wait_exponential

logger = logging.getLogger(name)
GITHUB_API = "https://api.github.com"

def _headers():
token = os.environ.get("GITHUB_TOKEN")
headers = {"Accept": "application/vnd.github+json"}
if token:
headers["Authorization"] = f"Bearer {token}"
return headers

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _get(url, params=None):
response = requests.get(url, headers=_headers(), params=params, timeout=10)
response.raise_for_status()
return response

def fetch_issues(repo: str, max_pages: int = 5):
page = 1
while page <= max_pages:
logger.info("Fetching %s issues page %d", repo, page)
resp = _get(
f"{GITHUB_API}/repos/{repo}/issues",
params={"state": "all", "per_page": 100, "page": page},
)
batch = resp.json()
if not batch:
break
for item in batch:
yield item
page += 1
`

5. Transform: Normalizing and Computing Metrics

This is what a standard GitHub API response looks like for an issue

json { "id": 1234567, "number": 42, "title": "Bug in the auth module", "state": "closed", "created_at": "2024-01-01T10:00:00Z", "closed_at": "2024-01-02T12:00:00Z", "user": { "login": "octocat" } }

Here is our transformation logic:

`python
transform.py
from datetime import datetime

def normalize_issue(raw: dict) -> dict:
return {
"id": raw["id"],
"number": raw["number"],
"title": raw["title"],
"state": raw["state"],
"is_pull_request": "pull_request" in raw,
"created_at": raw["created_at"],
"closed_at": raw.get("closed_at"),
"comments": raw["comments"],
"author": raw["user"]["login"] if raw.get("user") else None,
}

def time_to_close_hours(row: dict):
if not row["closed_at"]:
return None
created = datetime.fromisoformat(row["created_at"].replace("Z", "+00:00"))
closed = datetime.fromisoformat(row["closed_at"].replace("Z", "+00:00"))
return round((closed - created).total_seconds() / 3600, 2)
`

The Typo Cascade: KeyError: "author"
** The Lesson:** I wrote raw["author"] assuming the fiels name, but GitHub uses raw["user]["login"]. Field names are a contract. Never guess them from memory. ALways validate your transformation logic against the actual JSON payload or API documentation

6. Load: Creating the Schema and Upserting

`python
load.py
import logging
import psycopg

logger = logging.getLogger(name)

CREATE_TABLE = """
CREATE TABLE IF NOT EXISTS issues (
id BIGINT PRIMARY KEY,
number INT,
title TEXT,
state TEXT,
is_pull_request BOOLEAN,
created_at TIMESTAMPTZ,
closed_at TIMESTAMPTZ,
comments INT,
author TEXT,
time_to_close_hours NUMERIC
);
"""

UPSERT = """
INSERT INTO issues (id, number, title, state, is_pull_request, created_at, closed_at, comments, author, time_to_close_hours)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
state = EXCLUDED.state,
closed_at = EXCLUDED.closed_at,
comments = EXCLUDED.comments,
time_to_close_hours = EXCLUDED.time_to_close_hours;
"""

def get_connection(db_config: dict):
return psycopg.connect(**db_config)

def ensure_schema(conn):
with conn.cursor() as cur:
cur.execute(CREATE_TABLE)
conn.commit()

def load_rows(conn, rows: list[dict]):
if not rows:
logger.info("No rows to load.")
return
values = [
(r["id"], r["number"], r["title"], r["state"], r["is_pull_request"],
r["created_at"], r["closed_at"], r["comments"], r["author"], r["time_to_close_hours"])
for r in rows
]
with conn.cursor() as cur:
cur.executemany(UPSERT, values)
conn.commit()
logger.info("Loaded/updated %d rows.", len(rows))
`

The Schema Error:
psycopg.errors.UndefinedColumn: column "closed_at" of relation "issues" does not exist
The Lesson: __CREATE TABLE IF NOT EXISTS does exactly what it says. If the table already existis, it does nothing-even if you added a new colum (closet_at) to your SQL string. It is a migration tool.

TO fix this during local development, drop the table using Docker Compose so the script can recreate it with the new schema:

`bash
docker compose exec postgres psql -U etl_user -d github_analytics -c "DROP TABLE IF EXISTS issues;"

7. Querying the Data

Once your main.py runs successfully, you can query your data directly through the container:

`bash
docker compose exec postgres psql -U etl_user -d github_analytics

What I took away from this project

None of the errors I ran into were mathematically complex -they were typos, schema mismatches, and misremembered fields. What mattered was the order in which I tackled them: one traceback at a time, always reading fro the bottom up to find the root cause.

If you're building your own ETL, my advice is simple: run the pipeline often, in small steps, and let each error lead you to the next one. It's the most reliable way to end up with a robust architecture.

Next Steps: Taking it to the Cloud

Running this locally in DOcker Compose is perfect for development and debugging, but the natural next step for any ETL pipeline is moving it to a production environment.

My next goal for this project is to migrate this local stack into the cloud -likely containerizing the Python scripts as a CronJob and hosting the PostgreDQL database on a fast developer-friendly cloud provider (like Civo's Kubernetes or Compute instances) to fully automate the extractionlayer.

Full code: The complete repository is on GitHub: github.com/WhoIsMarce/github-analytics-etl-pipeline

Over to you: What's the most ridiculous typo or silent error that ever broke your data pipeline? Drop it in the comments so I feel less alone!

WebSockets are Overkill: Build Real-Time Dashboards with Server-Sent Events (SSE) 🚀

Marce — Fri, 25 Sep 2026 16:00:27 +0000

When developers hear "real-time," their first instinct is usually to reach for WebSockets. But what if I told you that for many use cases, WebSockets are actually overkill?

If you are building a chat application or a multiplayer game where the client and server constantly talk to each other, WebSockets are perfect. However, if you are building a live metrics dashboard, a stock ticker, or a news feed, the communication is strictly one-way: the server pushes updates to the client.

For these unidirectional streams, Server-Sent Events (SSE) are a much better architectural choice.

In this tutorial, I'll show you how I built a lightweight, real-time server dashboard using Node.js and native SSE—no heavy libraries required.

🧠 Why SSE over WebSockets?

Native HTTP: SSE operates over standard HTTP/HTTPS. You don't need custom protocols, complex proxy configurations, or firewall bypasses.

Auto-Reconnect: The browser's native API handles connection drops and retries automatically out of the box.

Resource Efficiency: It uses a single, long-lived HTTP connection, making it much lighter on server resources for one-way data broadcasting.

🛠️ Step 1: The Node.js Backend

We need a simple Express server that configures the specific headers required for an SSE stream (text/event-stream).

const express = require('express');
const cors = require('cors');
const app = express();


app.use(cors());

app.get('/events', (req, res) => {
    // 1. Mandatory SSE Headers
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    // 2. Push data to the client every 3 seconds
    const intervalId = setInterval(() => {
        const data = JSON.stringify({
            message: "System update notification",
            activeUsers: Math.floor(Math.random() * (500 - 100 + 1) + 100),
            cpuLoad: (Math.random() * 100).toFixed(1) + "%",
            timestamp: new Date().toISOString()
        });

        // Note: The SSE format strictly requires "data: " and TWO newlines
        res.write(`data: ${data}\n\n`);
    }, 3000);

    // 3. Clean up when the client disconnects
    req.on('close', () => {
        clearInterval(intervalId);
    });
});

app.listen(3000, () => console.log('SSE Server running!'));

💻 Step 2: The Vanilla JS Frontend

The best part about SSE is that you don't need any external libraries on the frontend. Modern browsers have the EventSource API built-in.

// Connect to the backend stream
const eventSource = new EventSource('http://localhost:3000/events');

// Listen for incoming messages
eventSource.onmessage = function(event) {
    // Parse the JSON payload from the server
    const data = JSON.parse(event.data);

    console.log(`Update at ${data.timestamp}: CPU is at ${data.cpuLoad}`);

    // Here you would dynamically update your DOM elements
};

// Handle connection errors (the browser will auto-reconnect!)
eventSource.onerror = function(error) {
    console.error('SSE Error:', error);
};

🏁 Conclusion

While WebSockets often steal the spotlight, Server-Sent Events remain one of the most elegant, efficient, and underutilized tools in a backend developer's arsenal for real-time, one-way data streaming.

You can check out the full source code and the complete UI dashboard in My GitHub Repository Here.

Have you used SSE in production? Let me know in the comments!

I'm Marcela Zapata Vanegas, Technical Writer & Full Stack Developer. I specialize in building and documenting cloud-native solutions.

Data Cleaning Microservice: FastAPI + Pandas + Docker

Marce — Thu, 24 Sep 2026 18:02:35 +0000

A high-performance, containerized Python microservice designed to ingest messy CSV files, clean the data using Pandas, and return a structured JSON response.

This project demonstrates how to build and containerize a modern, asynchronous API capable of handling data transformation pipelines.

Architecture

FastAPI: A modern, fast (high-performance) web framework for building APIs with Python, built on standard Python type hints.
Pandas: The industry-standard Python library for data manipulation and analysis, used here to handle missing values and normalize dataset headers.
Docker: Containerizes the microservice (using a lightweight python:3.10-slim image) to ensure environment consistency and easy deployment on any cloud provider (e.g., Civo, AWS, DigitalOcean).

Prerequisites

Before you begin, ensure you have the following installed on your local machine:

Docker Desktop
Git

⚙️ Setup & Deployment

Clone the repository:

   git clone https://github.com/whoismarce/proyecto-fastapi-csv.git
   cd proyecto-fastapi-csv

Build the Docker Image: Execute the following command to build the container image. This will download the Python base image and install the required dependencies (fastapi, pandas, uvicorn, etc.).

   docker build -t fastapi-limpiador .

Run the Container: Spin up the microservice in detached mode on port 8000:

   docker run -d -p 8000:8000 --name api_python fastapi-limpiador

Testing via Swagger UI

One of the greatest features of FastAPI is its auto-generated, interactive API documentation. You don't need external tools like Postman to test this service.

Open your web browser and navigate to: http://localhost:8000/docs
You will see the Swagger UI dashboard. Click on the green POST /clean endpoint.
Click the "Try it out" button on the right side.
Click "Choose File" and upload the datos_sucios.csv sample file provided in this repository.
Click "Execute".
Scroll down to the "Server response" section to see the data seamlessly transformed from a messy CSV into clean, structured JSON.

Teardown

To stop the microservice and remove the container from your environment to free up resources, run the following commands:

# Stop the running container
docker stop api_python

# Remove the container
docker rm api_python

Developed by Marcela Zapata Vanegas - Technical Writer & Full Stack Developer.

Build a Local Automation Enviroment: Self-Hosted n8n + Node.js API with Docker Compose 🐳

Marce — Wed, 23 Sep 2026 22:52:27 +0000

When it comes to workflow automation, n8n is one of the most powerful open-source tools avalible. However, relying exclusively on third-party APIs or cloud instances can introduce latency, strict rate limits, and data privacy concerns.

The solution? A self-hosted, cloud-native architecture.

In this tutorial, I will Show you how to spin up a local development environment that orchestrates an n8n container and a custom Node.js REST API, communicating securely through an internal Docker bridge network.

The Architecture

Instead of exposing aour database or API to the public internet for n8n to consume, we will encapsulate both services within a Docker bridge network.

This provides two main benefits:

Security: Services can communicate with each other internally, but you maintain full control over which ports are exposed to the host machine.
POrtability: You can take this exact boilerplate and deploy it to a VPS (like Civo, DigitalOcean, or AWS) in seconds.

Step 1: Building the Node.js API

First, we need a backend service for n8n to interact with. We will use Express for its simplicity.
In the project root, create a subdirectory called API with an index.js file:

const express = require('express');
const app = express();
const PORT = process.env.PORT || 3000;

app.use(express.json());

//Endpoint de prueba que n8n va a consumir
app.get('/API/data', (req, res) => {
    res.json({
        success: true,
        message: "Hi from your API!",
        data: {
            id: 1,
            status: "active",
            timestamp: new Date()
        }
    });
} );

app.listen(PORT, () => {
    console.log('API de NODE.js running on port ${PORT}');
});

To containerize this, we add a lightweight Dockerfile based on the Node Alpine image:

# Usar una imagen oficial y ligera de Node.js
FROM node:18-alpine

#Crear y establecer el directotrio de trabajo dentro del contenedor
WORKDIR /usrc/app

#Copiar los archivos de dependencias
COPY package*.json ./

#Copiar el resto de código de la API
COPY . .

#Exponer el puerto que una la API
EXPOSE 3000

#Comando para iniciar la aplicación
CMD ["node", "index.js"]

Step 2: The Docker Compose Magic

This is where everything como together. In the root of our project, we create our docker-compose.yml file. This acts as the blueprint for our infraestructure.

services:
 my-api-node:
  build: ./API
  container_name: api_nodejs
  ports:
    - "3000:3000"
  networks:
    - n8n-network
  restart: unless-stopped

 n8n:
  image: docker.n8n.io/n8nio/n8n
  container_name: n8n_autoalojado
  ports:
    - "5678:5678"
  environment:
    - N8N_HOST=localhost
    - N8N_PORT=5678
    - N8N_PROTOCOL=http
    - NODE_ENV=production
    - WEBHOOK_URL=http://localhost:5678/
    - GENERIC_TIMEZONE=America/Argentina/Buenos_Aires
  volumes:
    - n8n_data:/home/node/.n8n
  networks:
    - n8n-network
  restart: unless-stopped
  depends_on:
    - my-api-node

networks:
  n8n-network:
    driver: bridge

volumes:
  n8n_data:

Key takeaways from this configuration:

Shared Network: Both services are tied to n8n-network. Inside n8n, instead of calling localhost:3000, we can make HTTP requests directly to https://api_nodejs:3000. Docker handles the DNS resolution automatically.
Data Persistence: The n8n_data volume ensures thet even if you tear down the container, your workflows remain safely stored on your local machine.

Step 3: Spin Up the Environment

With Docker running on your machine, open your terminal in the project root and execute:

docker compose up -d --build

The -d (detached) flag runs the services in the background, while the --build flag ensures our custom Node.js image is compiled with the latest changes.

That's it!

Navigate to https://localhost:5678 to access your n8n dashboard.
Navigate to https://localhost:3000/api/data to test your backend.

Conclusion

Setting up a local containerized environment saves you from dealing with cross-platform configuration issues, broken global dependencies, and Node.js version conflicts.

You can check out the complete source code for this boilerplate on my GitHub Repository: https://github.com/WhoIsMarce/proyecto-n8n-nodejs.git

Have you used n8n with local APIs before? Let me know in the comments what kind of automations you would build with this setup!

I'm Marcela Zapata Vanegas. Technical Writer & Full Stack Developer. I am passionate about simplifying cloud architecture and automating workflows.

Deploying Your Streaming Chatbot with Docker: From Localhost to a Live URL

Marce — Sun, 20 Sep 2026 12:26:05 +0000

In the previous article, we built a Node.js chatbot with real-time streaming, running from the terminal on our own machine. That's a great starting point, but it only works on your machine. To actually ship something — a demo people can open, a service other apps can call — it needs to run somewhere else, packaged in a way that behaves the same regardless of the underlying machine. That's what Docker and a deployment platform give us.

In this article, we'll take that same chatbot, add a small web interface, package it as a Docker image, publish it, and deploy it live with a public URL.

What you'll learn

Turning a script that exits into a long-running web service
Writing a Dockerfile for a Node.js app
Publishing an image to Docker Hub
Deploying a Docker image to Railway, with a public URL

Requirements

The chatbot project from the previous article
Docker Desktop installed and running
A free Docker Hub account
A free Railway account

A note on cost: Railway no longer has a permanent free tier — new accounts get a one-time trial credit (no card required to start). That's more than enough to build, test, and demo this project.

Step 1: From a script to a service

Our original chatbot.js runs three hardcoded questions and exits. That's fine for a terminal demo, but a container that exits immediately looks "crashed" to any hosting platform. We need something that keeps listening.

We also want something more interesting to look at than raw text in a browser tab — so this version serves a small web interface instead, with the same live-streaming effect visible on screen.

Create server.js (this sits alongside chatbot.js, which stays untouched):

require("dotenv").config();
const http = require("http");
const fs = require("fs");
const path = require("path");
const OpenAI = require("openai");

const client = new OpenAI({
  apiKey: process.env.GROQ_API_KEY,
  baseURL: "https://api.groq.com/openai/v1",
});

const MODEL = "openai/gpt-oss-120b";
const PORT = process.env.PORT || 3000;

const indexHtml = fs.readFileSync(path.join(__dirname, "public", "index.html"));

function readBody(req) {
  return new Promise((resolve, reject) => {
    let data = "";
    req.on("data", (chunk) => (data += chunk));
    req.on("end", () => resolve(data));
    req.on("error", reject);
  });
}

const server = http.createServer(async (req, res) => {
  if (req.method === "GET" && req.url === "/") {
    res.writeHead(200, { "Content-Type": "text/html; charset=utf-8" });
    res.end(indexHtml);
    return;
  }

  if (req.method === "POST" && req.url === "/chat") {
    const body = await readBody(req);
    let userMessage = "Hello!";
    try {
      userMessage = JSON.parse(body).message || userMessage;
    } catch (_) {}

    res.writeHead(200, {
      "Content-Type": "text/plain; charset=utf-8",
      "Cache-Control": "no-cache",
    });

    try {
      const stream = await client.chat.completions.create({
        model: MODEL,
        stream: true,
        messages: [{ role: "user", content: userMessage }],
      });

      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || "";
        if (text) res.write(text);
      }
      res.end();
    } catch (error) {
      res.end(`[error] ${error.message}`);
    }
    return;
  }

  res.writeHead(404, { "Content-Type": "text/plain" });
  res.end("Not found");
});

server.listen(PORT, () => {
  console.log(`Server listening on port ${PORT}`);
});

The / route serves a small HTML page (public/index.html) with an input box and a scrolling log; the /chat route is a streaming endpoint the page calls with fetch, reading the response body chunk by chunk with a ReadableStream — the same streaming pattern from Part 1, just moved from a terminal for await loop to a browser one. The full front-end code (a single self-contained HTML file, no build step) is in the GitHub repo.

Also add a start script in package.json:

"scripts": {
  "start": "node server.js"
}

Step 2: Write the Dockerfile

FROM node:20-alpine

WORKDIR /app

COPY package*.json ./
RUN npm install --omit=dev

COPY . .

ENV PORT=3000
EXPOSE 3000

CMD ["node", "server.js"]

And a .dockerignore, so the image doesn't ship your local dependencies or secrets:

node_modules
.env
.git
.gitignore

Step 3: Build and test locally

docker build -t llm-chatbot .
docker run -p 3000:3000 --env-file .env llm-chatbot

Open http://localhost:3000 — you should see the web interface, and streamed responses when you send a message.
Common snag: if you change your code and rebuild, but still see the old behavior, stop any running containers first (docker stop $(docker ps -q)) before rebuilding — an old container can still be holding the port.

Step 4: Publish to Docker Hub

docker login
docker tag llm-chatbot <your-username>/llm-chatbot-tutorial:latest
docker push <your-username>/llm-chatbot-tutorial:latest

Check hub.docker.com/r//llm-chatbot-tutorial — your image should be listed there, publicly.

Step 5: Deploy to Railway
Railway's dashboard flow for connecting a Docker Hub image can be inconsistent about actually triggering the first deploy — the CLI is the more reliable path:

npm install -g @railway/cli
railway login
railway link --project <your-project-name> --environment production --service <your-service-name>
railway up

railway up builds directly from your Dockerfile and deploys it — no dependency on the Docker Hub push actually triggering anything.

Set your API key as a variable (via the dashboard's Variables tab, or railway variables set GROQ_API_KEY=your_key), and generate a public domain from Settings → Networking → Generate Domain.

Common snag: if the generated domain gives "Application failed to respond," check the port. Railway assigns its own internal port to your container (visible in the deploy logs, e.g. "Server listening on port 8080") — make sure the port you entered when generating the domain matches that, not the 3000 you used locally.

Step 6: Confirm it's live
Visit your Railway-provided URL (something like your-project.up.railway.app) — you should see the same interface, publicly reachable, streaming responses in real time.

Conclusion
We took a terminal script and turned it into a containerized, publicly deployed web service — the same pattern used to ship real backend services. From here: adding a database, rate-limiting, or a CI/CD pipeline that deploys automatically on every push — which is exactly what we'll build next.

Next in this series: automating tests and deploys with GitHub Actions.

Building a Streaming Chatbot with Node.js

Marce — Fri, 18 Sep 2026 21:45:22 +0000

Most backend tutorials show synchronous calls: the API generates the full response and only then prints it, after several seconds of waiting. That’s not how real chat products work — in interfaces like ChatGPT or Claude, the response starts appearing word by word, before it’s even complete. That’s called streaming, and it’s what we’re going to build today.

What you’ll learn

Rendering LLM responses token by token in the console
Understanding the streaming architecture (real-time data events, similar to SSE)
A reusable pattern that works the same way with OpenAI, Anthropic, and Groq

Requirements

Node.js 18 or higher
npm
A free API key, no credit card required — we’ll use Groq

Step 1: Project setup

mkdir llm-chatbot-tutorial
cd llm-chatbot-tutorial
npm init -y
...

Install the dependencies. We use the openai package because Groq’s API is compatible with its format — the same SDK, just pointed at a different URL:

npm install openai dotenv

Step 2: Protect your credentials from the start

Create .gitignore :

node_modules/
.env

Create .env.example (this one does get committed):

GROQ_API_KEY=your_api_key_here

And your real .env (this one is never committed):

GROQ_API_KEY=your_api_key_here

Step 3: The chatbot code
Create chatbot.js :

require("dotenv").config();
const OpenAI = require("openai");

const client = new OpenAI({
  apiKey: process.env.GROQ_API_KEY,
  baseURL: "https://api.groq.com/openai/v1",
});

// Check console.groq.com/docs/models for the current model list —
// providers rotate models periodically.
const MODEL = "openai/gpt-oss-120b";

async function chat(userMessage) {
  console.log(`\n📝 User: ${userMessage}\n`);
  console.log("🤖 Assistant: ");

  try {
    const stream = await client.chat.completions.create({
      model: MODEL,
      stream: true,
      messages: [
        { role: "user", content: userMessage },
      ],
    });

    for await (const chunk of stream) {
      const text = chunk.choices[0]?.delta?.content || "";
      process.stdout.write(text);
    }

    console.log("\n");
  } catch (error) {
    console.error("Error:", error.message);
  }
}

async function main() {
  console.log("=== Streaming Chatbot with Node.js ===\n");
  await chat("What is the capital of Argentina?");
  await chat("Explain what Node.js is in two sentences.");
  await chat("Give me 3 tips for learning to code.");
  console.log("\n✅ Done!");
}

main();

What stream: true actually does
Without streaming, the API waits until the full response is generated and only then returns it as a single block. With stream: true , the server instead sends the response as a sequence of small fragments ( delta objects) as the model generates each token.
The for await loop processes each fragment as soon as it arrives. We use
process.stdout.write(text) instead of console.log(text) on purpose: console.log adds a line break after every call, which would break the live-typing visual effect.
This pattern stays the same across providers — only the SDK import and a couple of field names change. Once you understand this shape, you can swap providers with minimal effort.

Step 4: Run it

node chatbot.js

You should see the three questions answered one after another, with text streaming in real time —not appearing all at once.

*If you hit an error: *- Check that .env is in the project root - Confirm the variable is named exactly GROQ_API_KEY - Verify the configured model is still available at console.groq.com/docs/models

Step 5: Customize it

Change the questions — edit the chat(...) calls inside main() .
Add a system prompt to control the assistant’s tone or role:

messages: [
  { role: "system", content: "You are a concise, friendly coding tutor." },
  { role: "user", content: userMessage },
],

Swap the model — just change the MODEL constant. Always check the provider’s current model list before publishing or deploying, since free-tier models get deprecated and replaced over time.

Conclusion
You now have a working, streaming LLM chatbot in under 60 lines of Node.js — and, more importantly, you understand the streaming pattern that powers every modern AI chat interface. This is the foundation for anything more advanced: a web-based chat UI, a Slack bot, or a backend service that talks to an LLM.
Next in this series: containerizing this app with Docker and deploying it to the cloud.

DEV Community: Marce

Architectural Patterns for High-Throughput Async Backends in Python: A Comprehensive Enterprise Guide

1. The Anatomy of Concurrency: Understanding the Event Loop at Scale

2. Advanced Non-Binding Database Operations with Async SQLAlchemy

3. Managing Concurrent External API Calls, Rate Limiting, and Timeouts

4. Preventing the Silent Killer: Event Loop Starvation

5. Observability: Structured JSON Logging in Distributed Microservices

6. Resilience Patterns: Implementing a Circuit Breaker for Fragile Dependencies

Conclusion

Building an ETL Pipeline with Python, Docker, and PostgreSQL (And Debugging the Real Errors)

What we'ew building

The Stack

1. Proyect Structure

2. Spinning up Postgres with Docker

3. Configuration via .env

4. Extract: Pulling the Issues

5. Transform: Normalizing and Computing Metrics

6. Load: Creating the Schema and Upserting

7. Querying the Data

What I took away from this project

Next Steps: Taking it to the Cloud

WebSockets are Overkill: Build Real-Time Dashboards with Server-Sent Events (SSE) 🚀

Data Cleaning Microservice: FastAPI + Pandas + Docker

Architecture

Prerequisites

⚙️ Setup & Deployment

Testing via Swagger UI

Teardown

Build a Local Automation Enviroment: Self-Hosted n8n + Node.js API with Docker Compose 🐳

The Architecture

Step 1: Building the Node.js API

Step 2: The Docker Compose Magic

Step 3: Spin Up the Environment

Conclusion

Deploying Your Streaming Chatbot with Docker: From Localhost to a Live URL

Building a Streaming Chatbot with Node.js

3. Configuration via `.env`