DEV Community: Sebastian Nicolas Fuentes Avalos

🚀 Building Your First MCP Server: A Step-by-Step Guide

Sebastian Nicolas Fuentes Avalos — Mon, 01 Dec 2025 05:24:26 +0000

🧐 link to the practice repository:
PRESS ME

In the rapidly evolving landscape of AI, the ability to connect Large Language Models (LLMs) to your own data and tools is a game-changer. The Model Context Protocol (MCP) has emerged as a standard for this connection, allowing developers to build servers that expose data and functionality to MCP clients like Claude Desktop, VS Code, and others.

In this article, we will explore what an MCP server is and build a simple "Weather Server" from scratch using Python. By the end, you'll have a working server that you can connect to Claude to ask for weather updates!

What is an MCP Server?

An MCP Server is a lightweight application that implements the Model Context Protocol. It acts as a bridge between your data sources (databases, APIs, files) and an AI client. Instead of building custom integrations for every AI tool, you build one MCP server that any MCP-compliant client can use.

Key concepts:

Resources: Data that the server can read (like files or database rows).
Tools: Functions that the server can execute (like "get_weather" or "query_database").
Prompts: Pre-defined templates for interacting with the server.

Prerequisites

To follow this tutorial, you will need:

Python 3.10+ installed on your machine.
A code editor (like VS Code).
Claude Desktop (optional, for testing the client connection).

Step 1: Setting Up the Project

First, let's create a directory for our project and set up a virtual environment.

mkdir weather-mcp-server
cd weather-mcp-server
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Next, install the official MCP SDK for Python:

pip install mcp

Step 2: Writing the Server Code

We will create a simple server that exposes a single tool: get_weather. This tool will take a city name as input and return a mock weather report.

Create a file named server.py and add the following code:

from typing import Any
import asyncio
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server

# Initialize the server
server = Server("weather-demo")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    List available tools.
    This server provides a single tool to get weather information.
    """
    return [
        types.Tool(
            name="get_weather",
            description="Get the current weather for a specific city",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "The name of the city (e.g., London, Lima, New York)"
                    }
                },
                "required": ["city"]
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    Handle tool execution requests.
    """
    if name != "get_weather":
        raise ValueError(f"Unknown tool: {name}")

    if not arguments:
        raise ValueError("Missing arguments")

    city = arguments.get("city")

    # Mock data for demonstration purposes
    weather_data = {
        "London": "Cloudy, 15°C",
        "New York": "Sunny, 22°C",
        "Tokyo": "Rainy, 18°C",
        "Lima": "Sunny, 24°C",
        "Paris": "Windy, 12°C"
    }

    result = weather_data.get(city, f"Weather data not available for {city}")

    return [types.TextContent(type="text", text=result)]

async def main():
    # Run the server using stdin/stdout streams
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="weather-demo",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(main())

Step 3: Connecting to an MCP Client (Claude Desktop)

Now that our server is ready, let's connect it to Claude Desktop.

Open your Claude Desktop configuration file.
- macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
- Windows: %APPDATA%\Claude\claude_desktop_config.json
Add your server to the mcpServers object. You need to provide the full path to your python executable and the server.py file.

{
  "mcpServers": {
    "weather-demo": {
      "command": "absolute/path/to/venv/bin/python",
      "args": ["absolute/path/to/weather-mcp-server/server.py"]
    }
  }
}

Restart Claude Desktop.

Step 4: Testing the Server

Once Claude restarts, you should see a plug icon indicating that the MCP server is connected.

Try asking Claude:

"What is the weather in Lima?"

Claude will recognize the get_weather tool, call your local Python server, retrieve the data ("Sunny, 24°C"), and present it to you in the chat!

Conclusion

Congratulations! You've just built and connected your first MCP server. This simple example demonstrates the power of the Model Context Protocol: standardizing how AI models interact with external tools.

From here, you can expand your server to:

Query a real SQL database.
Fetch data from a live REST API.
Read and write files on your local system.

The possibilities are endless. Happy coding!

Resources:

🚀 Demystifying Observability: A Practical Guide with Python, Prometheus, and Grafana

Sebastian Nicolas Fuentes Avalos — Mon, 01 Dec 2025 05:12:21 +0000

🧐 link to the practice repository:
PRESS ME

In the modern era of distributed systems and microservices, "monitoring" is no longer enough. We need Observability. But what exactly is the difference, and how can we implement it without getting lost in complex configurations?

In this article, we will explore the core concepts of observability and build a real-world example using Python (Flask), Prometheus, and Grafana.

What is Observability?

While monitoring tells you when something is wrong (e.g., "The server is down"), observability allows you to understand why it is wrong by asking questions to your system from the outside.

It is often categorized into three pillars:

Logs: A record of discrete events (e.g., "User X logged in").
Metrics: Aggregated numerical data over time (e.g., "CPU usage is at 80%").
Traces: The path of a request through your distributed system.

Today, we will focus heavily on Metrics using the industry-standard tool: Prometheus.

The Stack

For our practical exercise, we will use:

Application: A simple Python Flask API.
Instrumentation: prometheus-client library to expose metrics.
Storage & Scraping: Prometheus.
Visualization: Grafana.

Step 1: The Application Code

We need an application that doesn't just work, but talks to us. We will create a Flask app and instrument it to track:

Request Count: How many requests we receive (broken down by status code and method).
Latency: How long requests take to process.

Here is the core logic (app/main.py):

from flask import Flask, Response, request
import time
import random
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

app = Flask(__name__)

# 1. Define Prometheus Metrics
REQUEST_COUNT = Counter(
    'app_request_count', 
    'Application Request Count', 
    ['method', 'endpoint', 'http_status']
)

REQUEST_LATENCY = Histogram(
    'app_request_latency_seconds', 
    'Application Request Latency', 
    ['method', 'endpoint']
)

@app.before_request
def before_request():
    request.start_time = time.time()

@app.after_request
def after_request(response):
    request_latency = time.time() - request.start_time
    # 2. Record metrics after every request
    REQUEST_LATENCY.labels(request.method, request.path).observe(request_latency)
    REQUEST_COUNT.labels(request.method, request.path, response.status_code).inc()
    return response

@app.route('/metrics')
def metrics():
    # 3. Expose metrics for Prometheus to scrape
    return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST)

# ... routes for / and /error ...

Step 2: Configuring Prometheus

Prometheus needs to know where to look for metrics. We create a prometheus.yml file:

global:
  scrape_interval: 5s

scrape_configs:
  - job_name: 'flask_app'
    static_configs:
      - targets: ['app:5000']

This tells Prometheus to ping our app on port 5000 every 5 seconds and read the data from /metrics.

Step 3: Orchestration with Docker Compose

To make this easy to run anywhere, we use Docker Compose to spin up the App, Prometheus, and Grafana simultaneously.

version: '3.8'
services:
  app:
    build: ./app
    ports: ["5000:5000"]
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
    ports: ["9090:9090"]
  grafana:
    image: grafana/grafana:latest
    ports: ["3000:3000"]

Seeing it in Action

Once the containers are running, we can generate some traffic to our app.

Generate Traffic: Refresh http://localhost:5000/ and http://localhost:5000/error a few times.
Check Prometheus: Go to http://localhost:9090 and search for app_request_count. You will see the raw data increasing.
Visualize in Grafana:
- Go to http://localhost:3000 (admin/admin).
- Add Prometheus as a Data Source (http://prometheus:9090).
- Create a dashboard to visualize rate(app_request_count[1m]).

Conclusion

Observability is not just for giant tech companies. With a few lines of code and open-source tools, you can gain deep insights into your application's health. Instead of guessing why your app is slow, you can look at the histogram metrics and know.

Happy Coding! 🚀

Facade Pattern — Catalog of Enterprise Application Architecture Patterns 🏢

Sebastian Nicolas Fuentes Avalos — Fri, 07 Nov 2025 04:52:34 +0000

🔗 Useful Links

Complete GitHub Repository - Complete source code

🎯 The Problem: Complex Service Orchestration

In modern enterprise application development, it's common to encounter systems that require coordination of multiple services to complete a business operation. Imagine you need to process an order in an e-commerce system:

Check inventory - Is stock available?
Process payment - Is the transaction valid?
Schedule shipping - How will it reach the customer?
Notify customer - How do we inform about the status?

Without proper design, the client code becomes complex and tightly coupled:

# ❌ Complex client code without Facade
def process_order_without_facade(customer, product, quantity, payment):
    # Client must know ALL internal details
    inventory = InventoryService()
    payment_gateway = PaymentGateway() 
    shipping = ShippingService()
    notifications = NotificationService()

    # Complex orchestration logic
    if not inventory.check_stock(product, quantity):
        raise Exception("Insufficient stock")

    if not inventory.reserve(product, quantity):
        raise Exception("Error reserving stock")

    try:
        receipt = payment_gateway.charge(payment, calculate_total(product, quantity))
        if not receipt.success:
            inventory.release(product, quantity)  # Manual rollback
            raise Exception(f"Payment failed: {receipt.message}")

        shipment = shipping.create_shipment(customer, [product])
        if not shipment.success:
            inventory.release(product, quantity)  # More manual rollback
            # What about refunds? More complex code?
            raise Exception("Shipping error")

        # Still need to notify the customer...

    except Exception as e:
        # Complex error handling prone to bugs
        inventory.release(product, quantity)
        raise e

Evident problems:

🔗 High coupling: Client knows internal details
🔄 Complex rollback logic: Difficult to maintain
🐛 Error-prone: Many failure points
📈 Hard to scale: Adding services increases complexity

💡 The Solution: Facade Pattern

The Facade pattern provides a unified and simplified interface for a set of subsystems, hiding their internal complexity and providing a single entry point.

✨ Key Benefits

🎯 Simplicity: One interface for multiple operations
🔓 Decoupling: Client doesn't depend on internal implementations
🛡️ Abstraction: Hides subsystem complexity
🧪 Testable: Easy creation of mocks and tests
🔧 Maintainable: Internal changes don't affect client

🏗️ Pattern Architecture

graph TB
    Client[👤 Client] --> Facade[🏛️ OrderFacade]

    Facade --> Inventory[📦 InventoryService]
    Facade --> Payment[💳 PaymentGateway] 
    Facade --> Shipping[🚚 ShippingService]
    Facade --> Notification[📧 NotificationService]

    Inventory --> DB1[(📋 Stock DB)]
    Payment --> Bank[🏦 Bank API]
    Shipping --> Carrier[📦 Carrier API]
    Notification --> Email[📧 Email Service]

🚀 Practical Implementation in Python

1. Defining the Main Facade

from dataclasses import dataclass
from typing import Optional
from decimal import Decimal

@dataclass
class OrderResult:
    """Unified result of order operation."""
    success: bool
    order_id: Optional[str] = None
    reason: Optional[str] = None
    transaction_id: Optional[str] = None
    tracking_number: Optional[str] = None
    total_amount: Optional[Decimal] = None

class OrderFacade:
    """
    Facade that simplifies subsystem orchestration
    for enterprise order processing.
    """

    def __init__(self, inventory=None, payments=None, 
                 shipping=None, notifications=None):
        # Dependency injection for flexibility
        self.inventory = inventory or InventoryService()
        self.payments = payments or PaymentGateway()
        self.shipping = shipping or ShippingService()
        self.notifications = notifications or NotificationService()

    def place_order(self, customer_id: str, sku: str, qty: int,
                   payment_info: dict, unit_price: float) -> OrderResult:
        """
        ✨ ONE SINGLE METHOD to process a complete order.
        The Facade orchestrates all subsystems internally.
        """
        order_id = str(uuid.uuid4())

        try:
            # 🔍 Step 1: Verify and reserve inventory
            if not self._reserve_inventory(sku, qty):
                return OrderResult(False, order_id, "Insufficient stock")

            # 💳 Step 2: Process payment
            total_amount = Decimal(str(qty * unit_price))
            payment_result = self._process_payment(payment_info, total_amount)
            if not payment_result.success:
                self._rollback_inventory(sku, qty)
                return OrderResult(False, order_id, f"Payment failed: {payment_result.message}")

            # 🚚 Step 3: Schedule shipping
            shipping_result = self._create_shipment(customer_id, sku, qty)
            if not shipping_result.success:
                self._rollback_inventory(sku, qty)
                return OrderResult(False, order_id, f"Shipping error: {shipping_result.message}",
                                 transaction_id=payment_result.transaction_id)

            # 📧 Step 4: Notify customer
            self._notify_customer(customer_id, order_id, payment_result.transaction_id, 
                                shipping_result.tracking_number)

            # ✅ Success: Return complete result
            return OrderResult(
                success=True,
                order_id=order_id,
                transaction_id=payment_result.transaction_id,
                tracking_number=shipping_result.tracking_number,
                total_amount=total_amount
            )

        except Exception as e:
            # 🛡️ Centralized error handling
            self._handle_unexpected_error(sku, qty, order_id, str(e))
            return OrderResult(False, order_id, f"Internal error: {str(e)}")

    def _reserve_inventory(self, sku: str, qty: int) -> bool:
        """Encapsulates inventory reservation logic."""
        return (self.inventory.check_stock(sku, qty) and 
                self.inventory.reserve(sku, qty))

    def _process_payment(self, payment_info: dict, amount: Decimal):
        """Encapsulates payment processing."""
        return self.payments.charge(payment_info, float(amount))

    def _create_shipment(self, customer_id: str, sku: str, qty: int):
        """Encapsulates shipment creation."""
        return self.shipping.create_shipment(
            customer_id, 
            [{"sku": sku, "qty": qty}]
        )

    def _notify_customer(self, customer_id: str, order_id: str, 
                        transaction_id: str, tracking_number: str):
        """Encapsulates customer notifications."""
        self.notifications.send_order_notification(
            customer_id, 
            "order_confirmed",
            {
                "order_id": order_id,
                "transaction_id": transaction_id,
                "tracking_number": tracking_number
            }
        )

    def _rollback_inventory(self, sku: str, qty: int):
        """Centralized inventory rollback handling."""
        self.inventory.release(sku, qty)

    def _handle_unexpected_error(self, sku: str, qty: int, order_id: str, error: str):
        """Centralized unexpected error handling."""
        try:
            self._rollback_inventory(sku, qty)
        except:
            pass  # Log error but don't fail twice

        # Log error for monitoring
        print(f"CRITICAL ERROR - Order {order_id}: {error}")

2. Orchestrated Subsystems

📦 Inventory Service

class InventoryService:
    """Manages product stock and reservations."""

    def __init__(self):
        self._stock = {"LAPTOP-15": 5, "MONITOR-27": 10, "TABLET-10": 3}

    def check_stock(self, sku: str, qty: int) -> bool:
        """Verifies stock availability."""
        return self._stock.get(sku, 0) >= qty

    def reserve(self, sku: str, qty: int) -> bool:
        """Reserves products from inventory."""
        if self.check_stock(sku, qty):
            self._stock[sku] -= qty
            return True
        return False

    def release(self, sku: str, qty: int) -> None:
        """Releases reservations in case of rollback."""
        self._stock[sku] = self._stock.get(sku, 0) + qty

💳 Payment Gateway

import uuid
from dataclasses import dataclass

@dataclass
class PaymentReceipt:
    success: bool
    transaction_id: str = ""
    message: str = ""

class PaymentGateway:
    """Processes financial transactions."""

    def charge(self, payment_info: dict, amount: float) -> PaymentReceipt:
        """Processes a card charge."""
        card_number = payment_info.get("card_number", "")

        # Basic validations
        if not card_number or len(card_number) < 15:
            return PaymentReceipt(False, "", "Invalid card")

        if amount <= 0:
            return PaymentReceipt(False, "", "Invalid amount")

        # Processing simulation by card type
        if card_number.startswith("4"):  # Visa
            return PaymentReceipt(
                True, 
                str(uuid.uuid4()), 
                "Payment processed successfully"
            )
        elif card_number.startswith("5"):  # MasterCard
            return PaymentReceipt(
                True, 
                str(uuid.uuid4()), 
                "Payment processed with MasterCard"
            )
        else:
            return PaymentReceipt(False, "", "Unsupported card")

🚚 Shipping Service

@dataclass
class ShipmentInfo:
    success: bool
    shipment_id: str = ""
    tracking_number: str = ""
    eta_days: int = 0
    message: str = ""

class ShippingService:
    """Manages logistics and shipment tracking."""

    def create_shipment(self, customer_id: str, items: list) -> ShipmentInfo:
        """Creates a new shipment."""
        if not items:
            return ShipmentInfo(False, message="No items to ship")

        shipment_id = str(uuid.uuid4())
        tracking_number = f"TRK{shipment_id[:8].upper()}"

        return ShipmentInfo(
            success=True,
            shipment_id=shipment_id,
            tracking_number=tracking_number,
            eta_days=3,
            message="Shipment scheduled successfully"
        )

3. Simplified Facade Usage

# ✅ SIMPLE client code with Facade
def process_order_with_facade():
    facade = OrderFacade()

    # ONE single call for the entire operation
    result = facade.place_order(
        customer_id="customer_123",
        sku="LAPTOP-15",
        qty=1, 
        payment_info={"card_number": "4111111111111111", "cvv": "123"},
        unit_price=899.99
    )

    # Simple result handling
    if result.success:
        print(f"✅ Order successful!")
        print(f"📦 ID: {result.order_id}")
        print(f"💳 Transaction: {result.transaction_id}")
        print(f"🚚 Tracking: {result.tracking_number}")
        print(f"💰 Total: ${result.total_amount}")
    else:
        print(f"❌ Error: {result.reason}")
        # Rollback already handled internally by the Facade

# Lines of code comparison:
# Without Facade: ~50 complex lines with error handling
# With Facade: ~15 simple and clear lines

🧪 Testing the Facade Pattern

One of the great advantages of the Facade pattern is that it greatly facilitates testing through dependency injection of mocks:

import pytest
from unittest.mock import Mock

class TestOrderFacade:
    def test_successful_order(self):
        """Test complete successful order flow."""
        # Arrange: Create subsystem mocks
        mock_inventory = Mock()
        mock_inventory.check_stock.return_value = True
        mock_inventory.reserve.return_value = True

        mock_payments = Mock()
        mock_payments.charge.return_value = PaymentReceipt(
            success=True, 
            transaction_id="tx-123"
        )

        mock_shipping = Mock()
        mock_shipping.create_shipment.return_value = ShipmentInfo(
            success=True, 
            tracking_number="TRK123"
        )

        mock_notifications = Mock()

        # Inject mocks into Facade
        facade = OrderFacade(
            inventory=mock_inventory,
            payments=mock_payments, 
            shipping=mock_shipping,
            notifications=mock_notifications
        )

        # Act: Execute operation
        result = facade.place_order(
            "customer_1", "LAPTOP-15", 1, 
            {"card_number": "4111111111111111"}, 899.99
        )

        # Assert: Verify result and interactions
        assert result.success is True
        assert result.transaction_id == "tx-123"
        assert result.tracking_number == "TRK123"

        # Verify all subsystems were called
        mock_inventory.check_stock.assert_called_once()
        mock_payments.charge.assert_called_once()
        mock_shipping.create_shipment.assert_called_once()
        mock_notifications.send_order_notification.assert_called_once()

    def test_payment_error_handling(self):
        """Test rollback when payment fails."""
        # Simulate payment failure
        mock_inventory = Mock()
        mock_inventory.check_stock.return_value = True
        mock_inventory.reserve.return_value = True

        mock_payments = Mock()
        mock_payments.charge.return_value = PaymentReceipt(
            success=False, 
            message="Card declined"
        )

        facade = OrderFacade(inventory=mock_inventory, payments=mock_payments)

        result = facade.place_order(
            "customer_1", "LAPTOP-15", 1,
            {"card_number": "1234567890123456"}, 899.99
        )

        # Verify it failed correctly
        assert result.success is False
        assert "Payment failed" in result.reason

        # Verify inventory rollback was performed
        mock_inventory.release.assert_called_once_with("LAPTOP-15", 1)

📊 Demo in Action

I've created an executable demo that shows the pattern working:

# Run complete demonstration
python -m src.order_facade.demo

# Expected results:

============================================================
  DEMO 1: SUCCESSFUL ORDERS
============================================================

🛒 Processing standard order...

=== Processing Order abc12345... ===
Customer: customer_001
Product: MONITOR-27 x 1
Unit price: $299.99

[Step 1] Verifying inventory...
[Inventory] Reserved 1 units of MONITOR-27. Remaining stock: 9

[Step 2] Processing payment...
Total: $309.99 (includes shipping)
[Payment] Successful charge: $309.99 on Visa card ****1111

[Step 3] Scheduling shipping...
[Shipping] Shipment created: TRK12345678 via National Mail
[Shipping] Estimated delivery: 2025-11-10

[Step 4] Sending notifications...
[Email] Order confirmed - ID: abc12345...
[SMS] Tracking number: TRK12345678

✅ Order abc12345... processed successfully!

🎯 Scenario: Standard Order - Monitor 27"
✅ Status: SUCCESS
📦 Order ID: abc12345...
💳 Transaction ID: tx-67890...
🚚 Tracking Number: TRK12345678
💰 Total Paid: $309.99
📅 Estimated Delivery: 2025-11-10

🏗️ When to Use the Facade Pattern

✅ Ideal Cases

Systems with multiple subsystems that need orchestration
Complex APIs you want to simplify for clients
Business processes involving multiple services
Gradual migration from legacy systems
Testing systems with many dependencies

❌ When NOT to Use It

Simple operations that don't require orchestration
When you need direct access to specific functionalities
Small systems without subsystem complexity
Performance critical scenarios where every call counts

🎯 Real-World Examples

E-commerce: Order processing (like our example)
Banking: Transfers involving multiple validations
Healthcare: Appointment systems coordinating doctors, rooms, equipment
Logistics: Package tracking through multiple carriers
Enterprise Software: Workflows integrating CRM, ERP, and billing systems

🚀 Demonstrated Benefits

For Business

⚡ Faster development: Less repetitive code
🐛 Fewer bugs: Centralized and tested logic
🔧 Easy maintenance: Internal changes without client impact
📈 Scalability: Easy to add new subsystems

For Developers

🧪 Simplified testing: Easy-to-create mocks
📖 More readable code: Clear and well-documented interface
🔄 Reusability: Facade used in multiple contexts
🛡️ Error handling: Centralized and consistent

For Client/User

🎯 Simple API: One call for complex operations
🔒 Consistency: Predictable behavior
⚡ Performance: Transparent internal optimizations
📊 Observability: Integrated logging and metrics

🎯 Conclusions and Best Practices

✅ Key Principles Applied

Single Responsibility: Facade orchestrates, doesn't implement business logic
Open/Closed: Easy to extend with new subsystems
Dependency Inversion: Depends on interfaces, not concrete implementations
Interface Segregation: Specific interface for each responsibility

🛡️ Best Practices

# ✅ DO: Dependency injection for flexibility
class OrderFacade:
    def __init__(self, inventory=None, payments=None):
        self.inventory = inventory or DefaultInventoryService()
        self.payments = payments or DefaultPaymentGateway()

# ✅ DO: Return structured results
@dataclass 
class OrderResult:
    success: bool
    reason: Optional[str] = None
    # More fields as needed

# ✅ DO: Centralized error handling
def place_order(self, ...):
    try:
        # Main logic
        pass
    except Exception as e:
        return self._handle_error(e, context)

# ❌ DON'T: Turn Facade into a God Object
# Don't implement ALL logic in the Facade, delegate to subsystems

# ❌ DON'T: Hide ALL subsystem functionality
# Allow direct access when necessary

🔮 Evolution and Extensibility

# Easy to add new subsystems
class OrderFacade:
    def __init__(self, inventory=None, payments=None, 
                 fraud_detection=None, analytics=None):  # ← New services
        # ...

    def place_order(self, ...):
        # Step 1: Check inventory
        # Step 2: Detect fraud ← New step
        # Step 3: Process payment  
        # Step 4: Ship
        # Step 5: Analytics ← New step
        # Step 6: Notify

📚 Additional Resources

🛠️ Technologies Used

Python 3.8+ - Main language
Pytest - Testing framework
Dataclasses - Data structures
Type Hints - Type documentation
UUID - Unique ID generation

📖 To Continue Learning

Experiment with the code: clone the repo and modify subsystems
Extend functionality: add new product types or payment methods
Practice testing: create new test cases and mocks
Apply in real projects: identify opportunities in your current code

Did you like this article? Give it a ❤️ and share it with other developers. Have questions or suggestions? Leave them in the comments!

designpatterns #python #softwarearchitecture #enterprisepatterns #facade #softwaredevelopment #programming #coding #python3

Other Visualization Tools: Streamlit, Dash, and Bokeh for Dashboards & Reports 🧑‍🏫

Sebastian Nicolas Fuentes Avalos — Mon, 15 Sep 2025 05:22:13 +0000

Introduction

Data visualization is a key component in business intelligence and analytics. While tools like Power BI and Tableau are popular, Python offers powerful open-source alternatives for building interactive dashboards and reports: Streamlit, Dash, and Bokeh. These tools allow rapid development and deployment of data apps, making them ideal for data scientists and analysts.

Streamlit

Streamlit is a Python library that makes it easy to create interactive web apps for data visualization with minimal code.

Example: Simple Dashboard

# streamlit_app.py
import streamlit as st
import pandas as pd
import numpy as np

data = pd.DataFrame(
    np.random.randn(100, 3),
    columns=['A', 'B', 'C']
)
st.title('Streamlit Dashboard Example')
st.line_chart(data)

Deploying to the Cloud

You can deploy Streamlit apps for free using Streamlit Cloud. Just push your code to GitHub and connect your repo in Streamlit Cloud.

Dash

Dash, developed by Plotly, is a framework for building analytical web applications with Python.

Example: Interactive Dashboard

# dash_app.py
import dash
from dash import html, dcc
import plotly.express as px
import pandas as pd

app = dash.Dash(__name__)
df = pd.DataFrame({"x": range(10), "y": [i**2 for i in range(10)]})
fig = px.line(df, x="x", y="y", title="Dash Example")

app.layout = html.Div([
    html.H1("Dash Dashboard Example"),
    dcc.Graph(figure=fig)
])

if __name__ == "__main__":
    app.run_server(debug=True)

Deploying to the Cloud

Dash apps can be deployed on Heroku or Render. Both platforms offer free tiers for small projects.

Bokeh

Bokeh is a Python library for creating interactive visualizations for modern web browsers.

Example: Interactive Plot

# bokeh_app.py
from bokeh.plotting import figure, output_file, show
output_file("bokeh_example.html")
p = figure(title="Bokeh Line Example", x_axis_label='x', y_axis_label='y')
p.line([1, 2, 3, 4, 5], [6, 7, 2, 4, 5], legend_label="Temp.", line_width=2)
show(p)

Deploying to the Cloud

Bokeh apps can be deployed using Bokeh Server on cloud platforms like Heroku or AWS.

Practice what you learned! 😎👍

GitHub Repository

Conclusion

Streamlit, Dash, and Bokeh are excellent choices for building interactive dashboards and reports in Python. They are easy to use, flexible, and can be deployed to the cloud for sharing insights with your team or the world.

Thank you for your time!

Enterprise Design Patterns: Applying Catalog Patterns for Robust Applications 👨‍🏫

Sebastian Nicolas Fuentes Avalos — Sun, 14 Sep 2025 03:00:20 +0000

Enterprise applications power the world's businesses, from banks to e-commerce platforms. Building robust, maintainable, and scalable enterprise systems requires not just good code, but the right architectural patterns. In this article, we'll explore several patterns from Martin Fowler's Catalog of Patterns of Enterprise Application Architecture, see how they solve real-world problems, and implement one in Python.

What Are Enterprise Design Patterns?

Enterprise design patterns are proven solutions to common problems encountered when building large, complex business systems. They provide best practices for organizing code, managing data, handling business logic, and supporting scalability.

Martin Fowler's catalog is a goldmine for anyone serious about enterprise development. Key patterns include:

Layered Architecture
Domain Model
Data Mapper
Service Layer
Repository
Active Record
Table Module
Unit of Work
(and many more)

Real-World Example: Service Layer and Repository Patterns in Python

Let’s take a common scenario: building a user management system for an enterprise app.

We'll use the Service Layer pattern to encapsulate business logic and the Repository pattern to abstract data access. This separation makes our code maintainable and testable.

1. Layered Architecture

We split our app into layers:

Presentation Layer: Handles HTTP requests, CLI, or UI.
Service Layer: Business logic.
Data Access Layer: Interacts with the database.

2. Repository Pattern

The Repository acts as a mediator between the domain and data mapping layers, providing access to data objects without exposing implementation details.

3. Service Layer Pattern

Encapsulates business logic, orchestrating domain objects and repositories.

Python Implementation Example

# domain.py
class User:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email

# repository.py
class UserRepository:
    def __init__(self):
        self._users = {}

    def add(self, user):
        self._users[user.user_id] = user

    def get(self, user_id):
        return self._users.get(user_id)

    def list_all(self):
        return list(self._users.values())

# service.py
class UserService:
    def __init__(self, repository: UserRepository):
        self.repository = repository

    def register_user(self, user_id, name, email):
        if self.repository.get(user_id):
            raise Exception("User already exists!")
        user = User(user_id, name, email)
        self.repository.add(user)
        return user

    def get_user(self, user_id):
        return self.repository.get(user_id)

    def list_users(self):
        return self.repository.list_all()

# main.py (presentation layer)
if __name__ == "__main__":
    repo = UserRepository()
    service = UserService(repo)

    service.register_user(1, "Alice", "alice@example.com")
    service.register_user(2, "Bob", "bob@example.com")

    print("All users:")
    for user in service.list_users():
        print(f"{user.user_id}: {user.name} ({user.email})")

Benefits:

Separation of concerns: Business logic is separated from data access.
Testability: Service and repository layers can be tested independently.
Maintainability: Changes in storage (for example, moving from in-memory to a database) do not affect business logic.

Other Patterns to Explore

Unit of Work: Manages changes to business objects and coordinates writing out changes as a single transaction.
Active Record: Domain objects also handle their own persistence.
Table Module: Encapsulates all business logic for all rows in a database table.
Domain Model: Complex business logic is captured in domain objects.

Repository Link

You can find a full implementation and more examples in the companion repository:

https://github.com/SebastianFuentesAvalos/article02_enterprise-design-patterns

References

Fowler, M. Patterns of Enterprise Application Architecture
SOLID Principles
Python Official Docs

Written by Sebastian Fuentes Avalos

Software Design Principles: Building Robust Applications in Python 🧑‍🏫

Sebastian Nicolas Fuentes Avalos — Sun, 14 Sep 2025 02:15:50 +0000

Software design principles are fundamental guidelines that help developers create code that is reliable, maintainable, and scalable. By following these principles, teams can reduce technical debt, ensure better collaboration, and deliver high-quality software.

Why Do Design Principles Matter?

Poorly designed software often leads to hard-to-maintain code, bugs, and frustration among developers. Design principles provide a foundation for writing code that is easy to understand, modify, and extend.

Let’s explore some essential software design principles, and see how they can be applied in a real Python project.

Key Principles

1. Single Responsibility Principle (SRP)

Definition:

A class should have only one reason to change, meaning it should have only one job or responsibility.

Why?

If a class does too much, changes in one part can unintentionally affect another, leading to bugs and confusion.

2. Open/Closed Principle (OCP)

Definition:

Software entities (classes, modules, functions) should be open for extension but closed for modification.

Why?

You should be able to add new features by adding new code, not by changing existing, stable code.

3. Liskov Substitution Principle (LSP)

Definition:

Objects of a superclass should be replaceable with objects of a subclass without affecting the correctness of the program.

Why?

This ensures that inheritance is used properly and that subclasses enhance, not break, the behavior of the base class.

4. Interface Segregation Principle (ISP)

Definition:

Clients should not be forced to depend on interfaces they do not use.

Why?

It’s better to have several small, specific interfaces than one big, general-purpose interface.

5. Dependency Inversion Principle (DIP)

Definition:

High-level modules should not depend on low-level modules. Both should depend on abstractions.

Why?

This reduces coupling between different parts of your application.

Example: Applying SRP and OCP in Python

Let’s build a simple notification system that sends messages via email or SMS. We'll apply the Single Responsibility Principle (SRP) and Open/Closed Principle (OCP).

Initial (Bad) Design

class Notifier:
    def send(self, message, type):
        if type == "email":
            # email sending logic
            print(f"Sending EMAIL: {message}")
        elif type == "sms":
            # sms sending logic
            print(f"Sending SMS: {message}")

Problems:

The Notifier class has multiple responsibilities.
Every time a new notification type is added, we have to modify the class (OCP violation).

Refactored (Good) Design

Let’s use SRP by separating each notification type, and OCP by allowing new types via extension.

from abc import ABC, abstractmethod

class NotificationSender(ABC):
    @abstractmethod
    def send(self, message):
        pass

class EmailSender(NotificationSender):
    def send(self, message):
        print(f"Sending EMAIL: {message}")
        # Actual email logic here

class SMSSender(NotificationSender):
    def send(self, message):
        print(f"Sending SMS: {message}")
        # Actual SMS logic here

def notify(sender: NotificationSender, message: str):
    sender.send(message)

# Usage example
email_sender = EmailSender()
sms_sender = SMSSender()

notify(email_sender, "Hello via Email!")
notify(sms_sender, "Hello via SMS!")

Benefits:

Each sender class has a single responsibility (SRP).
New senders (e.g., PushSender) can be added without modifying existing code (OCP).

You can find the complete code and exercises in the companion repository:

SebastianFuentesAvalos/article1_software-design-principles-python

Conclusion

Applying software design principles like SRP and OCP leads to cleaner, more maintainable, and scalable code. In Python, using abstract base classes and separation of concerns makes it easy to follow these guidelines.

If you want to see more principles in action, you can extend this example—try adding a PushSender for push notifications!

References:

Written by Sebastián Fuentes Avalos

Testing Management Tools Comparative: Real-World Examples

Sebastian Nicolas Fuentes Avalos — Thu, 03 Jul 2025 15:24:32 +0000

Introduction

In modern software development, automated testing and continuous integration have become essential practices. Testing management tools help teams automate their testing pipelines, ensuring code quality and faster deployment cycles. This article compares popular testing management tools with real-world examples and practical implementations.

1. GitHub Actions

GitHub Actions is a powerful CI/CD platform that allows you to automate your software workflows directly in your GitHub repository.

Key Features:

Native GitHub integration
Matrix builds for multiple environments
Extensive marketplace of actions
Free tier with generous limits

Real-World Example:

# .github/workflows/ci.yml
name: CI/CD Pipeline

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest

    strategy:
      matrix:
        node-version: [14.x, 16.x, 18.x]

    steps:
    - uses: actions/checkout@v3

    - name: Use Node.js ${{ matrix.node-version }}
      uses: actions/setup-node@v3
      with:
        node-version: ${{ matrix.node-version }}
        cache: 'npm'

    - name: Install dependencies
      run: npm ci

    - name: Run unit tests
      run: npm run test:unit

    - name: Run integration tests
      run: npm run test:integration

    - name: Generate coverage report
      run: npm run coverage

    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v3
      with:
        token: ${{ secrets.CODECOV_TOKEN }}

  security-scan:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3

    - name: Run security audit
      run: npm audit --audit-level high

    - name: OWASP ZAP Scan
      uses: zaproxy/action-full-scan@v0.4.0
      with:
        target: 'http://localhost:3000'

2. GitLab CI/CD

GitLab CI/CD is an integrated part of GitLab that provides powerful pipeline capabilities with excellent Docker integration.

Key Features:

Built-in container registry
Advanced pipeline visualization
Kubernetes integration
Auto DevOps capabilities

Real-World Example:

# .gitlab-ci.yml
stages:
  - build
  - test
  - security
  - deploy

variables:
  DOCKER_IMAGE: $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA
  POSTGRES_DB: test_db
  POSTGRES_USER: test_user
  POSTGRES_PASSWORD: test_pass

before_script:
  - apt-get update -qq && apt-get install -y -qq git curl

build:
  stage: build
  image: docker:20.10.16
  services:
    - docker:20.10.16-dind
  script:
    - docker build -t $DOCKER_IMAGE .
    - docker push $DOCKER_IMAGE

unit_tests:
  stage: test
  image: node:16
  services:
    - postgres:13
  script:
    - npm install
    - npm run test:unit
  coverage: '/Statements\s*:\s*([^%]+)/'
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage/cobertura-coverage.xml

integration_tests:
  stage: test
  image: $DOCKER_IMAGE
  services:
    - postgres:13
    - redis:6
  script:
    - npm run test:integration
  artifacts:
    reports:
      junit: test-results.xml

security_scan:
  stage: security
  image: registry.gitlab.com/gitlab-org/security-products/sast:latest
  script:
    - /analyzer run
  artifacts:
    reports:
      sast: gl-sast-report.json

deploy_staging:
  stage: deploy
  script:
    - kubectl apply -f k8s/staging/
  environment:
    name: staging
    url: https://staging.example.com
  only:
    - develop

3. Jenkins

Jenkins is one of the most established CI/CD tools, offering extensive plugin ecosystem and flexibility.

Key Features:

Massive plugin ecosystem (1800+ plugins)
Highly customizable
Pipeline as Code (Jenkinsfile)
Distributed builds

Real-World Example:

// Jenkinsfile
pipeline {
    agent any

    tools {
        nodejs '16.x'
        maven '3.8.1'
    }

    environment {
        DOCKER_REGISTRY = 'your-registry.com'
        APP_NAME = 'your-app'
    }

    stages {
        stage('Checkout') {
            steps {
                checkout scm
            }
        }

        stage('Install Dependencies') {
            steps {
                sh 'npm install'
            }
        }

        stage('Code Quality') {
            parallel {
                stage('ESLint') {
                    steps {
                        sh 'npm run lint'
                    }
                }
                stage('SonarQube Analysis') {
                    steps {
                        withSonarQubeEnv('SonarQube') {
                            sh 'npm run sonar'
                        }
                    }
                }
            }
        }

        stage('Testing') {
            parallel {
                stage('Unit Tests') {
                    steps {
                        sh 'npm run test:unit'
                    }
                    post {
                        always {
                            publishTestResults testResultsPattern: 'test-results.xml'
                            publishCoverage adapters: [
                                coberturaAdapter('coverage/cobertura-coverage.xml')
                            ]
                        }
                    }
                }
                stage('Integration Tests') {
                    steps {
                        sh 'docker-compose up -d'
                        sh 'npm run test:integration'
                    }
                    post {
                        always {
                            sh 'docker-compose down'
                        }
                    }
                }
                stage('E2E Tests') {
                    steps {
                        sh 'npm run test:e2e'
                    }
                    post {
                        always {
                            archiveArtifacts artifacts: 'cypress/screenshots/**', allowEmptyArchive: true
                            archiveArtifacts artifacts: 'cypress/videos/**', allowEmptyArchive: true
                        }
                    }
                }
            }
        }

        stage('Security Scan') {
            steps {
                sh 'npm audit --audit-level high'
                sh 'docker run --rm -v $(pwd):/app owasp/zap2docker-stable zap-baseline.py -t http://localhost:3000'
            }
        }

        stage('Build Docker Image') {
            steps {
                script {
                    def image = docker.build("${DOCKER_REGISTRY}/${APP_NAME}:${BUILD_NUMBER}")
                    image.push()
                    image.push('latest')
                }
            }
        }

        stage('Deploy') {
            when {
                branch 'main'
            }
            steps {
                sh 'kubectl apply -f k8s/production/'
            }
        }
    }

    post {
        always {
            cleanWs()
        }
        failure {
            emailext (
                subject: "Build Failed: ${env.JOB_NAME} - ${env.BUILD_NUMBER}",
                body: "Build failed. Check console output at ${env.BUILD_URL}",
                to: "${env.CHANGE_AUTHOR_EMAIL}"
            )
        }
    }
}

4. CircleCI

CircleCI offers cloud-based CI/CD with excellent Docker support and performance optimization.

Key Features:

Cloud-native architecture
Excellent Docker support
Advanced caching mechanisms
Workflow orchestration

Real-World Example:

# .circleci/config.yml
version: 2.1

orbs:
  node: circleci/node@5.0.0
  docker: circleci/docker@2.1.0
  kubernetes: circleci/kubernetes@1.3.0

executors:
  node-executor:
    docker:
      - image: cimg/node:16.14
      - image: cimg/postgres:13.3
        environment:
          POSTGRES_USER: test_user
          POSTGRES_DB: test_db
          POSTGRES_PASSWORD: test_pass

jobs:
  install-dependencies:
    executor: node-executor
    steps:
      - checkout
      - restore_cache:
          keys:
            - v1-dependencies-{{ checksum "package-lock.json" }}
            - v1-dependencies-
      - run: npm ci
      - save_cache:
          paths:
            - node_modules
          key: v1-dependencies-{{ checksum "package-lock.json" }}
      - persist_to_workspace:
          root: .
          paths:
            - node_modules

  code-quality:
    executor: node-executor
    steps:
      - checkout
      - attach_workspace:
          at: .
      - run:
          name: ESLint
          command: npm run lint
      - run:
          name: Prettier Check
          command: npm run format:check
      - run:
          name: TypeScript Check
          command: npm run type-check

  unit-tests:
    executor: node-executor
    steps:
      - checkout
      - attach_workspace:
          at: .
      - run:
          name: Run Unit Tests
          command: npm run test:unit -- --coverage
      - store_test_results:
          path: test-results
      - store_artifacts:
          path: coverage

  integration-tests:
    executor: node-executor
    steps:
      - checkout
      - attach_workspace:
          at: .
      - run:
          name: Wait for DB
          command: dockerize -wait tcp://localhost:5432 -timeout 1m
      - run:
          name: Run Integration Tests
          command: npm run test:integration
      - store_test_results:
          path: test-results

  security-scan:
    executor: node-executor
    steps:
      - checkout
      - attach_workspace:
          at: .
      - run:
          name: Security Audit
          command: npm audit --audit-level high
      - run:
          name: OWASP Dependency Check
          command: |
            curl -L https://github.com/jeremylong/DependencyCheck/releases/download/v7.1.1/dependency-check-7.1.1-release.zip -o dependency-check.zip
            unzip dependency-check.zip
            ./dependency-check/bin/dependency-check.sh --scan . --format JSON --out dependency-check-report.json
      - store_artifacts:
          path: dependency-check-report.json

  build-and-push:
    executor: docker/docker
    steps:
      - checkout
      - setup_remote_docker
      - docker/check
      - docker/build:
          image: myapp
          tag: << pipeline.number >>
      - docker/push:
          image: myapp
          tag: << pipeline.number >>

  deploy:
    executor: kubernetes/default
    steps:
      - checkout
      - kubernetes/install-kubectl
      - run:
          name: Deploy to Kubernetes
          command: |
            kubectl set image deployment/myapp myapp=myapp:<< pipeline.number >>
            kubectl rollout status deployment/myapp

workflows:
  version: 2
  build-test-deploy:
    jobs:
      - install-dependencies
      - code-quality:
          requires:
            - install-dependencies
      - unit-tests:
          requires:
            - install-dependencies
      - integration-tests:
          requires:
            - install-dependencies
      - security-scan:
          requires:
            - install-dependencies
      - build-and-push:
          requires:
            - code-quality
            - unit-tests
            - integration-tests
            - security-scan
          filters:
            branches:
              only: main
      - deploy:
          requires:
            - build-and-push
          filters:
            branches:
              only: main

5. Travis CI

Travis CI is a hosted CI service that integrates well with GitHub repositories.

Key Features:

Easy GitHub integration
Matrix builds
Deployment to multiple providers
Open source friendly

Real-World Example:

# .travis.yml
language: node_js
node_js:
  - "14"
  - "16"
  - "18"

services:
  - postgresql
  - redis-server
  - docker

env:
  global:
    - NODE_ENV=test
    - DATABASE_URL=postgres://postgres@localhost/test_db
    - REDIS_URL=redis://localhost:6379
  matrix:
    - TEST_SUITE=unit
    - TEST_SUITE=integration
    - TEST_SUITE=e2e

addons:
  postgresql: "13"
  chrome: stable

cache:
  directories:
    - node_modules
    - "$HOME/.npm"

before_script:
  - psql -c 'create database test_db;' -U postgres
  - npm run db:migrate
  - npm run db:seed

script:
  - |
    case "$TEST_SUITE" in
      unit)
        npm run test:unit -- --coverage
        ;;
      integration)
        npm run test:integration
        ;;
      e2e)
        npm run test:e2e
        ;;
    esac

after_script:
  - npm run coverage:report

after_success:
  - bash <(curl -s https://codecov.io/bash)

before_deploy:
  - npm run build
  - docker build -t $DOCKER_USERNAME/myapp:$TRAVIS_BUILD_NUMBER .

deploy:
  - provider: script
    script: bash scripts/deploy.sh
    skip_cleanup: true
    on:
      branch: main
      condition: $TEST_SUITE = unit
  - provider: dockerhub
    username: $DOCKER_USERNAME
    password: $DOCKER_PASSWORD
    repository: $DOCKER_USERNAME/myapp
    tag: $TRAVIS_BUILD_NUMBER
    on:
      branch: main
      condition: $TEST_SUITE = unit

notifications:
  email:
    on_success: change
    on_failure: always
  slack:
    secure: [encrypted-slack-token]

Best Practices for Testing Management

1. Pipeline Structure

Use parallel execution for independent tests
Implement proper dependency management
Cache dependencies and artifacts
Use matrix builds for cross-platform testing

2. Testing Strategy

Unit Tests: Fast, isolated, high coverage
Integration Tests: Test component interactions
E2E Tests: User workflow validation
Security Tests: Vulnerability scanning
Performance Tests: Load and stress testing

3. Quality Gates

Code coverage thresholds
Security vulnerability limits
Performance benchmarks
Code quality metrics

4. Deployment Strategies

Blue-green deployments
Canary releases
Feature flags
Rollback capabilities

Example Repository Structure

project/
├── .github/
│   └── workflows/
│       └── ci.yml
├── .gitlab-ci.yml
├── Jenkinsfile
├── .circleci/
│   └── config.yml
├── .travis.yml
├── src/
├── tests/
│   ├── unit/
│   ├── integration/
│   └── e2e/
├── docker/
├── k8s/
├── scripts/
└── package.json

Conclusion

Each testing management tool has its strengths and ideal use cases:

GitHub Actions: Best for GitHub-hosted projects with simple to moderate complexity
GitLab CI/CD: Excellent for teams using GitLab with strong DevOps requirements
Jenkins: Most flexible and powerful for complex enterprise environments
CircleCI: Great performance and Docker support for cloud-native applications
Travis CI: Simple and effective for open-source projects

The choice depends on your team's specific needs, existing infrastructure, and complexity requirements. Consider factors like integration capabilities, pricing, learning curve, and long-term maintenance when making your decision.

Public Repository Examples

You can find complete working examples of these configurations in these public repositories:

🧪 Mastering API Testing: Practical Guide with Postman and Newman for Automation

Sebastian Nicolas Fuentes Avalos — Mon, 09 Jun 2025 04:58:09 +0000

🌍 Introduction

API testing is fundamental in modern software development, especially in microservices architectures and distributed applications. A robust API testing framework allows us to validate functionality, performance, and security of our endpoints before reaching production.

In this article, we'll explore how to implement a complete API testing framework using Postman and Newman, automating it with GitHub Actions and creating a practical demonstration project.

🔧 Tools and Setup

Newman Installation

Newman is the command-line version of Postman, perfect for CI/CD integration:

npm install -g newman
npm install -g newman-reporter-html

Verify installation:

newman --version

Project Configuration

Testing project structure:

api-testing-framework/
├── collections/
│   └── user-api-tests.postman_collection.json
├── environments/
│   ├── dev.postman_environment.json
│   └── prod.postman_environment.json
├── tests/
│   └── integration/
├── reports/
├── .github/workflows/
│   └── api-tests.yml
└── package.json

🎯 Practical Implementation

Demo API We create a simple REST API to demonstrate testing:

// server.js - Simple users API
const express = require('express');
const app = express();
app.use(express.json());

let users = [
  { id: 1, name: "John Doe", email: "john@example.com" },
  { id: 2, name: "Jane Smith", email: "jane@example.com" }
];

// GET /users - List users
app.get('/users', (req, res) => {
  res.json({ users, total: users.length });
});

// POST /users - Create user
app.post('/users', (req, res) => {
  const { name, email } = req.body;

  if (!name || !email) {
    return res.status(400).json({ error: 'Name and email are required' });
  }

  const newUser = {
    id: users.length + 1,
    name,
    email
  };

  users.push(newUser);
  res.status(201).json(newUser);
});

// GET /users/:id - Get specific user
app.get('/users/:id', (req, res) => {
  const user = users.find(u => u.id === parseInt(req.params.id));

  if (!user) {
    return res.status(404).json({ error: 'User not found' });
  }

  res.json(user);
});

app.listen(3000, () => {
  console.log('API running on port 3000');
});

Postman Test Collection Example of automated tests in Postman:

// Pre-request Script to generate dynamic data
pm.globals.set("randomName", pm.variables.replaceIn('{{$randomFirstName}} {{$randomLastName}}'));
pm.globals.set("randomEmail", pm.variables.replaceIn('{{$randomEmail}}'));

// Test Scripts - Automatic validations
pm.test("Status code is 200", function () {
    pm.response.to.have.status(200);
});

pm.test("Response time is less than 500ms", function () {
    pm.expect(pm.response.responseTime).to.be.below(500);
});

pm.test("Response contains users array", function () {
    const jsonData = pm.response.json();
    pm.expect(jsonData).to.have.property('users');
    pm.expect(jsonData.users).to.be.an('array');
});

pm.test("User has required fields", function () {
    const jsonData = pm.response.json();
    if (jsonData.users && jsonData.users.length > 0) {
        const user = jsonData.users[0];
        pm.expect(user).to.have.property('id');
        pm.expect(user).to.have.property('name');
        pm.expect(user).to.have.property('email');
    }
});

// JSON Schema validation
const schema = {
    type: "object",
    properties: {
        users: {
            type: "array",
            items: {
                type: "object",
                properties: {
                    id: { type: "number" },
                    name: { type: "string" },
                    email: { type: "string", format: "email" }
                },
                required: ["id", "name", "email"]
            }
        },
        total: { type: "number" }
    },
    required: ["users", "total"]
};

pm.test("Schema validation", function () {
    pm.response.to.have.jsonSchema(schema);
});

Environment Configuration

// environments/dev.postman_environment.json
{
    "name": "Development",
    "values": [
        {
            "key": "baseUrl",
            "value": "http://localhost:3000",
            "enabled": true
        },
        {
            "key": "apiKey",
            "value": "dev-api-key-123",
            "enabled": true
        }
    ]
}

🤖 GitHub Actions Automation

# .github/workflows/api-tests.yml
name: API Testing Framework

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]
  schedule:
    - cron: '0 8 * * *'  # Run daily at 8 AM UTC

jobs:
  api-tests:
    runs-on: ubuntu-latest

    services:
      api:
        image: node:16
        ports:
          - 3000:3000

    steps:
      - name: Checkout repository
        uses: actions/checkout@v3

      - name: Setup Node.js
        uses: actions/setup-node@v3
        with:
          node-version: '16'
          cache: 'npm'

      - name: Install dependencies
        run: |
          npm install
          npm install -g newman newman-reporter-html

      - name: Start API server
        run: |
          npm start &
          sleep 5  # Wait for server to start

      - name: Run API Health Check
        run: |
          curl -f http://localhost:3000/users || exit 1

      - name: Execute Postman Collection
        run: |
          newman run collections/user-api-tests.postman_collection.json \
            -e environments/dev.postman_environment.json \
            --reporters cli,html \
            --reporter-html-export reports/api-test-report.html \
            --bail

      - name: Upload Test Reports
        uses: actions/upload-artifact@v3
        if: always()
        with:
          name: api-test-reports
          path: reports/

      - name: Performance Testing
        run: |
          newman run collections/user-api-tests.postman_collection.json \
            -e environments/dev.postman_environment.json \
            --iteration-count 50 \
            --reporters cli,json \
            --reporter-json-export reports/performance-report.json

      - name: Security Tests
        run: |
          # Run additional security tests
          newman run collections/security-tests.postman_collection.json \
            -e environments/dev.postman_environment.json \
            --reporters cli

📊 Reporting and Monitoring

Results Analysis Script

// scripts/analyze-results.js
const fs = require('fs');
const path = require('path');

function analyzeTestResults(reportPath) {
    const report = JSON.parse(fs.readFileSync(reportPath, 'utf8'));

    const summary = {
        totalTests: report.run.stats.tests.total,
        passedTests: report.run.stats.tests.total - report.run.stats.tests.failed,
        failedTests: report.run.stats.tests.failed,
        averageResponseTime: calculateAverageResponseTime(report.run.executions),
        successRate: ((report.run.stats.tests.total - report.run.stats.tests.failed) / report.run.stats.tests.total * 100).toFixed(2)
    };

    console.log('📊 Test Results Summary:');
    console.log(`✅ Passed: ${summary.passedTests}/${summary.totalTests}`);
    console.log(`❌ Failed: ${summary.failedTests}`);
    console.log(`⚡ Avg Response Time: ${summary.averageResponseTime}ms`);
    console.log(`📈 Success Rate: ${summary.successRate}%`);

    if (summary.failedTests > 0) {
        console.log('\n🚨 Failed Tests:');
        report.run.executions.forEach(execution => {
            execution.assertions.forEach(assertion => {
                if (assertion.error) {
                    console.log(`- ${assertion.assertion}: ${assertion.error.message}`);
                }
            });
        });
    }

    return summary;
}

function calculateAverageResponseTime(executions) {
    const totalTime = executions.reduce((sum, exec) => sum + exec.response.responseTime, 0);
    return Math.round(totalTime / executions.length);
}

// Run analysis
if (process.argv[2]) {
    analyzeTestResults(process.argv[2]);
}

🔗 Demo Project

👉 GitHub Repository: API Testing Framework Demo
The repository includes:

✅ Complete demo API
✅ Configured Postman collections
✅ CI/CD automation
✅ Detailed HTML reports
✅ Performance testing
✅ Security validations

Running the Project Locally

# Clone the repository
git clone https://github.com/SebastianFuentesAvalos/api-testing-framework-demo.git
cd api-testing-framework-demo

# Install dependencies
npm install

# Start the API
npm start

# In another terminal, run the tests
newman run collections/user-api-tests.postman_collection.json \
  -e environments/dev.postman_environment.json \
  --reporters cli,html \
  --reporter-html-export reports/test-report.html

📹 Video Demonstration

A complete 5-minute demonstration showing:

🎯 Testing framework configuration
🧪 Automated test execution
📊 Report analysis
🤖 CI/CD integration in action

🎯 Best Practices

Test Organization
Separate by functionality: Group tests by modules or features
Use descriptive names: GET_Users_Should_Return_Valid_Schema
Implement data-driven testing: Use CSV/JSON files for test data
Robust Validations

// Complete API response validation
pm.test("Complete API Response Validation", function() {
    const response = pm.response.json();

    // Status and response time
    pm.expect(pm.response.code).to.be.oneOf([200, 201, 202]);
    pm.expect(pm.response.responseTime).to.be.below(1000);

    // Security headers
    pm.expect(pm.response.headers.get('Content-Type')).to.include('application/json');
    pm.expect(pm.response.headers.get('X-Content-Type-Options')).to.eql('nosniff');

    // Data structure
    pm.expect(response).to.have.property('data');
    pm.expect(response.data).to.be.an('array');

    // Business logic validation
    if (response.data.length > 0) {
        response.data.forEach(item => {
            pm.expect(item.id).to.be.a('number').and.above(0);
            pm.expect(item.email).to.match(/^[^\s@]+@[^\s@]+\.[^\s@]+$/);
        });
    }
});

Test Data Management

// Cleanup and setup test data
pm.test("Test Data Cleanup", function() {
    // Clean up previous test data
    pm.sendRequest({
        url: pm.environment.get("baseUrl") + "/test-cleanup",
        method: 'DELETE',
        header: {
            'Authorization': 'Bearer ' + pm.environment.get("testToken")
        }
    }, function(err, response) {
        pm.expect(response.code).to.be.oneOf([200, 404]);
    });
});

🧾 Conclusion

A well-implemented API testing framework is crucial for maintaining quality and reliability in modern applications. The combination of Postman and Newman provides a complete solution that spans from manual testing to full CI/CD automation.

Key benefits implemented:

✅ Early error detection before production
✅ Complete automation with scheduled execution
✅ Detailed reports for analysis and decision making
✅ Seamless integration with existing DevOps workflows
✅ Scalability for teams of any size By integrating these practices into your development workflow, you'll significantly reduce production bugs and increase confidence in your releases.

This is the complete article in markdown format ready for Dev.to publication! The article includes all the necessary elements:

- Proper markdown formatting with headers (`##`, `###`)
- Code blocks with syntax highlighting
- Emojis for visual appeal
- Real-world examples
- Complete project structure
- GitHub repository reference
- Video demonstration section
- Best practices
- Tags for discoverability

You can copy this directly into Dev.to's editor or save it as a `.md` file.