File size: 2,087 Bytes
de1e3fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session
from ..db import get_db
from ..models import User
from ..models.base import UserRole
from ..schemas.auth import (
LoginRequest,
RegisterRequest,
TokenResponse,
UserOut,
)
from ..schemas.common import Message
from ..security import (
create_access_token,
get_current_user,
hash_password,
verify_password,
)
router = APIRouter(prefix="/auth", tags=["auth"])
@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
def register(payload: RegisterRequest, db: Session = Depends(get_db)) -> TokenResponse:
existing = db.execute(select(User).where(User.email == payload.email)).scalar_one_or_none()
if existing:
raise HTTPException(status.HTTP_409_CONFLICT, "Email already registered")
user = User(
name=payload.name,
email=payload.email,
phone=payload.phone,
zip=payload.zip,
password_hash=hash_password(payload.password),
role=UserRole.owner,
)
db.add(user)
db.commit()
db.refresh(user)
return TokenResponse(access_token=create_access_token(user.id))
@router.post("/login", response_model=TokenResponse)
def login(payload: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse:
user = db.execute(select(User).where(User.email == payload.email)).scalar_one_or_none()
if not user or not user.password_hash or not verify_password(payload.password, user.password_hash):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid email or password")
return TokenResponse(access_token=create_access_token(user.id))
@router.post("/logout", response_model=Message)
def logout() -> Message:
# Stateless JWT: client simply discards the token. Endpoint exists for symmetry (spec §12).
return Message(detail="Logged out")
@router.get("/me", response_model=UserOut)
def me(user: User = Depends(get_current_user)) -> User:
return user
|