bonamin
5dbb775308
Token creation/decoding/blacklisting was split across auth.py and deps.py causing a circular import. Consolidate make_token, decode_token, and blacklist_token in deps.py; auth.py now imports from there. Also switches /login to accept JSON body (username+pin) instead of form-encoded, and returns a proper user object in the response. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
import bcrypt
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database import get_db
|
|
from models.user import User
|
|
from schemas.auth import LoginRequest, TokenResponse
|
|
from schemas.user import UserOut
|
|
from routers.deps import get_current_user, make_token, decode_token, blacklist_token
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(body: LoginRequest, db: Session = Depends(get_db)):
|
|
user = db.query(User).filter(User.username == body.username, User.is_active == True).first()
|
|
if not user or not bcrypt.checkpw(body.pin.encode(), user.pin_hash.encode()):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
|
|
token = make_token(user)
|
|
return TokenResponse(access_token=token, user=UserOut.model_validate(user))
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
def refresh(token: str, db: Session = Depends(get_db)):
|
|
payload = decode_token(token)
|
|
user = db.query(User).filter(User.id == int(payload["sub"]), User.is_active == True).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
blacklist_token(token)
|
|
new_token = make_token(user)
|
|
return TokenResponse(access_token=new_token, user=UserOut.model_validate(user))
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(token: str):
|
|
blacklist_token(token)
|
|
return {"status": "logged out"}
|
|
|
|
|
|
@router.get("/me", response_model=UserOut)
|
|
def me(user: User = Depends(get_current_user)):
|
|
return user
|