diff --git a/backend/app/routers/auth.py b/backend/app/routers/auth.py new file mode 100644 index 0000000..fc71e96 --- /dev/null +++ b/backend/app/routers/auth.py @@ -0,0 +1,66 @@ +"""Authentication routes.""" +from datetime import timedelta +from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm +from sqlalchemy.orm import Session +from passlib.context import CryptContext + +from app.database import get_db +from app.models.user import User +from app.schemas.auth import Token, LoginRequest +from app.schemas.user import UserCreate, UserResponse +from app.config import settings + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") + +router = APIRouter(prefix="/auth", tags=["auth"]) + +def verify_password(plain_password: str, hashed_password: str) -> bool: + """Verify password.""" + return pwd_context.verify(plain_password, hashed_password) + +def get_password_hash(password: str) -> str: + """Hash password.""" + return pwd_context.hash(password) + +def create_access_token(data: dict, expires_delta: timedelta | None = None): + """Create access token.""" + from jose import jwt + to_encode = data.copy() + expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) + to_encode.update({"exp": expire}) + return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + +@router.post("/register", response_model=UserResponse) +def register(user: UserCreate, db: Session = Depends(get_db)): + """Register new user.""" + db_user = db.query(User).filter(User.email == user.email).first() + if db_user: + raise HTTPException(status_code=400, detail="Email already registered") + hashed_password = get_password_hash(user.password) + db_user = User( + email=user.email, + username=user.username, + hashed_password=hashed_password, + full_name=user.full_name + ) + db.add(db_user) + db.commit() + db.refresh(db_user) + return db_user + +@router.post("/login", response_model=Token) +def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)): + """Login user.""" + user = db.query(User).filter(User.username == form_data.username).first() + if not user or not verify_password(form_data.password, user.hashed_password): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Incorrect username or password" + ) + access_token = create_access_token( + data={"sub": str(user.id)}, + expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) + ) + return {"access_token": access_token, "token_type": "bearer"}