āĻ āĻŋāĻ āĻāĻā§, āĻāĻŦāĻžāϰ āĻāĻŽāĻŋ āĻāĻāĻĻāĻŽ āϤā§āĻŽāĻžāϰ āĻĻā§āĻāϝāĻŧāĻž āĻāύāĻā§āύā§āĻāĻāĻžāĻ āϧāϰ⧠â
āĻā§āϞ āĻā§āĻĄ āĻ āĻŋāĻ āĻāϰ⧠+ āϧāĻžāĻĒā§ āϧāĻžāĻĒā§ āĻāĻŋāĻāĻā§āϰāĻŋāϝāĻŧāĻžāϞ āĻāĻāĻžāϰ⧠āϏāĻžāĻāĻŋāϝāĻŧā§ āĻĻāĻŋāĻā§āĻāĻŋāĨ¤
āύāĻž āĻŦāĻžāĻĄāĻŧāϤāĻŋ āĻāĻĨāĻž, āύāĻž āĻāϞāĻžāĻĻāĻž āϏā§āĻāĻžāĻāϞ â āĻļā§āϧ⧠āϤā§āĻŽāĻžāϰ āϞā§āĻāĻž â āĻĒāϰāĻŋāώā§āĻāĻžāϰ â āϏāĻ āĻŋāĻ â āĻāĻŋāĻāĻā§āϰāĻŋāϝāĻŧāĻžāϞāĨ¤
đ 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 āϏāĻš āϏāĻžāĻāĻŋāϝāĻŧā§ āĻĻāĻŋāϤ⧠āĻĒāĻžāϰāĻŋ