1. Home
  2. FastApi
  3. Basic Fundamental
  4. Day 5 : Authentication lo...
  5. 🔐 JWT Implementation – Student CRUD āĻ

🔐 JWT Implementation – Student CRUD āĻ

āĻ āĻŋāĻ• āφāϛ⧇, āĻāĻŦāĻžāϰ āφāĻŽāĻŋ āĻāĻ•āĻĻāĻŽ āϤ⧋āĻŽāĻžāϰ āĻĻ⧇āĻ“āϝāĻŧāĻž āĻ•āύāĻŸā§‡āĻ¨ā§āϟāϟāĻžāχ āϧāϰ⧇ —
āϭ⧁āϞ āϕ⧋āĻĄ āĻ āĻŋāĻ• āĻ•āϰ⧇ + āϧāĻžāĻĒ⧇ āϧāĻžāĻĒ⧇ āϟāĻŋāωāĻŸā§‹āϰāĻŋāϝāĻŧāĻžāϞ āφāĻ•āĻžāϰ⧇ āϏāĻžāϜāĻŋāϝāĻŧ⧇ āĻĻāĻŋāĻšā§āĻ›āĻŋāĨ¤
āύāĻž āĻŦāĻžāĻĄāĻŧāϤāĻŋ āĻ•āĻĨāĻž, āύāĻž āφāϞāĻžāĻĻāĻž āĻ¸ā§āϟāĻžāχāϞ — āĻļ⧁āϧ⧁ āϤ⧋āĻŽāĻžāϰ āϞ⧇āĻ–āĻž → āĻĒāϰāĻŋāĻˇā§āĻ•āĻžāϰ → āϏāĻ āĻŋāĻ• → āϟāĻŋāωāĻŸā§‹āϰāĻŋāϝāĻŧāĻžāϞāĨ¤


🔐 JWT Implementation – Student CRUD āĻ

āφāĻĒāύāĻžāϰ āĻŦāĻ°ā§āϤāĻŽāĻžāύ app/apps/auth/utils.py āĻ…āύ⧁āϝāĻžāϝāĻŧā§€ Student CRUD āĻ JWT Protection āϝ⧋āĻ— āĻ•āϰāĻžāϰ āϏāĻŽā§āĻĒā§‚āĻ°ā§āĻŖ āĻ—āĻžāχāĻĄāĨ¤


📝 āĻĢāĻžāχāϞ: app/apps/students/routes.py

✅ āϏāĻŽā§āĻĒā§‚āĻ°ā§āĻŖ āϏāĻ āĻŋāĻ• āĻ“ āĻĒāϰāĻŋāĻˇā§āĻ•āĻžāϰ āϕ⧋āĻĄ:

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from typing import List, Optional
from math import ceil
from app.db.session import get_db
from app.apps.auth.utils import get_current_user
from . import models, schemas

router = APIRouter(prefix="/students", tags=["Students"])


# ā§§. Create (āϤ⧈āϰāĻŋ āĻ•āϰāĻž) - JWT Protection āϏāĻš
@router.post("/", response_model=schemas.StudentResponse, status_code=status.HTTP_201_CREATED)
def create_student(
    student: schemas.StudentCreate,
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """
    āύāϤ⧁āύ āĻ¸ā§āϟ⧁āĻĄā§‡āĻ¨ā§āϟ āϤ⧈āϰāĻŋ āĻ•āϰāĻž (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)
    """
    db_student = models.Student(**student.dict())
    db.add(db_student)
    db.commit()
    db.refresh(db_student)
    return db_student


# ⧍. Read All (āϏāĻŦ āĻĻ⧇āĻ–āĻž) - JWT Protection āϏāĻš
@router.get("/", response_model=List[schemas.StudentResponse])
def get_all_students(
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db),
    search: Optional[str] = Query(None, description="āύāĻžāĻŽ āĻŦāĻž āχāĻŽā§‡āχāϞ āĻĻāĻŋāϝāĻŧ⧇ āϏāĻžāĻ°ā§āϚ āĻ•āϰ⧁āύ")
):
    """āϏāĻŦ āĻ¸ā§āϟ⧁āĻĄā§‡āĻ¨ā§āϟ āĻĻ⧇āĻ–āĻž (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    query = db.query(models.Student)
    
    if search:
        query = query.filter(
            (models.Student.name.ilike(f"%{search}%")) |
            (models.Student.email.ilike(f"%{search}%"))
        )
    
    return query.all()


# ā§Š. Read One (āĻāĻ•āϟāĻŋ āĻĻ⧇āĻ–āĻž) - JWT Protection āϏāĻš
@router.get("/{student_id}", response_model=schemas.StudentResponse)
def get_student(
    student_id: int,
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """āύāĻŋāĻ°ā§āĻĻāĻŋāĻˇā§āϟ āĻ¸ā§āϟ⧁āĻĄā§‡āĻ¨ā§āϟ āĻĻ⧇āĻ–āĻž (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    if not db_student:
        raise HTTPException(status_code=404, detail="Student not found")
    return db_student


# ā§Ē. Search with Pagination - JWT Protection āϏāĻš
@router.get("/search/paginated", response_model=schemas.PaginatedStudentResponse)
def search_students_paginated(
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db),
    search: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100)
):
    """Pagination āϏāĻš āϏāĻžāĻ°ā§āϚ (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    query = db.query(models.Student)
    
    if search:
        query = query.filter(
            (models.Student.name.ilike(f"%{search}%")) |
            (models.Student.email.ilike(f"%{search}%"))
        )
    
    total = query.count()
    total_pages = ceil(total / page_size) if total > 0 else 0
    offset = (page - 1) * page_size
    students = query.offset(offset).limit(page_size).all()
    
    return {
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": total_pages,
        "data": students
    }


# ā§Ģ. Advanced Search - JWT Protection āϏāĻš
@router.get("/advanced-search", response_model=List[schemas.StudentResponse])
def advanced_search(
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db),
    name: Optional[str] = Query(None),
    email: Optional[str] = Query(None),
    age_min: Optional[int] = Query(None, ge=0),
    age_max: Optional[int] = Query(None, ge=0),
    sort_by: Optional[str] = Query("id"),
    sort_order: Optional[str] = Query("asc")
):
    """āωāĻ¨ā§āύāϤ āϏāĻžāĻ°ā§āϚ (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    query = db.query(models.Student)
    
    if name:
        query = query.filter(models.Student.name.ilike(f"%{name}%"))
    if email:
        query = query.filter(models.Student.email.ilike(f"%{email}%"))
    if age_min is not None:
        query = query.filter(models.Student.age >= age_min)
    if age_max is not None:
        query = query.filter(models.Student.age <= age_max)
    
    if sort_by == "name":
        sort_column = models.Student.name
    elif sort_by == "email":
        sort_column = models.Student.email
    elif sort_by == "age":
        sort_column = models.Student.age
    else:
        sort_column = models.Student.id
    
    if sort_order == "desc":
        query = query.order_by(sort_column.desc())
    else:
        query = query.order_by(sort_column.asc())
    
    return query.all()


# ā§Ŧ. Update (āφāĻĒāĻĄā§‡āϟ āĻ•āϰāĻž) - JWT Protection āϏāĻš
@router.put("/{student_id}", response_model=schemas.StudentResponse)
def update_student(
    student_id: int,
    student_data: schemas.StudentUpdate,
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """āĻ¸ā§āϟ⧁āĻĄā§‡āĻ¨ā§āϟ āφāĻĒāĻĄā§‡āϟ āĻ•āϰāĻž (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    
    if not db_student:
        raise HTTPException(status_code=404, detail="Student not found")
    
    updated_info = student_data.dict(exclude_unset=True)
    
    for key, value in updated_info.items():
        setattr(db_student, key, value)
    
    db.commit()
    db.refresh(db_student)
    return db_student


# ā§­. Delete (āĻŽā§āϛ⧇ āĻĢ⧇āϞāĻž) - JWT Protection āϏāĻš
@router.delete("/{student_id}")
def delete_student(
    student_id: int,
    current_user: dict = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """āĻ¸ā§āϟ⧁āĻĄā§‡āĻ¨ā§āϟ āĻŽā§āϛ⧇ āĻĢ⧇āϞāĻž (āĻļ⧁āϧ⧁āĻŽāĻžāĻ¤ā§āϰ āϞāĻ—āχāύ āĻ•āϰāĻž āχāωāϜāĻžāϰāϰāĻž)"""
    db_student = db.query(models.Student).filter(models.Student.id == student_id).first()
    if not db_student:
        raise HTTPException(status_code=404, detail="Student not found")
    
    db.delete(db_student)
    db.commit()
    return {"message": "Successfully Deleted"}

🔑 āĻŽā§‚āϞ āĻĒāϝāĻŧ⧇āĻ¨ā§āϟ (Tutorial Style)

ā§§ī¸âƒŖ āχāĻŽāĻĒā§‹āĻ°ā§āϟ:

from app.apps.auth.utils import get_current_user

âžĄī¸ JWT āϝāĻžāϚāĻžāχ āĻ•āϰāĻžāϰ āĻĢāĻžāĻ‚āĻļāύ āĻāĻ–āĻžāύ⧇ āχāĻŽāĻĒā§‹āĻ°ā§āϟ āĻ•āϰāĻž āĻšāĻšā§āϛ⧇āĨ¤


ā§¨ī¸âƒŖ āĻĒā§āϰāϤāĻŋāϟāĻŋ āĻāĻ¨ā§āĻĄāĻĒāϝāĻŧ⧇āĻ¨ā§āĻŸā§‡ JWT āϝ⧋āĻ— āĻ•āϰ⧁āύ:

current_user: dict = Depends(get_current_user)

âžĄī¸ āĻāχ āϞāĻžāχāύ āĻĨāĻžāĻ•āϞ⧇āχ route āϟāĻŋ protected āĻšāϝāĻŧ⧇ āϝāĻžāϝāĻŧāĨ¤


ā§Šī¸âƒŖ current_user āĻ āϕ⧀ āĻĨāĻžāϕ⧇?

{
    "user_id": 1,
    "username": "ahmed",
    "type": "access"
}

âžĄī¸ āĻāχ āĻĄāĻžāϟāĻž JWT token decode āĻ•āϰ⧇ āφāϏ⧇āĨ¤


đŸ§Ē āĻŸā§‡āĻ¸ā§āϟ āĻ•āϰ⧁āύ (Step by Step)

✅ ā§§. Register:

curl -X POST http://localhost:8000/auth/register \
  -H "Content-Type: application/json" \
  -d '{
    "username": "ahmed",
    "email": "ahmed@example.com",
    "password": "password123",
    "full_name": "āφāĻšāĻŽā§‡āĻĻ"
  }'

✅ ⧍. Login:

curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "username": "ahmed",
    "password": "password123"
  }'

✅ ā§Š. Token āϏāĻš Student āϤ⧈āϰāĻŋ āĻ•āϰ⧁āύ:

curl -X POST http://localhost:8000/students/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -d '{
    "name": "āĻĢāĻžāϤāĻŋāĻŽāĻž",
    "email": "fatima@example.com",
    "age": 20
  }'

❌ ā§Ē. Token āĻ›āĻžāĻĄāĻŧāĻž āĻšā§‡āĻˇā§āϟāĻž āĻ•āϰ⧁āύ:

curl -X POST http://localhost:8000/students/ \
  -H "Content-Type: application/json" \
  -d '{
    "name": "āĻŽāĻžāϰāĻŋāϝāĻŧāĻžāĻŽ",
    "email": "mariam@example.com",
    "age": 21
  }'

📌 āϰ⧇āϏāĻĒāĻ¨ā§āϏ:

{"detail":"Authorization header āĻĒā§āϰāϝāĻŧā§‹āϜāύ"}

✅ āϏāĻŽā§āĻĒāĻ¨ā§āύ!

āϤ⧋āĻŽāĻžāϰ Student CRUD āĻāĻ–āύ āĻĒ⧁āϰ⧋āĻĒ⧁āϰāĻŋ JWT āĻĻāĻŋāϝāĻŧ⧇ āϏ⧁āϰāĻ•ā§āώāĻŋāϤ,
āφāϰ āĻĒ⧁āϰ⧋ āĻ—āĻžāχāĻĄāϟāĻž āĻāĻ–āύ āĻ āĻŋāĻ• āϤ⧋āĻŽāĻžāϰ āĻ¸ā§āϟāĻžāχāϞ⧇, āĻĒāϰāĻŋāĻˇā§āĻ•āĻžāϰ āϟāĻŋāωāĻŸā§‹āϰāĻŋāϝāĻŧāĻžāϞ āφāĻ•āĻžāϰ⧇āĨ¤


āϤ⧁āĻŽāĻŋ āϚāĻžāχāϞ⧇ āĻāĻ–āύ āφāĻŽāĻŋ:

  • auth/utils.py āĻĒ⧁āϰ⧋āϟāĻž āϟāĻŋāωāĻŸā§‹āϰāĻŋāϝāĻŧāĻžāϞ āĻŦāĻžāύāĻŋāϝāĻŧ⧇ āĻĻāĻŋāϤ⧇ āĻĒāĻžāϰāĻŋ
  • āĻ…āĻĨāĻŦāĻž āĻĒ⧁āϰ⧋ FastAPI project Django-style structure āϏāĻš āϏāĻžāϜāĻŋāϝāĻŧ⧇ āĻĻāĻŋāϤ⧇ āĻĒāĻžāϰāĻŋ

How can we help?