DEV Community: Diego Fernando Castillo Mamani

# Unit of Work: Managing Database Transactions Like a Pro with Python

Diego Fernando Castillo Mamani — Wed, 24 Jun 2026 06:21:04 +0000

Introduction

Every serious backend developer eventually faces the same problem: you need to make multiple changes to a database as part of a single business operation, and you need all of them to succeed or none of them to go through. Partial updates are worse than no updates at all - they leave your data in an inconsistent state that can be nearly impossible to debug in production.
This is not a new problem. Enterprise developers have been solving it for decades, and Martin Fowler documented the canonical solution in his 2002 book Patterns of Enterprise Application Architecture: the Unit of Work pattern.
In this article we are going to go deep on what Unit of Work is, why it exists, how it works internally, and how to build a clean, production-quality implementation from scratch in Python using only the standard library. By the end you will have a working implementation you can adapt to any project, and a solid understanding of how popular frameworks like SQLAlchemy and Django ORM implement this pattern under the hood.
The full source code is available on GitHub:
👉 github.com/diegocastillo12/unit-of-work-python

- ## Background: What is the Unit of Work Pattern? The Unit of Work pattern is part of Martin Fowler's catalog of Patterns of Enterprise Application Architecture (PoEAA), a collection of battle-tested solutions for common problems in enterprise software design. Fowler defines it as follows: > "A Unit of Work maintains a list of objects affected by a business transaction and coordinates the writing out of changes and the resolution of concurrency problems." Let's unpack that definition carefully. "Maintains a list of objects affected by a business transaction" - this means the Unit of Work acts as a tracker. When your business logic creates a new object, modifies an existing one, or marks one for deletion, it does not immediately write to the database. Instead, it registers the change with the Unit of Work, which keeps an in-memory list of everything that needs to happen. "Coordinates the writing out of changes" - when the business transaction is complete and everything looks correct, the Unit of Work flushes all the accumulated changes to the database in a single coordinated operation. This is the commit() call. "Resolution of concurrency problems" - by batching all writes into a single transaction, the Unit of Work also helps manage concurrent access. Either all changes go through together or none do, which prevents other processes from seeing partial states. The pattern is closely related to two others from the same catalog: the Identity Map (which ensures that each database record is loaded only once per transaction, preventing duplicate in-memory objects) and the Repository (which abstracts the data access layer and uses the Unit of Work to coordinate persistence). Together these three patterns form the backbone of most modern ORM implementations.
- ## The Problem it Solves: A Concrete Example To understand why Unit of Work matters, let's look at what happens without it. Imagine you are building an e-commerce platform. When a customer places an order, your system needs to:
Create a new order record in the orders table
Deduct the purchased quantity from the inventory table
Record the total price for billing Without Unit of Work, a naive implementation might look like this:

def process_order_naive(conn, item_id, quantity):
 cursor = conn.cursor()
# Step 1: insert the order
 cursor.execute("INSERT INTO orders (item_id, quantity) VALUES (?, ?)", (item_id, quantity))
 conn.commit() # committed immediately
# Step 2: update inventory
 cursor.execute("UPDATE inventory SET stock = stock - ? WHERE id = ?", (quantity, item_id))
 conn.commit() # committed immediately
# Step 3: record billing
 cursor.execute("INSERT INTO billing (item_id, quantity) VALUES (?, ?)", (item_id, quantity))
 conn.commit() # committed immediately

This looks reasonable but has a critical flaw. If step 2 fails - say, due to a network error, a constraint violation, or a bug - the order already exists in the database but the stock was never decremented. You now have:

An order that references a product
Inventory that still shows the original stock level
No billing record Your database is now inconsistent. A customer placed an order and your system thinks the stock is still available. The next customer might be able to order the same item even though it is already sold. Debugging this in production at 2am is not a good time. The root cause is simple: each operation was committed independently. There was no coordination. With Unit of Work, all three operations are registered in memory and only committed together as a single atomic transaction. If any step fails, all three are rolled back automatically. The database either reflects the complete order or it reflects nothing - there is no in-between state.
- ## Core Concepts in Detail Before diving into code, let's understand the four core operations that define a Unit of Work: ### register_new(obj) Called when a new domain object is created that does not yet exist in the database. The Unit of Work adds it to an internal new_objects list. On commit, these become INSERT statements. ### register_dirty(obj) Called when an existing domain object has been modified. "Dirty" is a classic database term meaning "modified but not yet persisted". The Unit of Work adds it to a dirty_objects list. On commit, these become UPDATE statements. ### register_deleted(obj) Called when a domain object needs to be removed from the database. The Unit of Work adds it to a deleted_objects list. On commit, these become DELETE statements. ### commit() The moment of truth. When called, the Unit of Work iterates over all three lists and generates the corresponding SQL statements, executes them inside a single database transaction, and calls connection.commit(). If anything fails at any point, it calls connection.rollback() and raises the exception so the caller knows something went wrong. The beauty of this design is that your business logic never writes to the database directly. It only works with in-memory Python objects and registers intent. The Unit of Work handles all the SQL coordination.
- ## Implementation: Step by Step Now let's build the full implementation. We are using Python's built-in sqlite3 library and dataclasses - no external dependencies required. ### Step 1: Define the Domain Models

import sqlite3
from dataclasses import dataclass
from typing import List, Any, Optional
@dataclass
class InventoryItem:
 id: int
 name: str
 price: float
 stock: int
@dataclass
class Order:
 id: Optional[int]
 item_id: int
 quantity: int
 total_price: float

We use Python dataclasses for simplicity. In a real project these would be your domain entities - classes that represent core business concepts in your system, completely ignorant of how they are persisted.
Notice that Order.id is Optional[int] - when we create a new order it does not have a database ID yet. The database will assign one on insert.

Step 2: Build the Unit of Work Class

class UnitOfWork:
 """
 Implements the Unit of Work design pattern.
 Maintains a list of objects affected by a business transaction
 and coordinates the writing out of changes and the resolution
 of concurrency problems.
Reference: Fowler, M. (2002). Patterns of Enterprise Application
 Architecture. Addison-Wesley.
 https://martinfowler.com/eaaCatalog/unitOfWork.html
 """
def __init__(self, connection: sqlite3.Connection):
 self.connection = connection
 self.new_objects: List[Any] = []
 self.dirty_objects: List[Any] = []
 self.deleted_objects: List[Any] = []
def __enter__(self):
 return self
def __exit__(self, exc_type, exc_val, exc_tb):
 if exc_type is not None:
 self.rollback()
 else:
 self.commit()
 self._clear()
def _clear(self):
 self.new_objects.clear()
 self.dirty_objects.clear()
 self.deleted_objects.clear()
def register_new(self, obj: Any):
 if obj not in self.new_objects:
 self.new_objects.append(obj)
def register_dirty(self, obj: Any):
 if obj not in self.dirty_objects:
 self.dirty_objects.append(obj)
def register_deleted(self, obj: Any):
 if obj not in self.deleted_objects:
 self.deleted_objects.append(obj)
def commit(self):
 cursor = self.connection.cursor()
 try:
 # Process new objects - INSERT
 for obj in self.new_objects:
 if isinstance(obj, Order):
 cursor.execute(
 "INSERT INTO orders (item_id, quantity, total_price) "
 "VALUES (?, ?, ?)",
 (obj.item_id, obj.quantity, obj.total_price)
 )
 elif isinstance(obj, InventoryItem):
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) "
 "VALUES (?, ?, ?, ?)",
 (obj.id, obj.name, obj.price, obj.stock)
 )
# Process dirty objects - UPDATE
 for obj in self.dirty_objects:
 if isinstance(obj, InventoryItem):
 cursor.execute(
 "UPDATE inventory SET name=?, price=?, stock=? WHERE id=?",
 (obj.name, obj.price, obj.stock, obj.id)
 )
 elif isinstance(obj, Order):
 cursor.execute(
 "UPDATE orders SET item_id=?, quantity=?, total_price=? WHERE id=?",
 (obj.item_id, obj.quantity, obj.total_price, obj.id)
 )
# Process deleted objects - DELETE
 for obj in self.deleted_objects:
 if isinstance(obj, InventoryItem):
 cursor.execute(
 "DELETE FROM inventory WHERE id=?", (obj.id,)
 )
 elif isinstance(obj, Order):
 cursor.execute(
 "DELETE FROM orders WHERE id=?", (obj.id,)
 )
self.connection.commit()
 print("[UoW] Transaction committed successfully.")
except Exception as e:
 self.rollback()
 raise e
def rollback(self):
 self.connection.rollback()
 print("[UoW] Transaction rolled back.")

A few things worth highlighting here:
Context manager support - implementing __enter__ and __exit__ allows the with statement. If an exception is raised anywhere inside the with block, Python automatically calls __exit__ with the exception information, which triggers rollback(). If everything succeeds, it calls commit(). This means the caller does not need any try/except logic - the Unit of Work handles it.
Duplicate guard - both register_new and register_dirty check if obj not in list before appending. This prevents the same object from being processed twice if your business logic accidentally registers it more than once.
Clear on exit - _clear() is called in __exit__ regardless of success or failure. This ensures the Unit of Work is clean and ready for reuse after the transaction completes.

Step 3: Set Up the Database

def setup_db() -> sqlite3.Connection:
 """
 Initializes an in-memory SQLite database with schema and seed data.
 In a real application this would point to a persistent database.
 """
 conn = sqlite3.connect(":memory:")
 cursor = conn.cursor()
cursor.execute("""
 CREATE TABLE inventory (
 id INTEGER PRIMARY KEY,
 name TEXT NOT NULL,
 price REAL NOT NULL,
 stock INTEGER NOT NULL
 )
 """)
cursor.execute("""
 CREATE TABLE orders (
 id INTEGER PRIMARY KEY AUTOINCREMENT,
 item_id INTEGER NOT NULL,
 quantity INTEGER NOT NULL,
 total_price REAL NOT NULL,
 FOREIGN KEY (item_id) REFERENCES inventory(id)
 )
 """)
# Seed with two products
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) VALUES (1, 'Laptop', 1200.0, 10)"
 )
 cursor.execute(
 "INSERT INTO inventory (id, name, price, stock) VALUES (2, 'Smartphone', 800.0, 5)"
 )
conn.commit()
 return conn

We use :memory: for simplicity - this creates a temporary SQLite database that lives only in RAM. For a real application you would replace this with a connection string to your actual database.

Step 4: Business Logic Using Unit of Work

def process_order(conn: sqlite3.Connection, item_id: int, quantity: int):
 """
 Processes a purchase order using the Unit of Work pattern.
 All database changes are coordinated through the UoW - none
 of them touch the database directly until commit() is called.
 """
 cursor = conn.cursor()
 cursor.execute(
 "SELECT id, name, price, stock FROM inventory WHERE id=?", (item_id,)
 )
 row = cursor.fetchone()
if not row:
 raise ValueError(f"Product with ID {item_id} not found.")
# Load the domain object into memory
 item = InventoryItem(id=row[0], name=row[1], price=row[2], stock=row[3])
print(
 f"\n[Process] Attempting to purchase {quantity}x "
 f"'{item.name}' (Available stock: {item.stock})"
 )
# The entire business transaction runs inside the UoW context
 with UnitOfWork(conn) as uow:
# Validate business rules BEFORE registering any changes
 if item.stock < quantity:
 raise ValueError(
 f"Insufficient stock for '{item.name}'. "
 f"Requested: {quantity}, Available: {item.stock}"
 )
# Modify the in-memory object and register it as dirty
 item.stock -= quantity
 uow.register_dirty(item)
# Create the new order object and register it as new
 total_price = item.price * quantity
 order = Order(
 id=None,
 item_id=item_id,
 quantity=quantity,
 total_price=total_price
 )
 uow.register_new(order)
# At this point nothing has been written to the database yet.
 # When the 'with' block exits successfully, __exit__ calls commit()
 # and both changes (stock update + order insert) go through together.

This is the key design point. Inside the with block, we:

Load the current state from the database
Validate the business rule (enough stock available)
Modify the in-memory objects
Register the changes with the Unit of Work Nothing touches the database until the with block exits cleanly. If the stock check fails and raises a ValueError, the __exit__ method sees the exception and calls rollback() instead of commit(). The database remains untouched. ### Step 5: Utility and Main Execution

def print_db_state(conn: sqlite3.Connection):
 cursor = conn.cursor()
 print("\n - - DATABASE CURRENT STATE - -")
 print("Inventory Table:")
 cursor.execute("SELECT * FROM inventory")
 for row in cursor.fetchall():
 print(f" ID: {row[0]} | Name: {row[1]:<12} | Price: ${row[2]:.2f} | Stock: {row[3]}")
print("Orders Table:")
 cursor.execute("SELECT * FROM orders")
 orders = cursor.fetchall()
 if not orders:
 print(" (empty)")
 for row in orders:
 print(f" ID: {row[0]} | Item ID: {row[1]} | Quantity: {row[2]} | Total: ${row[3]:.2f}")
 print(" - - - - - - - - - - - - - - - ")
if __name__ == "__main__":
 conn = setup_db()
print("=== Initial Database State ===")
 print_db_state(conn)
print("\n=== Test Case 1: Successful Order (3 Laptops) ===")
 try:
 process_order(conn, item_id=1, quantity=3)
 print("Result: Order processed successfully.")
 except Exception as e:
 print(f"Result: Failed - {e}")
 print_db_state(conn)
print("\n=== Test Case 2: Failed Order (6 Smartphones, only 5 in stock) ===")
 try:
 process_order(conn, item_id=2, quantity=6)
 print("Result: Order processed successfully.")
 except Exception as e:
 print(f"Result: Failed as expected - {e}")
 print_db_state(conn)

- ## Expected Output Running python unit_of_work.py produces:

# 🌟 Design Principles of Software 🌟

Diego Fernando Castillo Mamani — Wed, 29 Apr 2026 23:02:23 +0000

Introduction 🚀

Software design principles are fundamental guidelines that help developers create robust, maintainable, and scalable software systems. Applying these principles leads to code that is easier to understand, test, and evolve over time. This article explores several key design principles and illustrates them with a practical example using Go.

Key Design Principles 💡

Here's a breakdown of some essential design principles:

SOLID: A cornerstone of object-oriented design, SOLID represents five principles:
- Single Responsibility Principle: A class should have only one reason to change.
- Open/Closed Principle: Software entities (classes, modules, functions, etc.) should be open for extension, but closed for modification.
- Liskov Substitution Principle: Subtypes should be substitutable for their base types without altering the correctness of the program.
- Interface Segregation Principle: Clients should not be forced to depend on methods they do not use.
- Dependency Inversion Principle: High-level modules should not depend on low-level modules. Both should depend on abstractions.
DRY (Don't Repeat Yourself): Avoid code duplication. Extract common logic into reusable components.
KISS (Keep It Simple, Stupid): Favor simplicity over complexity.
YAGNI (You Ain't Gonna Need It): Don't implement features until they are actually needed.
Law of Demeter (LoD): An object should only talk to its immediate neighbors.

Real-World Example: Order Processing System (Go) 💻

Let's consider a simplified order processing system. We'll demonstrate the Single Responsibility Principle and Dependency Inversion Principle.

Bad Design (Violating SRP & DIP):

package main

import "fmt"

type Order struct {
    OrderID    int
    CustomerID int
    Items      []string
    TotalAmount float64
}

type OrderProcessor struct{}

func (o *OrderProcessor) ProcessOrder(order *Order) error {
    // Validate order
    if order.TotalAmount <= 0 {
        return fmt.Errorf("invalid order amount")
    }

    // Calculate shipping cost
    shippingCost := 5.0

    // Apply discount
    discount := 0.0
    if order.CustomerID == 123 {
        discount = 0.1 // 10% discount for customer 123
    }

    // Save order to database
    // ... (database saving logic here) ...

    // Send confirmation email
    // ... (email sending logic here) ...

    fmt.Println("Order processed successfully!")
    return nil
}

func main() {
    order := &Order{
        OrderID:       1,
        CustomerID:    123,
        Items:         []string{"Product A", "Product B"},
        TotalAmount:   100.0,
    }

    processor := &OrderProcessor{}
    err := processor.ProcessOrder(order)
    if err != nil {
        fmt.Println("Error:", err)
    }
}

In this bad design, the OrderProcessor class handles order validation, shipping cost calculation, discount application, database saving, and email sending. This violates the Single Responsibility Principle because it has too many responsibilities. It also violates the Dependency Inversion Principle because it directly depends on concrete implementations (database, email service).

Good Design (Applying SRP & DIP):

package main

import (
    "fmt"
)

type Order struct {
    OrderID    int
    CustomerID int
    Items      []string
    TotalAmount float64
}

// OrderValidator interface
type OrderValidator interface {
    ValidateOrder(order *Order) error
}

// OrderSaver interface
type OrderSaver interface {
    SaveOrder(order *Order) error
}

// EmailService interface
type EmailService interface {
    SendConfirmationEmail(order *Order) error
}

type OrderValidatorImpl struct{}

func (o *OrderValidatorImpl) ValidateOrder(order *Order) error {
    if order.TotalAmount <= 0 {
        return fmt.Errorf("invalid order amount")
    }
    return nil
}

type OrderSaverImpl struct{}

func (o *OrderSaverImpl) SaveOrder(order *Order) error {
    // Simulate saving to database
    fmt.Println("Saving order to database...")
    return nil
}

type EmailServiceImpl struct{}

func (e *EmailServiceImpl) SendConfirmationEmail(order *Order) error {
    // Simulate sending email
    fmt.Println("Sending confirmation email...")
    return nil
}

type OrderProcessor struct {
    validator OrderValidator
    saver     OrderSaver
    emailService EmailService
}

func NewOrderProcessor(validator OrderValidator, saver OrderSaver, emailService EmailService) *OrderProcessor {
    return &OrderProcessor{validator: validator, saver: saver, emailService: emailService}
}

func (o *OrderProcessor) ProcessOrder(order *Order) error {
    if err := o.validator.ValidateOrder(order); err != nil {
        return err
    }

    // Calculate shipping cost, apply discount, etc. (simplified)

    if err := o.saver.SaveOrder(order); err != nil {
        return err
    }

    if err := o.emailService.SendConfirmationEmail(order); err != nil {
        return err
    }

    fmt.Println("Order processed successfully!")
    return nil
}

func main() {
    order := &Order{
        OrderID:       1,
        CustomerID:    123,
        Items:         []string{"Product A", "Product B"},
        TotalAmount:   100.0,
    }

    validator := &OrderValidatorImpl{}
    saver := &OrderSaverImpl{}
    emailService := &EmailServiceImpl{}

    processor := NewOrderProcessor(validator, saver, emailService)
    err := processor.ProcessOrder(order)
    if err != nil {
        fmt.Println("Error:", err)
    }
}

In this improved design:

We've introduced interfaces (OrderValidator, OrderSaver, EmailService) to abstract the dependencies.
The OrderProcessor now depends on these interfaces instead of concrete implementations.
Each responsibility (validation, saving, email sending) is handled by a separate class, adhering to the Single Responsibility Principle.
This design is more flexible and testable because we can easily swap out different implementations of the interfaces.

Conclusion 🎉

Applying design principles is crucial for building maintainable and scalable software. By adhering to principles like SOLID, DRY, KISS, and YAGNI, developers can create code that is easier to understand, test, and evolve. The example demonstrates how the Single Responsibility Principle and Dependency Inversion Principle can significantly improve the structure and flexibility of a software system. Remember to always prioritize clarity and simplicity in your designs.

Securing Infrastructure as Code with Checkov: A Practical SAST Approach

Diego Fernando Castillo Mamani — Fri, 05 Dec 2025 20:31:57 +0000

Introduction

Infrastructure as Code (IaC) has revolutionized how we manage and deploy cloud resources. However, with great power comes great responsibility. Security misconfigurations in IaC templates can lead to serious vulnerabilities in production environments. This is where Static Application Security Testing (SAST) tools like Checkov come into play.

In this article, we'll explore how to implement Checkov, a powerful open-source SAST tool, to scan and secure your Infrastructure as Code across multiple platforms including Terraform, CloudFormation, Kubernetes, and more.

What is Checkov?

Checkov is a static code analysis tool developed by Bridgecrew (now part of Palo Alto Networks) that scans infrastructure code for security and compliance misconfigurations. It supports:

Terraform
CloudFormation
Kubernetes
Helm
ARM Templates
Serverless Framework
Docker
And many more...

Why Use SAST for Infrastructure as Code?

Early Detection: Identify security issues before they reach production
Compliance: Ensure adherence to industry standards (CIS, PCI-DSS, HIPAA)
Cost Reduction: Fix issues early in the development cycle
Automation: Integrate seamlessly into CI/CD pipelines
Policy as Code: Define and enforce organizational security policies

Getting Started with Checkov

Installation

Checkov can be installed using pip:

pip install checkov

Or using Docker:

docker pull bridgecrew/checkov

Basic Usage

Scan a Terraform directory:

checkov -d /path/to/terraform

Scan a specific file:

checkov -f main.tf

Practical Example: Securing Terraform Code

Let's examine a vulnerable Terraform configuration and how Checkov identifies the issues.

Vulnerable Configuration

# main.tf - INSECURE EXAMPLE
resource "aws_s3_bucket" "data_bucket" {
  bucket = "my-data-bucket"

  # Missing encryption
  # Missing versioning
  # Missing logging
}

resource "aws_security_group" "web_sg" {
  name        = "web-security-group"
  description = "Security group for web servers"

  ingress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]  # Too permissive!
  }
}

resource "aws_db_instance" "database" {
  allocated_storage    = 20
  engine              = "mysql"
  instance_class      = "db.t3.micro"
  username            = "admin"
  password            = "hardcodedpassword123"  # Hardcoded password!
  skip_final_snapshot = true

  # Missing encryption
  # Publicly accessible by default
}

Running Checkov

checkov -f main.tf --output json --output-file results.json

Checkov Output

Checkov will identify multiple security issues:

Check: CKV_AWS_18: "Ensure the S3 bucket has access logging enabled"
Check: CKV_AWS_19: "Ensure the S3 bucket has server-side encryption enabled"
Check: CKV_AWS_21: "Ensure the S3 bucket has versioning enabled"
Check: CKV_AWS_23: "Ensure Security Group does not allow ingress from 0.0.0.0/0 to port 22"
Check: CKV_AWS_16: "Ensure RDS database is encrypted at rest"
Check: CKV_AWS_41: "Ensure RDS instance has IAM authentication enabled"

Secure Configuration

Now let's fix these issues:

# main.tf - SECURE VERSION
resource "aws_s3_bucket" "data_bucket" {
  bucket = "my-data-bucket"
}

resource "aws_s3_bucket_versioning" "data_bucket_versioning" {
  bucket = aws_s3_bucket.data_bucket.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "data_bucket_encryption" {
  bucket = aws_s3_bucket.data_bucket.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = "AES256"
    }
  }
}

resource "aws_s3_bucket_logging" "data_bucket_logging" {
  bucket = aws_s3_bucket.data_bucket.id

  target_bucket = aws_s3_bucket.log_bucket.id
  target_prefix = "log/"
}

resource "aws_security_group" "web_sg" {
  name        = "web-security-group"
  description = "Security group for web servers"

  ingress {
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/8"]  # Restricted to internal network
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

resource "aws_db_instance" "database" {
  allocated_storage    = 20
  engine              = "mysql"
  instance_class      = "db.t3.micro"
  username            = "admin"
  password            = var.db_password  # Using variable instead
  skip_final_snapshot = false

  storage_encrypted          = true
  iam_database_authentication_enabled = true
  publicly_accessible        = false

  backup_retention_period = 7
  enabled_cloudwatch_logs_exports = ["error", "general", "slowquery"]
}

Integrating Checkov into CI/CD

GitHub Actions Example

name: IaC Security Scan

on: [push, pull_request]

jobs:
  checkov:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout code
        uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'

      - name: Install Checkov
        run: pip install checkov

      - name: Run Checkov
        run: |
          checkov -d . --output cli --output json --output-file checkov-report.json

      - name: Upload results
        uses: actions/upload-artifact@v3
        if: always()
        with:
          name: checkov-report
          path: checkov-report.json

GitLab CI Example

checkov:
  stage: test
  image: bridgecrew/checkov:latest
  script:
    - checkov -d . --output cli --output junitxml --output-file checkov-report.xml
  artifacts:
    reports:
      junit: checkov-report.xml

Custom Policies

Checkov allows you to create custom policies using Python:

from checkov.common.models.enums import CheckResult, CheckCategories
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck

class S3BucketEncryption(BaseResourceCheck):
    def __init__(self):
        name = "Ensure S3 bucket is encrypted with KMS"
        id = "CKV_AWS_CUSTOM_1"
        supported_resources = ['aws_s3_bucket']
        categories = [CheckCategories.ENCRYPTION]
        super().__init__(name=name, id=id, categories=categories, 
                         supported_resources=supported_resources)

    def scan_resource_conf(self, conf):
        if 'server_side_encryption_configuration' in conf:
            encryption_config = conf['server_side_encryption_configuration'][0]
            if 'rule' in encryption_config:
                rule = encryption_config['rule'][0]
                if 'apply_server_side_encryption_by_default' in rule:
                    encryption = rule['apply_server_side_encryption_by_default'][0]
                    if encryption.get('sse_algorithm') == ['aws:kms']:
                        return CheckResult.PASSED
        return CheckResult.FAILED

check = S3BucketEncryption()

Best Practices

Run Checkov Early: Integrate it into your development workflow
Use Suppressions Wisely: Document why certain checks are suppressed
Keep Checkov Updated: New checks are added regularly
Set Baseline: Use --baseline to track improvements over time
Combine with Other Tools: Use alongside dynamic testing and manual reviews
Educate Your Team: Ensure developers understand the security implications

Suppressing False Positives

Sometimes you need to suppress legitimate findings:

resource "aws_security_group" "special_case" {
  name = "special-sg"

  #checkov:skip=CKV_AWS_23:This SG is for internal load balancer
  ingress {
    from_port   = 80
    to_port     = 80
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

Measuring Success

Track these metrics to measure the effectiveness of Checkov:

Vulnerability Detection Rate: Number of issues found per scan
Fix Time: Time from detection to remediation
False Positive Rate: Percentage of findings that are not actual issues
Coverage: Percentage of infrastructure code being scanned
Compliance Score: Pass rate against security benchmarks

Conclusion

Implementing SAST tools like Checkov is essential for maintaining secure infrastructure as code. By catching security issues early in the development cycle, you can:

Reduce security risks in production
Ensure compliance with industry standards
Foster a security-first culture
Save time and money on remediation

Start small by integrating Checkov into one project, measure the results, and gradually expand coverage across your organization. Security is a journey, not a destination, and tools like Checkov are invaluable companions along the way.

References

🚀Building Your First MCP Server: A Complete Guide

Diego Fernando Castillo Mamani — Tue, 02 Dec 2025 04:42:44 +0000

Introduction

The Model Context Protocol (MCP) is revolutionizing how AI applications interact with external tools and data sources. Whether you're building Claude integrations, VSCode extensions, or custom AI workflows, understanding MCP servers is essential for modern AI development.

In this comprehensive guide, we'll build a fully functional MCP server from scratch, deploy it to the cloud, and connect it to popular MCP clients. By the end, you'll have a working weather information server that any MCP client can use.

What is MCP?

MCP (Model Context Protocol) is an open protocol that standardizes how AI applications connect to external data sources and tools. Think of it as a universal adapter that allows AI models like Claude to safely access your databases, APIs, file systems, and business tools.

Key Benefits

Standardized Integration: One protocol works across all MCP-compatible clients
Security First: Built-in authentication and permission controls
Flexible Architecture: Support for local and remote servers
Tool Discovery: Clients automatically discover available capabilities

Architecture Overview

An MCP server consists of three main components:

Resources: Data sources the server can provide (files, database records, API data)
Tools: Actions the server can perform (create, update, delete operations)
Prompts: Pre-defined templates for common tasks

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│             │         │             │         │             │
│ MCP Client  │◄───────►│ MCP Server  │◄───────►│  External   │
│  (Claude)   │   MCP   │   (Your     │         │   APIs      │
│             │ Protocol│    Code)    │         │             │
└─────────────┘         └─────────────┘         └─────────────┘

Building Our Weather MCP Server

Let's build a practical MCP server that provides weather information using a public API.

Project Setup

First, create a new project and install dependencies:

mkdir weather-mcp-server
cd weather-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk axios dotenv
npm install -D @types/node typescript

Project Structure

weather-mcp-server/
├── src/
│   ├── index.ts
│   ├── server.ts
│   └── weatherService.ts
├── package.json
├── tsconfig.json
└── .env

Configuration Files

tsconfig.json

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules"]
}

package.json (add scripts)

{
  "scripts": {
    "build": "tsc",
    "start": "node dist/index.js",
    "dev": "tsc && node dist/index.js"
  },
  "type": "module"
}

Weather Service Implementation

src/weatherService.ts

import axios from 'axios';

export interface WeatherData {
  location: string;
  temperature: number;
  condition: string;
  humidity: number;
  windSpeed: number;
  description: string;
}

export class WeatherService {
  private apiKey: string;
  private baseUrl = 'https://api.openweathermap.org/data/2.5';

  constructor(apiKey: string) {
    this.apiKey = apiKey;
  }

  async getCurrentWeather(city: string): Promise<WeatherData> {
    try {
      const response = await axios.get(`${this.baseUrl}/weather`, {
        params: {
          q: city,
          appid: this.apiKey,
          units: 'metric'
        }
      });

      const data = response.data;

      return {
        location: `${data.name}, ${data.sys.country}`,
        temperature: Math.round(data.main.temp),
        condition: data.weather[0].main,
        humidity: data.main.humidity,
        windSpeed: data.wind.speed,
        description: data.weather[0].description
      };
    } catch (error) {
      throw new Error(`Failed to fetch weather data: ${error}`);
    }
  }

  async getForecast(city: string, days: number = 5): Promise<any> {
    try {
      const response = await axios.get(`${this.baseUrl}/forecast`, {
        params: {
          q: city,
          appid: this.apiKey,
          units: 'metric',
          cnt: days * 8 // API returns 3-hour intervals
        }
      });

      return response.data;
    } catch (error) {
      throw new Error(`Failed to fetch forecast data: ${error}`);
    }
  }
}

MCP Server Implementation

src/server.ts

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  ListResourcesRequestSchema,
  ReadResourceRequestSchema
} from '@modelcontextprotocol/sdk/types.js';
import { WeatherService } from './weatherService.js';

export class WeatherMCPServer {
  private server: Server;
  private weatherService: WeatherService;

  constructor(apiKey: string) {
    this.weatherService = new WeatherService(apiKey);
    this.server = new Server(
      {
        name: 'weather-mcp-server',
        version: '1.0.0',
      },
      {
        capabilities: {
          tools: {},
          resources: {}
        },
      }
    );

    this.setupHandlers();
  }

  private setupHandlers() {
    // List available tools
    this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
      tools: [
        {
          name: 'get_current_weather',
          description: 'Get current weather information for a city',
          inputSchema: {
            type: 'object',
            properties: {
              city: {
                type: 'string',
                description: 'City name (e.g., "London", "New York")'
              }
            },
            required: ['city']
          }
        },
        {
          name: 'get_forecast',
          description: 'Get weather forecast for upcoming days',
          inputSchema: {
            type: 'object',
            properties: {
              city: {
                type: 'string',
                description: 'City name'
              },
              days: {
                type: 'number',
                description: 'Number of days (1-5)',
                default: 5
              }
            },
            required: ['city']
          }
        }
      ]
    }));

    // Handle tool calls
    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      const { name, arguments: args } = request.params;

      try {
        if (name === 'get_current_weather') {
          const weather = await this.weatherService.getCurrentWeather(
            args.city as string
          );

          return {
            content: [
              {
                type: 'text',
                text: JSON.stringify(weather, null, 2)
              }
            ]
          };
        }

        if (name === 'get_forecast') {
          const forecast = await this.weatherService.getForecast(
            args.city as string,
            (args.days as number) || 5
          );

          return {
            content: [
              {
                type: 'text',
                text: JSON.stringify(forecast, null, 2)
              }
            ]
          };
        }

        throw new Error(`Unknown tool: ${name}`);
      } catch (error) {
        return {
          content: [
            {
              type: 'text',
              text: `Error: ${error}`
            }
          ],
          isError: true
        };
      }
    });

    // List available resources
    this.server.setRequestHandler(ListResourcesRequestSchema, async () => ({
      resources: [
        {
          uri: 'weather://cities',
          name: 'Popular Cities',
          description: 'List of popular cities for weather queries',
          mimeType: 'application/json'
        }
      ]
    }));

    // Read resources
    this.server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
      const { uri } = request.params;

      if (uri === 'weather://cities') {
        const cities = [
          'London', 'New York', 'Tokyo', 'Paris', 'Sydney',
          'Berlin', 'Toronto', 'Singapore', 'Dubai', 'Mumbai'
        ];

        return {
          contents: [
            {
              uri,
              mimeType: 'application/json',
              text: JSON.stringify(cities, null, 2)
            }
          ]
        };
      }

      throw new Error(`Unknown resource: ${uri}`);
    });
  }

  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
    console.error('Weather MCP Server running on stdio');
  }
}

Entry Point

src/index.ts

import dotenv from 'dotenv';
import { WeatherMCPServer } from './server.js';

dotenv.config();

const apiKey = process.env.OPENWEATHER_API_KEY;

if (!apiKey) {
  console.error('Error: OPENWEATHER_API_KEY environment variable is required');
  process.exit(1);
}

const server = new WeatherMCPServer(apiKey);
server.start().catch((error) => {
  console.error('Server error:', error);
  process.exit(1);
});

Environment Configuration

.env

OPENWEATHER_API_KEY=your_api_key_here

Get a free API key from OpenWeatherMap.

Connecting to Claude Desktop

To use your MCP server with Claude Desktop, add it to your configuration:

MacOS: ~/Library/Application Support/Claude/claude_desktop_config.json

Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "weather": {
      "command": "node",
      "args": ["/absolute/path/to/weather-mcp-server/dist/index.js"],
      "env": {
        "OPENWEATHER_API_KEY": "your_api_key_here"
      }
    }
  }
}

Restart Claude Desktop, and you'll see the weather tools available!

Testing Your Server

Build and run the server:

npm run build
npm start

You can test it with Claude Desktop or create a simple test client:

test-client.ts

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js';

async function testWeatherServer() {
  const transport = new StdioClientTransport({
    command: 'node',
    args: ['dist/index.js']
  });

  const client = new Client({
    name: 'test-client',
    version: '1.0.0'
  }, {
    capabilities: {}
  });

  await client.connect(transport);

  // List available tools
  const tools = await client.listTools();
  console.log('Available tools:', tools);

  // Call weather tool
  const result = await client.callTool({
    name: 'get_current_weather',
    arguments: { city: 'London' }
  });

  console.log('Weather result:', result);
}

testWeatherServer();

Deploying to Production

Option 1: Docker Deployment

Dockerfile

FROM node:18-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY dist ./dist

ENV NODE_ENV=production

CMD ["node", "dist/index.js"]

Option 2: Google Cloud Run

gcloud run deploy weather-mcp-server \
  --source . \
  --platform managed \
  --region us-central1 \
  --set-env-vars OPENWEATHER_API_KEY=your_key

Best Practices

Error Handling: Always wrap external API calls in try-catch blocks
Validation: Validate all input parameters before processing
Rate Limiting: Implement rate limiting for external API calls
Caching: Cache frequently requested data to reduce API costs
Logging: Use structured logging for debugging and monitoring
Security: Never expose API keys in code or version control

Advanced Features

Adding Authentication

private validateRequest(request: any): boolean {
  const token = request.params._meta?.authorization;
  return token === process.env.AUTH_TOKEN;
}

Implementing Caching

import NodeCache from 'node-cache';

private cache = new NodeCache({ stdTTL: 600 }); // 10 minutes

async getCachedWeather(city: string): Promise<WeatherData> {
  const cacheKey = `weather:${city}`;
  const cached = this.cache.get<WeatherData>(cacheKey);

  if (cached) return cached;

  const weather = await this.weatherService.getCurrentWeather(city);
  this.cache.set(cacheKey, weather);

  return weather;
}

Troubleshooting

Common Issues

Server not appearing in Claude Desktop

Verify the path in claude_desktop_config.json is absolute
Check that the build succeeded (npm run build)
Restart Claude Desktop completely

API errors

Confirm your OpenWeather API key is valid
Check rate limits on your API tier
Verify network connectivity

TypeScript errors

Ensure all dependencies are installed
Run npm run build to check for compilation errors

Conclusion

You've now built a complete MCP server that can integrate with any MCP-compatible client. This weather server demonstrates the core concepts:

Tool definition and execution
Resource management
Error handling
Client integration

The same patterns can be extended to build MCP servers for:

Database access
File system operations
API integrations
Business tool automation
Custom workflows

Next Steps

Extend functionality: Add more weather features (air quality, UV index, alerts)
Add tests: Implement unit and integration tests
Monitor performance: Add metrics and logging
Build a remote server: Convert to HTTP-based MCP server for multiple clients
Create documentation: Document your API for other developers

Resources

🚀 CI/CD Testing Tools Comparison: GitHub Actions vs GitLab CI vs Jenkins (With Real Examples!)

Diego Fernando Castillo Mamani — Sun, 30 Nov 2025 06:12:56 +0000

📋 Introducción

Las herramientas de testing automatizado son fundamentales en el desarrollo moderno. Este artículo compara las principales plataformas CI/CD con ejemplos prácticos y código real.

🔧 Herramientas Analizadas

1️⃣ GitHub Actions

¿Qué es? Plataforma CI/CD nativa de GitHub

✨ Características Principales:

Integración nativa con repositorios GitHub
Marketplace con miles de actions predefinidas
Matrix builds para múltiples entornos
2,000 minutos gratis/mes

💻 Ejemplo Real:

# .github/workflows/test.yml
name: Run Tests

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

jobs:
  test:
    runs-on: ubuntu-latest

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

    steps:
    - uses: actions/checkout@v3

    - name: Setup 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 tests
      run: npm test -- --coverage

    - name: Upload coverage
      uses: codecov/codecov-action@v3

🎯 Mejor Para: Proyectos en GitHub que necesitan integración simple y rápida

2️⃣ GitLab CI/CD

¿Qué es? Plataforma DevOps completa con CI/CD integrado

✨ Características Principales:

Plataforma todo-en-uno
Auto DevOps para configuración automática
Container registry incluido
Visualización potente de pipelines

💻 Ejemplo Real:

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

variables:
  POSTGRES_DB: test_db

build:
  stage: build
  image: node:18
  script:
    - npm install
  artifacts:
    paths:
      - node_modules/
    expire_in: 1 hour

unit_tests:
  stage: test
  image: node:18
  dependencies:
    - build
  script:
    - npm run test:unit
  coverage: '/Lines\s*:\s*(\d+\.\d+)%/'
  artifacts:
    reports:
      junit: junit.xml
      coverage_report:
        coverage_format: cobertura
        path: coverage/cobertura-coverage.xml

integration_tests:
  stage: test
  image: node:18
  services:
    - postgres:14
  script:
    - npm run test:integration
  only:
    - merge_requests
    - main

🎯 Mejor Para: Equipos que usan GitLab y necesitan una solución DevOps completa

3️⃣ Jenkins

¿Qué es? Servidor de automatización open-source con máxima customización

✨ Características Principales:

1800+ plugins disponibles
Auto-hospedado con control total
Builds distribuidos
Comunidad masiva

💻 Ejemplo Real:

// Jenkinsfile
pipeline {
    agent any

    tools {
        nodejs 'NodeJS-18'
    }

    environment {
        CI = 'true'
    }

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

        stage('Install') {
            steps {
                sh 'npm ci'
            }
        }

        stage('Test') {
            steps {
                sh 'npm test -- --ci'
            }
            post {
                always {
                    junit 'test-results/junit.xml'
                }
            }
        }

        stage('Coverage') {
            steps {
                sh 'npm run test:coverage'
            }
            post {
                always {
                    publishCoverage adapters: [
                        coberturaAdapter('coverage/cobertura-coverage.xml')
                    ]
                }
            }
        }
    }

    post {
        failure {
            emailext (
                subject: "Build Failed: ${env.JOB_NAME}",
                body: "Check: ${env.BUILD_URL}",
                to: "${env.CHANGE_AUTHOR_EMAIL}"
            )
        }
    }
}

🎯 Mejor Para: Empresas que necesitan customización extrema e infraestructura propia

4️⃣ CircleCI

¿Qué es? Plataforma cloud reconocida por su velocidad

✨ Características Principales:

Builds súper rápidos
Caching inteligente
Soporte Docker nativo
Analytics poderosos

💻 Ejemplo Real:

# .circleci/config.yml
version: 2.1

orbs:
  node: circleci/node@5.1.0

executors:
  node-executor:
    docker:
      - image: cimg/node:18.17

jobs:
  test:
    executor: node-executor
    parallelism: 4
    steps:
      - checkout
      - restore_cache:
          keys:
            - v1-deps-{{ checksum "package-lock.json" }}
      - run:
          name: Install
          command: npm ci
      - save_cache:
          paths:
            - node_modules
          key: v1-deps-{{ checksum "package-lock.json" }}
      - run:
          name: Run tests
          command: |
            FILES=$(circleci tests glob "src/**/*.test.js" | \
                    circleci tests split --split-by=timings)
            npm test -- $FILES --coverage
      - store_test_results:
          path: test-results

workflows:
  test-workflow:
    jobs:
      - test

🎯 Mejor Para: Teams que priorizan velocidad y ejecución paralela

5️⃣ Travis CI

¿Qué es? CI/CD clásico, popular en open-source

✨ Características Principales:

Configuración YAML simple
Gratis para proyectos open-source
Multi-lenguaje
Build matrix

💻 Ejemplo Real:

# .travis.yml
language: node_js

node_js:
  - '16'
  - '18'
  - '20'

services:
  - postgresql

env:
  global:
    - DATABASE_URL=postgresql://postgres@localhost/test_db
  matrix:
    - TEST_SUITE=unit
    - TEST_SUITE=integration

cache:
  directories:
    - node_modules

before_script:
  - psql -c 'create database test_db;' -U postgres

script:
  - npm run lint
  - |
    if [ "$TEST_SUITE" = "unit" ]; then
      npm run test:unit -- --coverage
    else
      npm run test:integration
    fi

after_success:
  - npm run coverage:upload

notifications:
  email:
    on_success: change
    on_failure: always

🎯 Mejor Para: Proyectos open-source y equipos que buscan simplicidad

💡 Best Practices

1. 🔺 Pirámide de Testing

Más tests unitarios
Menos tests de integración
Mínimos tests E2E

2. ⚡ Optimización

Cachear dependencias
Ejecutar en paralelo
Fail fast cuando sea posible

3. 📈 Cobertura

Monitorear tendencias
Apuntar a 80%+ en crítico
No sacrificar calidad por %

4. 🔒 Seguridad

Usar secrets para credenciales
Escanear dependencias
Actualizar regularmente

🎬 Repositorio de Ejemplo

He creado un repositorio completo con ejemplos funcionales:

📦 Repo: github.com/your-user/ci-cd-testing-comparison

Incluye:

✅ API Node.js con Express
✅ Tests unitarios (Jest)
✅ Tests de integración (Supertest)
✅ Tests E2E (Playwright)
✅ Configuraciones para todas las plataformas
✅ Docker setup
✅ Documentación completa

🎯 ¿Cuál Elegir?

Elige GitHub Actions si:

Ya usas GitHub
Quieres setup rápido
Necesitas gran ecosistema

Elige GitLab CI/CD si:

Quieres plataforma todo-en-uno
Necesitas self-hosting
Buscas DevOps completo

Elige Jenkins si:

Necesitas control total
Tienes infraestructura propia
Requieres customización extrema

Elige CircleCI si:

Velocidad es prioridad
Necesitas paralelismo avanzado
Quieres analytics detallados

Elige Travis CI si:

Proyecto open-source
Buscas simplicidad máxima
Configuración mínima

🚀 Conclusión

No existe una solución única para todos. La elección depende de tu stack, equipo, presupuesto y necesidades específicas. Todas estas herramientas son excelentes cuando se usan correctamente.

Pro tip: Empieza simple, itera y optimiza según tus necesidades reales.

📚 Recursos Adicionales

🏷️ Tags: #DevOps #CICD #Testing #Automation #GitHubActions #GitLab #Jenkins #CircleCI #TravisCI #QualityAssurance

👤 Autor: [Diego Fernando Castillo Mamani] | 📅 Fecha: Noviembre 2024

Interactive Data Visualization: Streamlit, Dash, and Bokeh

Diego Fernando Castillo Mamani — Mon, 15 Sep 2025 04:09:25 +0000

Introduction

Modern data analysis requires interactive visualizations that allow users to explore data dynamically. Today we'll explore three powerful Python libraries: Streamlit, Dash, and Bokeh - each offering unique advantages for different use cases.

🚀 Streamlit: Simple and Fast

Streamlit is perfect for rapid prototyping with minimal code. It's the go-to choice for data scientists who want to create web apps without web development knowledge.

Key Features

✅ Zero HTML/CSS knowledge required
✅ Real-time updates
✅ Easy deployment
✅ Rich widget ecosystem

Example: Sales Dashboard

import streamlit as st
import pandas as pd
import plotly.express as px
import numpy as np


# Page configuration
st.set_page_config(page_title="Sales Dashboard", layout="wide")
st.title("📊 Sales Dashboard")

# Generate sample data
@st.cache_data
def load_data():
    dates = pd.date_range('2024-01-01', '2024-12-31', freq='D')
    data = {
        'date': dates,
        'sales': np.random.normal(1000, 200, len(dates)),
        'region': np.random.choice(['North', 'South', 'East', 'West'], len(dates)),
        'product': np.random.choice(['Product A', 'Product B', 'Product C'], len(dates))
    }
    df = pd.DataFrame(data)
    df['sales'] = df['sales'].clip(lower=0)  # No negative sales
    return df

df = load_data()

# Sidebar filters
st.sidebar.header("Filters")
region = st.sidebar.selectbox("Select Region", df['region'].unique())
product = st.sidebar.selectbox("Select Product", df['product'].unique())

# Filter data
filtered_df = df[(df['region'] == region) & (df['product'] == product)]

# Metrics row
col1, col2, col3 = st.columns(3)
with col1:
    st.metric("Total Sales", f"${filtered_df['sales'].sum():,.2f}")
with col2:
    st.metric("Average Daily Sales", f"${filtered_df['sales'].mean():.2f}")
with col3:
    st.metric("Number of Days", len(filtered_df))

# Charts
col1, col2 = st.columns(2)

with col1:
    # Time series chart
    fig_line = px.line(
        filtered_df, 
        x='date', 
        y='sales', 
        title=f"Sales Trend - {region} ({product})"
    )
    st.plotly_chart(fig_line, use_container_width=True)

with col2:
    # Distribution chart
    fig_hist = px.histogram(
        filtered_df, 
        x='sales', 
        title="Sales Distribution",
        nbins=30
    )
    st.plotly_chart(fig_hist, use_container_width=True)

# Data table
if st.checkbox("Show Raw Data"):
    st.subheader("Raw Data")
    st.dataframe(filtered_df.head(100), use_container_width=True)

🏢 Dash: Enterprise-Ready

Dash offers more control and customization for production applications. It's built on Flask and React, making it perfect for enterprise-grade dashboards.

Key Features

✅ Production-ready
✅ Highly customizable
✅ Advanced callback system
✅ Enterprise authentication support

Example: Stock Portfolio Tracker

import dash
from dash import dcc, html, Input, Output, callback
import plotly.express as px
import plotly.graph_objects as go
import pandas as pd
import yfinance as yf

# Initialize the Dash app
app = dash.Dash(__name__)

# Portfolio data
STOCKS = ['AAPL', 'GOOGL', 'TSLA', 'MSFT', 'AMZN']

# Layout
app.layout = html.Div([
    html.Div([
        html.H1("💼 Stock Portfolio Tracker", 
                style={'textAlign': 'center', 'color': '#2E86AB', 'marginBottom': 30}),
    ]),

    html.Div([
        html.Div([
            html.Label("Select Stock:", style={'fontWeight': 'bold'}),
            dcc.Dropdown(
                id='stock-dropdown',
                options=[{'label': stock, 'value': stock} for stock in STOCKS],
                value='AAPL',
                style={'marginBottom': 20}
            ),

            html.Label("Time Period:", style={'fontWeight': 'bold'}),
            dcc.Dropdown(
                id='period-dropdown',
                options=[
                    {'label': '1 Month', 'value': '1mo'},
                    {'label': '3 Months', 'value': '3mo'},
                    {'label': '6 Months', 'value': '6mo'},
                    {'label': '1 Year', 'value': '1y'}
                ],
                value='3mo'
            )
        ], style={'width': '30%', 'display': 'inline-block', 'verticalAlign': 'top'}),

        html.Div([
            html.Div(id='stock-info')
        ], style={'width': '70%', 'display': 'inline-block', 'paddingLeft': 20})
    ]),

    html.Div([
        dcc.Graph(id='stock-chart')
    ]),

    html.Div([
        dcc.Graph(id='volume-chart')
    ])
])

@app.callback(
    [Output('stock-chart', 'figure'),
     Output('volume-chart', 'figure'),
     Output('stock-info', 'children')],
    [Input('stock-dropdown', 'value'),
     Input('period-dropdown', 'value')]
)
def update_dashboard(selected_stock, period):
    # Fetch stock data
    try:
        stock_data = yf.download(selected_stock, period=period)

        # Stock price chart
        fig_price = go.Figure()
        fig_price.add_trace(go.Candlestick(
            x=stock_data.index,
            open=stock_data['Open'],
            high=stock_data['High'],
            low=stock_data['Low'],
            close=stock_data['Close'],
            name=selected_stock
        ))
        fig_price.update_layout(
            title=f"{selected_stock} Stock Price",
            yaxis_title="Price ($)",
            xaxis_title="Date"
        )

        # Volume chart
        fig_volume = px.bar(
            x=stock_data.index,
            y=stock_data['Volume'],
            title=f"{selected_stock} Trading Volume"
        )

        # Stock info
        current_price = stock_data['Close'].iloc[-1]
        price_change = current_price - stock_data['Close'].iloc[-2]
        price_change_pct = (price_change / stock_data['Close'].iloc[-2]) * 100

        info = html.Div([
            html.H3(f"{selected_stock} Information"),
            html.P(f"Current Price: ${current_price:.2f}"),
            html.P(f"Change: ${price_change:.2f} ({price_change_pct:.2f}%)",
                   style={'color': 'green' if price_change >= 0 else 'red'}),
            html.P(f"Period High: ${stock_data['High'].max():.2f}"),
            html.P(f"Period Low: ${stock_data['Low'].min():.2f}")
        ])

        return fig_price, fig_volume, info

    except Exception as e:
        # Error handling
        empty_fig = go.Figure()
        error_info = html.P(f"Error loading data: {str(e)}")
        return empty_fig, empty_fig, error_info

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

⚡ Bokeh: Advanced Visualizations

Bokeh handles large datasets and complex interactions efficiently. It's perfect for creating sophisticated, publication-ready visualizations.

Key Features

✅ High performance with large datasets
✅ Advanced interactions
✅ Real-time capabilities
✅ Server and standalone applications

Example: Real-time Data Monitor

from bokeh.plotting import figure, curdoc
from bokeh.layouts import column, row
from bokeh.models import ColumnDataSource, Select, Button
from bokeh.models.widgets import DataTable, TableColumn
import numpy as np
import pandas as pd
from datetime import datetime
from functools import partial

# Global data source
source = ColumnDataSource(data=dict(
    x=[], y=[], timestamp=[], sensor_id=[], color=[]
))

# Create main plot
def create_time_plot():
    p = figure(
        title="Real-time Sensor Data Monitor",
        x_axis_label="Time",
        y_axis_label="Value",
        width=800,
        height=400,
        x_axis_type='datetime'
    )

    p.line('timestamp', 'y', source=source, line_width=2, color='blue', alpha=0.8)
    p.circle('timestamp', 'y', source=source, size=8, color='color', alpha=0.6)

    return p

# Create histogram
def create_histogram():
    p = figure(
        title="Value Distribution",
        x_axis_label="Value",
        y_axis_label="Frequency",
        width=400,
        height=300
    )

    return p

# Controls
sensor_select = Select(
    title="Sensor ID:", 
    value="All", 
    options=["All", "Sensor_1", "Sensor_2", "Sensor_3"]
)

start_button = Button(label="Start Monitoring", button_type="success")
stop_button = Button(label="Stop Monitoring", button_type="danger")

# Data table
columns = [
    TableColumn(field="timestamp", title="Timestamp"),
    TableColumn(field="y", title="Value"),
    TableColumn(field="sensor_id", title="Sensor ID")
]
data_table = DataTable(source=source, columns=columns, width=400, height=200)

# Callback for data generation
def update_data():
    """Generate new data point"""
    sensors = ["Sensor_1", "Sensor_2", "Sensor_3"]
    colors = ["red", "blue", "green"]

    new_data = {
        'timestamp': [datetime.now()],
        'y': [np.random.normal(50, 10)],
        'sensor_id': [np.random.choice(sensors)],
        'color': [np.random.choice(colors)]
    }

    # Update data source
    current_data = source.data.copy()
    for key in new_data:
        if key in current_data:
            current_data[key].extend(new_data[key])
            # Keep only last 100 points
            if len(current_data[key]) > 100:
                current_data[key] = current_data[key][-100:]
        else:
            current_data[key] = new_data[key]

    source.data = current_data

# Button callbacks
def start_monitoring():
    curdoc().add_periodic_callback(update_data, 1000)

def stop_monitoring():
    curdoc().remove_periodic_callback(update_data)

start_button.on_click(start_monitoring)
stop_button.on_click(stop_monitoring)

# Create plots
time_plot = create_time_plot()
histogram = create_histogram()

# Layout
controls = column(sensor_select, row(start_button, stop_button))
plots = row(time_plot, column(histogram, data_table))
layout = column(controls, plots)

# Add to document
curdoc().add_root(layout)
curdoc().title = "Real-time Data Monitor"

📦 Deployment Guide

Streamlit Cloud (Recommended)

Create requirements.txt:

txtstreamlit==1.28.0
pandas==2.0.3
plotly==5.15.0
numpy==1.24.3

Deploy Steps:

# Push to GitHub
git add .
git commit -m "Initial commit"
git push origin main

# Visit share.streamlit.io
# Connect your GitHub repository
# Auto-deploy!

Dash on Heroku

Prepare files:

# Add to your Dash app
server = app.server

if __name__ == '__main__':
    app.run_server(debug=False)

Create Procfile:

web: gunicorn app:server

Deploy:

heroku create your-app-name
git push heroku main

***Bokeh Server*

Start Bokeh server:**

bashbokeh serve --show --allow-websocket-origin=yourdomain.com app.py

For cloud deployment:

bashbokeh serve --port=$PORT --allow-websocket-origin=* app.py

📊 Comparison Matrix

Conclusion

Interactive data visualization has never been more accessible. Whether you choose Streamlit for rapid prototyping, Dash for enterprise applications, or Bokeh for advanced visualizations, you now have the power to create compelling, interactive dashboards that transform how users interact with data.
Start with Streamlit to get your feet wet, graduate to Dash when you need production features, and leverage Bokeh when performance and advanced interactions are critical.
The future of data visualization is interactive - and these tools are your gateway to that future.