Day 5 : Authentication login register jwt token

pip install fastapi uvicorn sqlalchemy pydantic[email] python-jose[cryptography] passlib[argon2] python-multipart

pip install -r requirements.txt

alembic init alembic

app/
├── apps/
│   ├── auth/                    # নতুন অ্যাপ
│   │   ├── __init__.py
│   │   ├── models.py            # User মডেল
│   │   ├── schemas.py           # User স্কিমা
│   │   ├── routes.py            # Auth রাউটস
│   │   └── utils.py             # JWT এবং পাসওয়ার্ড হেল্পার
│   └── students/
│       ├── models.py
│       ├── schemas.py
│       └── routes.py
├── db/
│   └── session.py
└── main.py

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from app.db.session import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(255), unique=True, nullable=False, index=True)
    hashed_password = Column(String(255), nullable=False)
    full_name = Column(String(100), nullable=True)
    is_active = Column(Integer, default=1)  # 1 = সক্রিয়, 0 = নিষ্ক্রিয়
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

from pydantic import BaseModel, EmailStr, Field, field_validator
from typing import Optional
from datetime import datetime

class UserRegister(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, description="ইউজারনেম (৩-৫০ ক্যারেক্টার)")
    email: EmailStr
    password: str = Field(..., min_length=6, max_length=128, description="পাসওয়ার্ড (৬-১২৮ ক্যারেক্টার)")
    full_name: Optional[str] = Field(None, max_length=100)
    
    @field_validator('password')
    @classmethod
    def validate_password(cls, v):
        if len(v.encode('utf-8')) > 128:
            raise ValueError('পাসওয়ার্ড খুব দীর্ঘ (সর্বোচ্চ ১২৮ বাইট)')
        return v

class UserLogin(BaseModel):
    username: str
    password: str

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    full_name: Optional[str]
    is_active: int
    created_at: datetime

    class Config:
        from_attributes = True

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"
    expires_in: int

class TokenData(BaseModel):
    user_id: int
    username: str

from passlib.context import CryptContext
from datetime import datetime, timedelta
from jose import JWTError, jwt
from typing import Optional

pwd_context = CryptContext(
    schemes=["argon2"],
    deprecated="auto"
)

SECRET_KEY = "your-secret-key-change-this-in-production-use-strong-random-string"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

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

def create_access_token(user_id: int, username: str) -> tuple:
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode = {"user_id": user_id, "username": username, "exp": expire, "type": "access"}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt, ACCESS_TOKEN_EXPIRE_MINUTES * 60

def create_refresh_token(user_id: int, username: str) -> str:
    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode = {"user_id": user_id, "username": username, "exp": expire, "type": "refresh"}
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str) -> Optional[dict]:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("user_id")
        username = payload.get("username")
        token_type = payload.get("type")
        if user_id is None or username is None:
            return None
        return {"user_id": user_id, "username": username, "type": token_type}
    except JWTError:
        return None

from fastapi import APIRouter, Depends, HTTPException, status, Header
from sqlalchemy.orm import Session
from typing import Optional
from app.db.session import get_db
from . import models, schemas, utils

router = APIRouter(prefix="/auth", tags=["Authentication"])

@router.post("/register", response_model=schemas.UserResponse, status_code=status.HTTP_201_CREATED)
def register(user_data: schemas.UserRegister, db: Session = Depends(get_db)):
    existing_user = db.query(models.User).filter(models.User.username == user_data.username).first()
    if existing_user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="ইউজারনেম ইতিমধ্যে ব্যবহৃত হয়েছে")
    
    existing_email = db.query(models.User).filter(models.User.email == user_data.email).first()
    if existing_email:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="ইমেইল ইতিমধ্যে ব্যবহৃত হয়েছে")
    
    hashed_password = utils.hash_password(user_data.password)
    db_user = models.User(username=user_data.username, email=user_data.email, hashed_password=hashed_password, full_name=user_data.full_name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@router.post("/login", response_model=schemas.TokenResponse)
def login(user_data: schemas.UserLogin, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.username == user_data.username).first()
    if not db_user or not utils.verify_password(user_data.password, db_user.hashed_password):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ইউজারনেম বা পাসওয়ার্ড ভুল")
    if not db_user.is_active:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="এই অ্যাকাউন্ট নিষ্ক্রিয়")
    
    access_token, expires_in = utils.create_access_token(db_user.id, db_user.username)
    refresh_token = utils.create_refresh_token(db_user.id, db_user.username)
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "expires_in": expires_in}

@router.post("/refresh", response_model=schemas.TokenResponse)
def refresh_access_token(refresh_token: str, db: Session = Depends(get_db)):
    token_data = utils.verify_token(refresh_token)
    if not token_data or token_data.get("type") != "refresh":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="অবৈধ Refresh Token")
    db_user = db.query(models.User).filter(models.User.id == token_data["user_id"]).first()
    if not db_user or not db_user.is_active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ইউজার খুঁজে পাওয়া যায়নি")
    
    access_token, expires_in = utils.create_access_token(db_user.id, db_user.username)
    return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer", "expires_in": expires_in}

@router.get("/me", response_model=schemas.UserResponse)
def get_current_user(authorization: Optional[str] = Header(None), db: Session = Depends(get_db)):
    if not authorization:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Authorization header প্রয়োজন")
    try:
        scheme, token = authorization.split()
        if scheme.lower() != "bearer":
            raise ValueError
    except ValueError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="অবৈধ Authorization header ফরম্যাট")
    
    token_data = utils.verify_token(token)
    if not token_data or token_data.get("type") != "access":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="অবৈধ বা এক্সপায়ার্ড Token")
    
    db_user = db.query(models.User).filter(models.User.id == token_data["user_id"]).first()
    if not db_user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ইউজার খুঁজে পাওয়া যায়নি")
    return db_user

from fastapi import FastAPI
from app.db.session import engine, Base
from app.apps.students import routes as student_routes
from app.apps.auth import routes as auth_routes

Base.metadata.create_all(bind=engine)

app = FastAPI(title="Pro Student CRUD with Authentication")

app.include_router(auth_routes.router)
app.include_router(student_routes.router)

@app.get("/")
def home():
    return {"message": "Welcome to Student API with Authentication"}

uvicorn app.main:app --reload