Archived
60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models, schemas
|
|
from ..database import get_db
|
|
from ..deps import get_current_user
|
|
from ..security import create_access_token, hash_password, verify_password
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/register", response_model=schemas.UserOut, status_code=status.HTTP_201_CREATED)
|
|
def register(payload: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
"""Регистрирует клиента; первого пользователя назначает администратором."""
|
|
email = payload.email.lower()
|
|
exists = db.scalar(select(models.User).where(models.User.email == email))
|
|
if exists:
|
|
raise HTTPException(status_code=409, detail="Пользователь уже существует")
|
|
|
|
# Роль первого пользователя определяется внутри транзакции и защищается
|
|
# уникальным email-ограничением от повторной регистрации одного адреса.
|
|
first_user = db.scalar(select(func.count(models.User.id))) == 0
|
|
user = models.User(
|
|
email=email,
|
|
password_hash=hash_password(payload.password),
|
|
role=models.UserRole.admin if first_user else models.UserRole.client,
|
|
)
|
|
db.add(user)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
raise HTTPException(status_code=409, detail="Пользователь уже существует")
|
|
db.refresh(user)
|
|
return user
|
|
|
|
|
|
@router.post("/login", response_model=schemas.TokenOut)
|
|
def login(
|
|
form_data: OAuth2PasswordRequestForm = Depends(),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Проверяет credentials и выдаёт Bearer JWT."""
|
|
email = form_data.username.lower()
|
|
user = db.scalar(select(models.User).where(models.User.email == email))
|
|
if user is None or not user.is_active or not verify_password(
|
|
form_data.password, user.password_hash
|
|
):
|
|
raise HTTPException(status_code=401, detail="Неверный email или пароль")
|
|
return schemas.TokenOut(access_token=create_access_token(str(user.id)))
|
|
|
|
|
|
@router.get("/me", response_model=schemas.UserOut)
|
|
def me(user: models.User = Depends(get_current_user)):
|
|
"""Возвращает профиль текущего пользователя без секретных данных."""
|
|
return user
|