LangChain vs Django vs FastAPI: Framework Comparison

Understanding the Three Frameworks

Before diving into code, it's essential to recognize that LangChain, Django, and FastAPI occupy different layers of the modern development stack. LangChain is an AI orchestration framework for building applications powered by large language models. Django is a full-stack web framework with batteries included for building traditional web applications. FastAPI is a high-performance async web framework optimized for building APIs. Each excels in its domain, and in many real-world projects, they can complement one another rather than compete.

LangChain: The AI Application Framework

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

What It Is

LangChain is an open-source framework designed to simplify the creation of applications that leverage large language models (LLMs) like GPT-4, Claude, or open-source models. It provides a standardized interface for chains, agents, tools, memory, and retrieval strategies. Rather than being a web server, LangChain is a library that orchestrates the flow of data between LLMs, external data sources, and custom logic.

Why It Matters

Building LLM-powered applications from scratch requires handling prompt engineering, output parsing, context window management, tool calling, and streaming — all of which become complex quickly. LangChain abstracts these concerns into reusable components. It matters because it drastically reduces the time to build sophisticated AI features like chatbots with memory, retrieval-augmented generation (RAG) systems, and autonomous agents that can interact with APIs and databases.

Core Concepts with Code Examples

Installing LangChain

pip install langchain langchain-openai langchain-community

Basic LLM Chain

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# Initialize the model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# Create a prompt template with placeholders
prompt = ChatPromptTemplate.from_template(
    "You are a helpful coding assistant. "
    "Write a {language} function that {task}. "
    "Include comments and error handling."
)

# Build the chain using the pipe operator
chain = prompt | model | StrOutputParser()

# Execute the chain
result = chain.invoke({
    "language": "Python",
    "task": "validates an email address using regex"
})

print(result)

Sequential Chain with Multiple Steps

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

# Step 1: Generate a product idea
idea_prompt = ChatPromptTemplate.from_template(
    "Generate a creative product idea for the {industry} industry. "
    "Return only the product name and a one-sentence description."
)
idea_chain = idea_prompt | model | StrOutputParser()

# Step 2: Generate marketing copy from the product idea
marketing_prompt = ChatPromptTemplate.from_template(
    "Based on this product idea: {product_idea}\n\n"
    "Write a compelling 3-sentence marketing pitch for social media."
)
marketing_chain = marketing_prompt | model | StrOutputParser()

# Combine into a sequential pipeline
full_chain = idea_chain | marketing_chain

# The output of idea_chain is passed as the 'input' variable to marketing_chain
result = full_chain.invoke({"industry": "sustainable fashion"})
print(result)

RAG with Vector Store and Retrieval

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader

# Load and split documents
loader = TextLoader("company_handbook.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)

# Create vector store with embeddings
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# Create a retriever that fetches top 3 relevant chunks
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# Build the RAG prompt
rag_prompt = ChatPromptTemplate.from_template(
    "You are an HR assistant. Answer the question based on the context provided.\n\n"
    "Context:\n{context}\n\n"
    "Question: {question}\n\n"
    "Answer concisely and cite specific policy sections when applicable."
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# Define a custom chain that retrieves context and formats it
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# Build the full RAG chain
rag_chain = (
    {
        "context": lambda x: format_docs(retriever.invoke(x["question"])),
        "question": lambda x: x["question"]
    }
    | rag_prompt
    | model
    | StrOutputParser()
)

# Query the system
response = rag_chain.invoke({"question": "What is the remote work policy for employees?"})
print(response)

Agent with Tools

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
import json
import math

# Define tools using decorators
@tool
def calculate_mortgage(principal: float, annual_rate: float, years: int) -> str:
    """Calculate monthly mortgage payment given principal, annual interest rate, and term in years."""
    monthly_rate = (annual_rate / 100) / 12
    num_payments = years * 12
    if monthly_rate == 0:
        monthly_payment = principal / num_payments
    else:
        monthly_payment = principal * (monthly_rate * (1 + monthly_rate) ** num_payments) / \
                          ((1 + monthly_rate) ** num_payments - 1)
    return json.dumps({
        "monthly_payment": round(monthly_payment, 2),
        "total_paid": round(monthly_payment * num_payments, 2),
        "total_interest": round(monthly_payment * num_payments - principal, 2)
    })

@tool
def search_web(query: str) -> str:
    """Simulate searching the web for current information. Replace with real search API."""
    # Mock response for demonstration
    knowledge_base = {
        "current mortgage rates": "As of 2025, average 30-year fixed mortgage rates are around 6.5%",
        "home prices san francisco": "Median home price in San Francisco is approximately $1.2 million",
    }
    for key, value in knowledge_base.items():
        if key in query.lower():
            return value
    return f"No specific information found for: {query}"

tools = [calculate_mortgage, search_web]

# Create the agent
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_react_agent(model=model, tools=tools)

# Run the agent with a complex query
response = agent.invoke({
    "messages": [
        ("user", "I want to buy a median-priced home in San Francisco. "
         "Find current mortgage rates and calculate my monthly payment "
         "with a 20% down payment on a 30-year fixed mortgage.")
    ]
})

for message in response["messages"]:
    if hasattr(message, "content"):
        print(f"{message.type}: {message.content}")

LangChain Best Practices

Use LCEL (LangChain Expression Language) — The pipe operator (|) syntax creates chains that are automatically optimized for streaming, async execution, and tracing.
Keep prompts in separate files or a registry — Treat prompts as code; version them alongside your application logic.
Implement observability with LangSmith — Trace every chain execution to debug, monitor costs, and evaluate performance across different model versions.
Start simple, then add complexity — Begin with a basic chain before introducing agents or complex retrieval patterns.
Set temperature and max_tokens explicitly — Never rely on defaults; deterministic tasks need low temperature, creative tasks benefit from higher values.
Cache embeddings and LLM responses — Use CacheBackedEmbeddings and response caching to reduce API costs during development and testing.

Django: The Full-Stack Web Framework

What It Is

Django is a high-level Python web framework that follows the "batteries-included" philosophy. It provides an ORM, an admin interface, form handling, authentication, session management, and a templating engine out of the box. Django follows the Model-View-Template (MVT) architectural pattern and emphasizes convention over configuration, making it ideal for rapid development of full-featured web applications.

Why It Matters

Django matters because it provides a complete toolkit for building secure, scalable web applications without requiring developers to make countless micro-decisions about which libraries to use. Its built-in admin interface saves weeks of development time on internal tools. The ORM abstracts database operations elegantly, and the migration system handles schema evolution. For startups and enterprises alike, Django's mature ecosystem (Django REST Framework, Django Channels, Django Allauth) covers nearly every web development need.

Core Concepts with Code Examples

Setting Up a Django Project

# Install Django
pip install django

# Create a new project
django-admin startproject bookstore
cd bookstore

# Create an app
python manage.py startapp inventory

Defining Models

# inventory/models.py
from django.db import models
from django.core.validators import MinValueValidator, MaxValueValidator
from django.utils import timezone

class Author(models.Model):
    """Represents a book author with biographical details."""
    first_name = models.CharField(max_length=100)
    last_name = models.CharField(max_length=100)
    birth_date = models.DateField(null=True, blank=True)
    email = models.EmailField(unique=True)
    bio = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        ordering = ['last_name', 'first_name']
        indexes = [
            models.Index(fields=['last_name', 'first_name']),
        ]

    def __str__(self):
        return f"{self.first_name} {self.last_name}"

    @property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"


class Publisher(models.Model):
    """Represents a book publishing house."""
    name = models.CharField(max_length=200, unique=True)
    address = models.TextField(blank=True)
    website = models.URLField(blank=True)
    established_year = models.IntegerField(
        null=True,
        blank=True,
        validators=[MinValueValidator(1800), MaxValueValidator(2025)]
    )
    is_active = models.BooleanField(default=True)

    def __str__(self):
        return self.name


class Book(models.Model):
    """The core book model with relationships to Author and Publisher."""
    title = models.CharField(max_length=300)
    subtitle = models.CharField(max_length=300, blank=True)
    authors = models.ManyToManyField(Author, related_name='books')
    publisher = models.ForeignKey(
        Publisher,
        on_delete=models.SET_NULL,
        null=True,
        related_name='books'
    )
    publication_date = models.DateField()
    isbn = models.CharField('ISBN', max_length=13, unique=True)
    page_count = models.IntegerField(validators=[MinValueValidator(1)])
    price = models.DecimalField(max_digits=8, decimal_places=2)
    stock_quantity = models.IntegerField(default=0)
    description = models.TextField(blank=True)
    cover_image = models.ImageField(upload_to='book_covers/', blank=True)
    is_featured = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        ordering = ['-publication_date']
        constraints = [
            models.CheckConstraint(
                check=models.Q(stock_quantity__gte=0),
                name='stock_quantity_non_negative'
            )
        ]

    def __str__(self):
        return self.title

    def is_in_stock(self):
        return self.stock_quantity > 0

Custom Managers and QuerySets

# inventory/managers.py
from django.db import models
from django.utils import timezone

class BookQuerySet(models.QuerySet):
    """Custom queryset with chainable filter methods."""

    def in_stock(self):
        return self.filter(stock_quantity__gt=0)

    def featured(self):
        return self.filter(is_featured=True)

    def published_this_year(self):
        current_year = timezone.now().year
        return self.filter(publication_date__year=current_year)

    def by_price_range(self, min_price, max_price):
        return self.filter(price__gte=min_price, price__lte=max_price)

    def with_authors(self):
        return self.select_related('publisher').prefetch_related('authors')


class BookManager(models.Manager):
    """Custom manager that uses the custom queryset."""

    def get_queryset(self):
        return BookQuerySet(self.model, using=self._db)

    def in_stock(self):
        return self.get_queryset().in_stock()

    def featured(self):
        return self.get_queryset().featured()

    def bestsellers(self, threshold=50):
        """Return books that have sold above a certain threshold."""
        return self.get_queryset().filter(
            stock_quantity__lt=threshold,
            is_featured=True
        )

# Update the Book model to use the custom manager:
# Add to Book model:
# objects = BookManager()  # replaces default manager

Admin Configuration

# inventory/admin.py
from django.contrib import admin
from django.utils.html import format_html
from .models import Author, Publisher, Book

@admin.register(Author)
class AuthorAdmin(admin.ModelAdmin):
    list_display = ['full_name', 'email', 'birth_date', 'book_count']
    list_filter = ['created_at']
    search_fields = ['first_name', 'last_name', 'email']
    readonly_fields = ['created_at', 'updated_at']
    fieldsets = (
        ('Personal Information', {
            'fields': ('first_name', 'last_name', 'birth_date', 'email')
        }),
        ('Biography', {
            'fields': ('bio',),
            'classes': ('wide',)
        }),
        ('Metadata', {
            'fields': ('created_at', 'updated_at'),
            'classes': ('collapse',)
        }),
    )

    @admin.display(description='Number of Books')
    def book_count(self, obj):
        return obj.books.count()


@admin.register(Publisher)
class PublisherAdmin(admin.ModelAdmin):
    list_display = ['name', 'website_link', 'is_active', 'established_year']
    list_filter = ['is_active']
    search_fields = ['name']

    @admin.display(description='Website')
    def website_link(self, obj):
        if obj.website:
            return format_html('Visit', obj.website)
        return '-'


@admin.register(Book)
class BookAdmin(admin.ModelAdmin):
    list_display = ['title', 'publisher', 'price_display', 'stock_status', 'is_featured']
    list_filter = ['is_featured', 'publisher', 'publication_date']
    search_fields = ['title', 'isbn']
    filter_horizontal = ['authors']
    readonly_fields = ['created_at', 'updated_at']
    actions = ['mark_as_featured', 'restock_50']

    @admin.display(description='Price')
    def price_display(self, obj):
        return f"${obj.price:.2f}"

    @admin.display(description='Stock')
    def stock_status(self, obj):
        if obj.stock_quantity > 50:
            return format_html('In Stock ({})', obj.stock_quantity)
        elif obj.stock_quantity > 0:
            return format_html('Low Stock ({})', obj.stock_quantity)
        return format_html('Out of Stock')

    @admin.action(description='Mark selected books as featured')
    def mark_as_featured(self, request, queryset):
        updated = queryset.update(is_featured=True)
        self.message_user(request, f'{updated} books marked as featured.')

    @admin.action(description='Add 50 units to stock')
    def restock_50(self, request, queryset):
        for book in queryset:
            book.stock_quantity += 50
            book.save()
        self.message_user(request, f'Restocked {queryset.count()} books with 50 units each.')

Views with Django REST Framework Integration

# inventory/serializers.py
from rest_framework import serializers
from .models import Author, Publisher, Book

class AuthorSerializer(serializers.ModelSerializer):
    book_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Author
        fields = ['id', 'first_name', 'last_name', 'full_name',
                  'email', 'bio', 'book_count', 'created_at']

class PublisherSerializer(serializers.ModelSerializer):
    book_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Publisher
        fields = '__all__'

class BookSerializer(serializers.ModelSerializer):
    authors = AuthorSerializer(many=True, read_only=True)
    publisher_name = serializers.CharField(source='publisher.name', read_only=True)
    is_in_stock = serializers.BooleanField(read_only=True)

    class Meta:
        model = Book
        fields = ['id', 'title', 'subtitle', 'authors', 'publisher',
                  'publisher_name', 'publication_date', 'isbn', 'page_count',
                  'price', 'stock_quantity', 'is_in_stock', 'is_featured',
                  'description', 'created_at']
        read_only_fields = ['created_at', 'updated_at']

    def validate_isbn(self, value):
        """Custom validation for ISBN field."""
        if len(value) != 13:
            raise serializers.ValidationError("ISBN must be exactly 13 characters.")
        if not value.isdigit():
            raise serializers.ValidationError("ISBN must contain only digits.")
        return value

    def validate(self, data):
        """Cross-field validation."""
        if data.get('stock_quantity', 0) < 0:
            raise serializers.ValidationError(
                {'stock_quantity': 'Stock quantity cannot be negative.'}
            )
        return data

# inventory/views.py
from rest_framework import viewsets, filters, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from django.db.models import Q, Count
from .models import Author, Publisher, Book
from .serializers import AuthorSerializer, PublisherSerializer, BookSerializer

class BookViewSet(viewsets.ModelViewSet):
    """Full CRUD viewset for the Book model with custom actions."""
    queryset = Book.objects.with_authors().all()
    serializer_class = BookSerializer
    permission_classes = [IsAuthenticatedOrReadOnly]
    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
    search_fields = ['title', 'authors__first_name', 'authors__last_name', 'isbn']
    ordering_fields = ['price', 'publication_date', 'stock_quantity']
    ordering = ['-publication_date']

    def get_queryset(self):
        queryset = super().get_queryset()
        # Apply price range filter from query parameters
        min_price = self.request.query_params.get('min_price')
        max_price = self.request.query_params.get('max_price')
        if min_price and max_price:
            queryset = queryset.by_price_range(float(min_price), float(max_price))
        # Filter by author name
        author_name = self.request.query_params.get('author')
        if author_name:
            queryset = queryset.filter(
                Q(authors__first_name__icontains=author_name) |
                Q(authors__last_name__icontains=author_name)
            ).distinct()
        return queryset

    @action(detail=False, methods=['get'])
    def featured(self, request):
        """Return only featured books that are in stock."""
        featured_books = self.get_queryset().featured().in_stock()
        page = self.paginate_queryset(featured_books)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)
        serializer = self.get_serializer(featured_books, many=True)
        return Response(serializer.data)

    @action(detail=False, methods=['get'])
    def stats(self, request):
        """Return aggregate statistics about the book inventory."""
        total_books = self.get_queryset().count()
        total_in_stock = self.get_queryset().in_stock().count()
        avg_price = self.get_queryset().aggregate(
            avg=models.Avg('price')
        )['avg']
        return Response({
            'total_books': total_books,
            'books_in_stock': total_in_stock,
            'average_price': round(float(avg_price or 0), 2),
            'featured_count': self.get_queryset().featured().count(),
        })

    @action(detail=True, methods=['post'])
    def restock(self, request, pk=None):
        """Restock a specific book by a given quantity."""
        book = self.get_object()
        quantity = request.data.get('quantity', 10)
        book.stock_quantity += quantity
        book.save()
        return Response({
            'message': f'Restocked "{book.title}" by {quantity} units.',
            'new_stock': book.stock_quantity
        })

URL Routing

# inventory/urls.py
from rest_framework.routers import DefaultRouter
from django.urls import path, include
from .views import BookViewSet, AuthorViewSet, PublisherViewSet

router = DefaultRouter()
router.register(r'books', BookViewSet, basename='book')
router.register(r'authors', AuthorViewSet, basename='author')
router.register(r'publishers', PublisherViewSet, basename='publisher')

urlpatterns = [
    path('api/', include(router.urls)),
    path('api/auth/', include('rest_framework.urls')),
]

Django Best Practices

Use custom managers for complex queryset logic — Keep your views thin by moving filter chains and aggregation logic into reusable queryset methods.
Structure projects with multiple apps — Decompose functionality into focused apps (e.g., separate inventory, orders, users) rather than monolithic single-app projects.
Leverage Django's admin for rapid prototyping — Customize the admin extensively; it often serves as the internal tool with zero frontend work required.
Use select_related and prefetch_related aggressively — Profile your queries with django-debug-toolbar to catch N+1 query problems early.
Implement custom model validators and clean methods — Never rely solely on form validation; model-level validation ensures data integrity regardless of the entry point.
Keep migrations meaningful and atomic — Avoid auto-generated migration squashing until necessary, and always test migrations with production-sized datasets.

FastAPI: The Modern Async API Framework

What It Is

FastAPI is a modern Python web framework for building APIs with Python 3.8+ type hints. It's built on top of Starlette for the web layer and Pydantic for data validation and serialization. FastAPI is designed for high performance, supporting asynchronous request handling natively, automatic OpenAPI documentation generation, and a developer experience that rivals Node.js frameworks in speed while maintaining Python's readability.

Why It Matters

FastAPI matters because it bridges the gap between development speed and runtime performance. Its use of Python type hints provides editor autocompletion, runtime validation, and automatic API documentation simultaneously. For microservices, real-time applications, and high-throughput API gateways, FastAPI's async-first design handles thousands of concurrent connections efficiently. It has become the go-to choice for machine learning model serving, data pipelines, and modern API backends.

Core Concepts with Code Examples

Installing FastAPI with Production Server

pip install fastapi uvicorn[standard] pydantic[email]

Basic API with Async Endpoints

# main.py
from fastapi import FastAPI, APIRouter
from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime
from enum import Enum

app = FastAPI(
    title="Task Management API",
    description="A production-grade task management system built with FastAPI",
    version="2.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

class TaskPriority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    critical = "critical"

class TaskStatus(str, Enum):
    todo = "todo"
    in_progress = "in_progress"
    review = "review"
    done = "done"
    archived = "archived"

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=3, max_length=200, description="Task title")
    description: Optional[str] = Field(None, max_length=2000)
    priority: TaskPriority = Field(TaskPriority.medium)
    assigned_to: Optional[EmailStr] = None
    tags: List[str] = Field(default_factory=list, max_items=10)
    due_date: Optional[datetime] = None

    class Config:
        json_schema_extra = {
            "example": {
                "title": "Implement user authentication",
                "description": "Add JWT-based authentication to the API gateway",
                "priority": "high",
                "assigned_to": "developer@example.com",
                "tags": ["security", "backend"],
                "due_date": "2025-12-31T23:59:59"
            }
        }

class TaskResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    priority: TaskPriority
    status: TaskStatus
    assigned_to: Optional[EmailStr]
    tags: List[str]
    due_date: Optional[datetime]
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True  # enables ORM mode

# Simulated database
tasks_db = {}
task_counter = 0

router = APIRouter(prefix="/tasks", tags=["Tasks"])

@router.get("/", response_model=List[TaskResponse])
async def list_tasks(
    status: Optional[TaskStatus] = None,
    priority: Optional[TaskPriority] = None,
    limit: int = Field(50, ge=1, le=200),
    offset: int = Field(0, ge=0)
):
    """Retrieve tasks with optional filtering and pagination."""
    tasks = list(tasks_db.values())
    if status:
        tasks = [t for t in tasks if t["status"] == status]
    if priority:
        tasks = [t for t in tasks if t["priority"] == priority]
    # Apply pagination
    paginated = tasks[offset:offset + limit]
    return paginated

@router.post("/", response_model=TaskResponse, status_code=201)
async def create_task(task: TaskCreate):
    """Create a new task with validation."""
    global task_counter
    task_counter += 1
    now = datetime.utcnow()
    task_record = {
        "id": task_counter,
        **task.model_dump(),
        "status": TaskStatus.todo,
        "created_at": now,
        "updated_at": now,
    }
    tasks_db[task_counter] = task_record
    return task_record

@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int):
    """Retrieve a single task by ID."""
    if task_id not in tasks_db:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return tasks_db[task_id]

@router.patch("/{task_id}", response_model=TaskResponse)
async def update_task(task_id: int, status: TaskStatus):
    """Update task status."""
    if task_id not in tasks_db:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    tasks_db[task_id]["status"] = status
    tasks_db[task_id]["updated_at"] = datetime.utcnow()
    return tasks_db[task_id]

@router.delete("/{task_id}", status_code=204)
async def delete_task(task_id: int):
    """Delete a task permanently."""
    if task_id not in tasks_db:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    del tasks_db[task_id]
    return None

app.include_router(router)

Dependency Injection and Middleware

# dependencies.py
from fastapi import Depends, HTTPException, Header, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Optional
import secrets
import time

# Bearer token security scheme
security = HTTPBearer(auto_error=False)

class User:
    """Authenticated user model for dependency injection."""
    def __init__(self, user_id: str, role: str, permissions: list[str]):
        self.user_id = user_id
        self.role = role
        self.permissions = permissions

def verify_token(
    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
) -> Optional[User]:
    """Verify Bearer token and return authenticated user or None."""
    if credentials is None:
        return None
    # Simulated token verification — replace with real JWT validation
    token = credentials.credentials
    if token == "admin-secret-token":
        return User(user_id="admin-001", role="admin",
                     permissions=["read", "write", "delete"])
    elif token == "user-secret-token":
        return User(user_id="user-001", role="user",
                     permissions=["read"])
    raise HTTPException(status_code=401, detail="Invalid authentication token")

def get_current_user(user: Optional[User] = Depends(verify_token)):
    """Require authentication — raises 401 if no valid user."""
    if user is None:
        raise HTTPException(
            status_code=401,
            detail="Authentication required. Provide a valid Bearer token."
        )
    return user

def require_role(required_role: str):
    """Factory that creates a dependency requiring a specific role."""
    def role_checker(user: User = Depends(get_current_user)) -> User:
        if user.role != required_role and user.role != "admin":
            raise HTTPException(
                status_code=403,
                detail=f"Role '{required_role}' required, but you have '{user.role}'"
            )
        return user
    return role_checker

# Middleware for request timing
from starlette.middleware.base import BaseHTTPMiddleware

class TimingMiddleware(BaseHTTPMiddleware):
    """Adds X-Process-Time header to all responses."""
    async def dispatch(self, request: Request, call_next):
        start_time = time.perf_counter()
        response = await call_next(request)
        process_time = (time.perf_counter() - start_time) * 1000
        response.headers["X-Process-Time"] = f"{process_time:.2f}ms"
        return response

# main.py continued — adding secured endpoints
from dependencies import get_current_user, require_role, TimingMiddleware, User

app.add_middleware(TimingMiddleware)

# Protected admin router
admin_router = APIRouter(prefix="/admin", tags=["Admin"],
                         dependencies=[Depends(require_role("admin"))])

@admin_router.get("/system-info")
async def system_info(user: User = Depends(get_current_user)):
    """Admin-only endpoint returning system statistics."""
    return {
        "authenticated_user": user.user_id,
        "role": user.role,
        "total_tasks": len(tasks_db),
        "tasks_by_status": {
            status.value: sum(1 for t in tasks_db.values()
                            if t["status"] == status)
            for status in TaskStatus
        },
        "uptime_seconds": 86400,  # simulated
    }

@admin_router.delete("/tasks/all", status_code=204)
async def clear_all_tasks(user: User = Depends(get_current_user)):
    """Admin-only: clear all tasks from the database."""
    tasks_db.clear()
    return None

app.include_router(admin_router)

WebSocket Endpoint for Real-Time Updates

# websocket_handler