← Back to DevBytes
Go July 07, 2026 14 min read

Django vs FastAPI: A Comprehensive Comparison for 2026

Introduction: The Python Web Framework Landscape in 2026

As we move deeper into the mid-2020s, Python remains the dominant language for backend web development. Two frameworks stand at the forefront: Django, the battle-tested full-stack framework with over two decades of maturity, and FastAPI, the high-performance async framework that has taken the industry by storm since its release. By 2026, both frameworks have evolved significantly — Django now ships with native async support, while FastAPI has matured its ecosystem with stable middleware, dependency injection patterns, and production-grade tooling. This comprehensive tutorial will guide you through every facet of both frameworks, helping you make an informed architectural decision for your next project.

What Is Django?

🚀 Deploy your AI agent in 10 minutes

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

Try it free →

Django is a full-stack web framework that follows the "batteries included" philosophy. Created in 2005, it provides an ORM, admin interface, form handling, authentication system, template engine, and session management — all out of the box. Django enforces a clean MTV (Model-Template-View) architectural pattern and emphasizes convention over configuration. By 2026, Django 5.x and beyond support asynchronous views, async ORM queries, and built-in WebSocket handling via Django Channels integration.

Core Django Concepts

What Is FastAPI?

FastAPI is a modern, high-performance web framework designed specifically for building APIs. Released in 2018, it leverages Python's native async/await syntax and is built on top of Starlette (for web routing) and Pydantic (for data validation). FastAPI is unopinionated about project structure but provides automatic interactive API documentation via OpenAPI/Swagger, dependency injection, and exceptional performance rivaling Node.js and Go frameworks. By 2026, FastAPI's ecosystem includes stable database integrations, admin panels, and production deployment patterns.

Core FastAPI Concepts

Why This Comparison Matters in 2026

The choice between Django and FastAPI is no longer simply "monolith vs microservices." Django's async capabilities now allow it to handle WebSocket connections and long-lived requests efficiently. FastAPI's ecosystem has grown to include admin interfaces (SQLAdmin), ORM integrations (SQLAlchemy 2.0+), and file upload handling that rivals Django's. The decision impacts:

Architectural Differences

Django: Synchronous MTV with Async Overlay

Django's architecture centers on the request-response cycle. A request hits the URL router, passes through middleware, reaches a view (which may query the ORM and render a template), and returns an HTTP response. Historically this was entirely synchronous WSGI-based. Django 4.x+ introduced async views that can be served under ASGI, allowing coroutines to handle I/O-bound work. However, the ORM still defaults to synchronous queries, requiring sync_to_async wrappers or the newer async ORM interface for non-blocking database access.

FastAPI: Async-First with Optional Sync Fallback

FastAPI is built on an ASGI foundation from day one. Every path operation can be async def, allowing the framework to handle thousands of concurrent connections without thread pools. Sync functions are also supported — FastAPI wraps them in a thread pool executor automatically. Pydantic models serve as the single source of truth for serialization, validation, and documentation. There is no built-in ORM, template engine, or admin interface — you choose your own tools, typically SQLAlchemy 2.0+ with async sessions for database work.

Code Examples: Building a REST API

Let's build the same REST API — a task management system with CRUD operations — in both frameworks. This side-by-side comparison illuminates the developer experience.

Django with Django REST Framework (DRF)

First, install dependencies and create the project:

# Install Django and DRF
pip install django djangorestframework

# Create project and app
django-admin startproject taskmanager .
python manage.py startapp tasks

Define the model in tasks/models.py:

from django.db import models
from django.utils import timezone

class Task(models.Model):
    STATUS_CHOICES = [
        ('pending', 'Pending'),
        ('in_progress', 'In Progress'),
        ('completed', 'Completed'),
    ]
    
    title = models.CharField(max_length=200)
    description = models.TextField(blank=True, null=True)
    status = models.CharField(
        max_length=20, 
        choices=STATUS_CHOICES, 
        default='pending'
    )
    due_date = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        ordering = ['-created_at']
    
    def __str__(self):
        return self.title

Create the serializer in tasks/serializers.py:

from rest_framework import serializers
from .models import Task

class TaskSerializer(serializers.ModelSerializer):
    days_remaining = serializers.SerializerMethodField()
    
    class Meta:
        model = Task
        fields = [
            'id', 'title', 'description', 'status',
            'due_date', 'days_remaining', 'created_at', 'updated_at'
        ]
        read_only_fields = ['id', 'created_at', 'updated_at']
    
    def get_days_remaining(self, obj):
        if obj.due_date:
            delta = obj.due_date - timezone.now()
            return max(0, delta.days)
        return None
    
    def validate_due_date(self, value):
        if value and value < timezone.now():
            raise serializers.ValidationError(
                "Due date cannot be in the past."
            )
        return value

Build the ViewSet in tasks/views.py:

from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from .models import Task
from .serializers import TaskSerializer

class TaskViewSet(viewsets.ModelViewSet):
    queryset = Task.objects.all()
    serializer_class = TaskSerializer
    permission_classes = [IsAuthenticatedOrReadOnly]
    
    def get_queryset(self):
        queryset = super().get_queryset()
        status_filter = self.request.query_params.get('status')
        if status_filter:
            queryset = queryset.filter(status=status_filter)
        return queryset
    
    @action(detail=True, methods=['post'])
    def mark_completed(self, request, pk=None):
        task = self.get_object()
        task.status = 'completed'
        task.save()
        return Response(
            TaskSerializer(task).data,
            status=status.HTTP_200_OK
        )
    
    def perform_create(self, serializer):
        serializer.save()

Wire up URLs in tasks/urls.py:

from rest_framework.routers import DefaultRouter
from .views import TaskViewSet

router = DefaultRouter()
router.register(r'tasks', TaskViewSet, basename='task')

urlpatterns = router.urls

Finally, update the project's urls.py and settings.py:

# In taskmanager/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/v1/', include('tasks.urls')),
]

# In settings.py, add to INSTALLED_APPS:
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'tasks',
]

# Add DRF configuration
REST_FRAMEWORK = {
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 20,
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.TokenAuthentication',
    ],
}

Run migrations and start the server:

python manage.py makemigrations
python manage.py migrate
python manage.py runserver

Your API is now live at http://localhost:8000/api/v1/tasks/ with full CRUD, filtering, pagination, and a custom mark_completed action. The browsable API (a DRF feature) lets you interact with endpoints directly in the browser.

FastAPI Equivalent

Install dependencies:

pip install fastapi uvicorn sqlalchemy asyncpg pydantic

Define Pydantic schemas in schemas.py:

from pydantic import BaseModel, Field, field_validator
from datetime import datetime, timezone
from typing import Optional
from enum import Enum

class TaskStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"

class TaskBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = Field(None, max_length=2000)
    status: TaskStatus = TaskStatus.PENDING
    due_date: Optional[datetime] = None
    
    @field_validator('due_date')
    @classmethod
    def validate_due_date_not_past(cls, v):
        if v and v < datetime.now(timezone.utc):
            raise ValueError("Due date cannot be in the past")
        return v

class TaskCreate(TaskBase):
    pass

class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    status: Optional[TaskStatus] = None
    due_date: Optional[datetime] = None

class TaskResponse(TaskBase):
    id: int
    days_remaining: Optional[int] = None
    created_at: datetime
    updated_at: datetime
    
    model_config = {"from_attributes": True}

Set up SQLAlchemy models in models.py:

from sqlalchemy import Column, Integer, String, Text, DateTime, Enum as SAEnum
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
from datetime import datetime, timezone

Base = declarative_base()

class TaskModel(Base):
    __tablename__ = "tasks"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    description = Column(Text, nullable=True)
    status = Column(
        SAEnum('pending', 'in_progress', 'completed', name='task_status'),
        default='pending',
        nullable=False
    )
    due_date = Column(DateTime(timezone=True), nullable=True)
    created_at = Column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False
    )
    updated_at = Column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False
    )

Create database configuration in database.py:

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    async_sessionmaker
)
from sqlalchemy.orm import sessionmaker
from models import Base

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

engine = create_async_engine(DATABASE_URL, echo=True)

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

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Build the main application in main.py:

from fastapi import FastAPI, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List, Optional
from datetime import datetime, timezone

from database import get_db, init_db
from models import TaskModel
from schemas import TaskCreate, TaskUpdate, TaskResponse, TaskStatus

app = FastAPI(
    title="Task Manager API",
    version="1.0.0",
    description="A high-performance task management API"
)

@app.on_event("startup")
async def startup():
    await init_db()

@app.get("/api/v1/tasks", response_model=List[TaskResponse])
async def list_tasks(
    status_filter: Optional[TaskStatus] = Query(None, alias="status"),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db)
):
    query = select(TaskModel)
    if status_filter:
        query = query.where(TaskModel.status == status_filter.value)
    query = query.offset(skip).limit(limit).order_by(TaskModel.created_at.desc())
    result = await db.execute(query)
    tasks = result.scalars().all()
    
    response_tasks = []
    for task in tasks:
        task_dict = {
            "id": task.id,
            "title": task.title,
            "description": task.description,
            "status": task.status,
            "due_date": task.due_date,
            "created_at": task.created_at,
            "updated_at": task.updated_at,
            "days_remaining": None
        }
        if task.due_date:
            delta = task.due_date - datetime.now(timezone.utc)
            task_dict["days_remaining"] = max(0, delta.days)
        response_tasks.append(TaskResponse.model_validate(task_dict))
    
    return response_tasks

@app.get("/api/v1/tasks/{task_id}", response_model=TaskResponse)
async def get_task(task_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(TaskModel).where(TaskModel.id == task_id)
    )
    task = result.scalar_one_or_none()
    if not task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found"
        )
    
    task_dict = {
        "id": task.id,
        "title": task.title,
        "description": task.description,
        "status": task.status,
        "due_date": task.due_date,
        "created_at": task.created_at,
        "updated_at": task.updated_at,
        "days_remaining": None
    }
    if task.due_date:
        delta = task.due_date - datetime.now(timezone.utc)
        task_dict["days_remaining"] = max(0, delta.days)
    
    return TaskResponse.model_validate(task_dict)

@app.post(
    "/api/v1/tasks",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED
)
async def create_task(task: TaskCreate, db: AsyncSession = Depends(get_db)):
    task_data = task.model_dump()
    db_task = TaskModel(**task_data)
    db.add(db_task)
    await db.flush()
    await db.refresh(db_task)
    
    task_dict = {
        "id": db_task.id,
        "title": db_task.title,
        "description": db_task.description,
        "status": db_task.status,
        "due_date": db_task.due_date,
        "created_at": db_task.created_at,
        "updated_at": db_task.updated_at,
        "days_remaining": None
    }
    if db_task.due_date:
        delta = db_task.due_date - datetime.now(timezone.utc)
        task_dict["days_remaining"] = max(0, delta.days)
    
    return TaskResponse.model_validate(task_dict)

@app.patch("/api/v1/tasks/{task_id}", response_model=TaskResponse)
async def update_task(
    task_id: int,
    task_update: TaskUpdate,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(TaskModel).where(TaskModel.id == task_id)
    )
    db_task = result.scalar_one_or_none()
    if not db_task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found"
        )
    
    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_task, key, value)
    
    await db.flush()
    await db.refresh(db_task)
    
    task_dict = {
        "id": db_task.id,
        "title": db_task.title,
        "description": db_task.description,
        "status": db_task.status,
        "due_date": db_task.due_date,
        "created_at": db_task.created_at,
        "updated_at": db_task.updated_at,
        "days_remaining": None
    }
    if db_task.due_date:
        delta = db_task.due_date - datetime.now(timezone.utc)
        task_dict["days_remaining"] = max(0, delta.days)
    
    return TaskResponse.model_validate(task_dict)

@app.delete("/api/v1/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(TaskModel).where(TaskModel.id == task_id)
    )
    db_task = result.scalar_one_or_none()
    if not db_task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found"
        )
    await db.delete(db_task)
    return None

@app.post("/api/v1/tasks/{task_id}/mark-completed", response_model=TaskResponse)
async def mark_task_completed(task_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(TaskModel).where(TaskModel.id == task_id)
    )
    db_task = result.scalar_one_or_none()
    if not db_task:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Task with id {task_id} not found"
        )
    
    db_task.status = "completed"
    await db.flush()
    await db.refresh(db_task)
    
    task_dict = {
        "id": db_task.id,
        "title": db_task.title,
        "description": db_task.description,
        "status": db_task.status,
        "due_date": db_task.due_date,
        "created_at": db_task.created_at,
        "updated_at": db_task.updated_at,
        "days_remaining": None
    }
    if db_task.due_date:
        delta = db_task.due_date - datetime.now(timezone.utc)
        task_dict["days_remaining"] = max(0, delta.days)
    
    return TaskResponse.model_validate(task_dict)

Run the server:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Visit http://localhost:8000/docs for interactive Swagger documentation. Notice that FastAPI requires more explicit database code (no ORM magic like DRF's ModelViewSet), but gives you complete control and async performance.

Performance Benchmarks

Performance is one of the most discussed differences. In controlled benchmarks using a simple JSON echo endpoint, FastAPI (running on Uvicorn with uvloop) consistently achieves 3-5x higher throughput than Django's WSGI mode and approximately 2x higher than Django's ASGI mode with async views. Here's a representative benchmark comparison for a simple API endpoint returning JSON:

# Benchmark setup using wrk for 30 seconds with 100 concurrent connections
# Django (gunicorn with 4 workers, sync)
wrk -t4 -c100 -d30s http://localhost:8000/api/echo
# Results: ~3,200 requests/sec, avg latency 28ms

# Django (daphne with 4 workers, async)
wrk -t4 -c100 -d30s http://localhost:8000/api/echo
# Results: ~5,800 requests/sec, avg latency 16ms

# FastAPI (uvicorn with 4 workers)
wrk -t4 -c100 -d30s http://localhost:8000/api/echo
# Results: ~14,500 requests/sec, avg latency 6ms

The performance gap widens dramatically under high concurrency. FastAPI's async architecture allows a single worker to handle thousands of simultaneous connections, while Django's sync workers are limited by the number of available threads. For WebSocket-heavy applications or real-time features, FastAPI's native ASGI support gives it a decisive advantage.

Async Capabilities

Django's Async Journey

Django's async support has been steadily improving. As of Django 5.x (the stable version by 2026), you can write:

# Async view in Django 5.x
from django.http import JsonResponse
from django.views.decorators.cache import cache_page
import asyncio

async def async_task_list(request):
    # Async ORM queries (Django 5.x+)
    from tasks.models import Task
    tasks = []
    async for task in Task.objects.filter(status='pending'):
        tasks.append({
            'id': task.id,
            'title': task.title,
            'status': task.status,
        })
    
    # Async external API call
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.external-service.com/data')
        external_data = response.json()
    
    return JsonResponse({
        'tasks': tasks,
        'external': external_data
    })

However, limitations remain. The Django admin, template rendering, and many middleware components still operate synchronously. The async ORM interface supports async for iteration and async queries, but complex queryset chaining with annotations and aggregations may require sync_to_async wrappers for full compatibility.

FastAPI's Async-First Design

FastAPI's entire request lifecycle is async by default. Every dependency, middleware, and path operation can be a coroutine:

from fastapi import FastAPI, Depends
import httpx
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_http_client() -> httpx.AsyncClient:
    async with httpx.AsyncClient(
        base_url="https://api.external-service.com",
        timeout=30.0
    ) as client:
        yield client

@app.get("/aggregated-data")
async def get_aggregated_data(
    db: AsyncSession = Depends(get_db),
    client: httpx.AsyncClient = Depends(get_http_client)
):
    # Execute database query and external API call concurrently
    db_task = asyncio.create_task(
        db.execute(select(TaskModel).where(TaskModel.status == 'pending'))
    )
    api_task = asyncio.create_task(
        client.get('/data')
    )
    
    # Await both concurrently
    db_result, api_response = await asyncio.gather(db_task, api_task)
    
    tasks = db_result.scalars().all()
    external_data = api_response.json()
    
    return {
        'tasks': [
            {'id': t.id, 'title': t.title} for t in tasks
        ],
        'external': external_data
    }

This concurrent execution model is natural in FastAPI but requires careful orchestration in Django.

Database ORM and Migrations

Django ORM: Mature and Feature-Rich

Django's ORM is arguably the most polished in the Python ecosystem. It supports:

# Django ORM complex query example
from django.db.models import Count, Q, Avg, OuterRef, Subquery
from tasks.models import Task

# Complex aggregation with conditional annotation
report = (
    Task.objects
    .filter(
        Q(status='completed') | Q(due_date__gte=timezone.now())
    )
    .values('status')
    .annotate(
        count=Count('id'),
        avg_completion_days=Avg(
            # Subquery for completion time calculation
            Task.objects.filter(
                status='completed',
                id=OuterRef('id')
            ).annotate(
                completion_time=models.F('updated_at') - models.F('created_at')
            ).values('completion_time')
        )
    )
    .order_by('-count')
)

# Prefetch with selective loading
tasks = (
    Task.objects
    .select_related('assigned_to')
    .prefetch_related('comments', 'attachments')
    .only('title', 'status', 'assigned_to__username')
    .filter(status='in_progress')
)

FastAPI + SQLAlchemy 2.0: Powerful but Manual

FastAPI doesn't ship with an ORM. The de facto standard is SQLAlchemy 2.0+ with async sessions. Migrations require Alembic, which you configure separately:

# Initialize Alembic for async SQLAlchemy
alembic init alembic

# alembic/env.py configuration for async
from sqlalchemy.ext.asyncio import create_async_engine
from models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/taskdb"
    )
    
    async def do_migrations(connection):
        await connection.run_sync(
            lambda sync_conn: target_metadata.create_all(sync_conn)
        )
    
    asyncio.run(do_migrations(connectable))

# Generate migration after model changes
# alembic revision --autogenerate -m "add priority field"
# alembic upgrade head

SQLAlchemy 2.0's async ORM supports the full query builder:

from sqlalchemy import select, func, and_
from sqlalchemy.ext.asyncio import AsyncSession

async def get_complex_report(db: AsyncSession):
    # Subquery for task statistics
    completed_subq = (
        select(
            TaskModel.assigned_user_id,
            func.count(TaskModel.id).label('total_tasks'),
            func.avg(
                func.extract('epoch', TaskModel.updated_at) -
                func.extract('epoch', TaskModel.created_at)
            ).label('avg_completion_seconds')
        )
        .where(TaskModel.status == 'completed')
        .group_by(TaskModel.assigned_user_id)
        .subquery()
    )
    
    # Main query joining with subquery
    query = (
        select(UserModel, completed_subq.c.total_tasks)
        .outerjoin(
            completed_subq,
            UserModel.id == completed_subq.c.assigned_user_id
        )
        .where(UserModel.is_active == True)
    )
    
    result = await db.execute(query)
    return result.all()

Authentication and Security

Django: Batteries Included Auth

Django provides a complete authentication system with User models, groups, permissions, session management, password hashing, and CSRF protection — all configured and working out of the box:

# Django's built-in auth in action
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.decorators import login_required, permission_required
from django.contrib.auth.models import User, Group, Permission

# Login view (session-based)
def user_login(request):
    username = request.POST['username']
    password = request.POST['password']
    user = authenticate(request, username=username, password=password)
    if user is not None:
        login(request, user)
        # Session cookie set automatically with HttpOnly flag
        return JsonResponse({'status': 'authenticated'})
    return JsonResponse({'error': 'Invalid credentials'}, status=401)

# Protected view with decorator
@login_required
@permission_required('tasks.can_manage_tasks', raise_exception=True)
def admin_dashboard(request):
    return JsonResponse({'message': 'Welcome to admin dashboard'})

# CSRF protection is automatic for all POST/PUT/DELETE requests
# Token is injected into templates via {% csrf_token %} or 
# available as cookie for JS frameworks

For API token authentication, Django REST Framework provides TokenAuthentication, JWT (via Simple JWT), and OAuth2 integrations.

FastAPI: Flexible but DIY

FastAPI provides security utilities but doesn't enforce any authentication pattern. You implement auth logic yourself, typically using JWT tokens:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from pydantic import BaseModel

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

SECRET_KEY = "your-secret-key-at-least-32-chars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class UserInDB(BaseModel):
    id: int
    username: str
    email: str
    hashed_password: str
    is_active: bool

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> UserInDB:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    result = await db.execute(
        select(UserModel).where(UserModel.id == user_id)
    )
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    
    return UserInDB(
        id=user.id,
        username=user.username,
        email=user

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles