update

common/auth.py

@@ -17,12 +17,11 @@ USERS = [
     {"username": "user", "password": "user123"},
 ]
 
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login/token")
 
 class LoginRequest(BaseModel):
     username: str
     password: str
-    grant_type: str
 
 def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
     to_encode = data.copy()

routes/auth.py

@@ -1,32 +1,34 @@
 from typing import Optional
-from fastapi import APIRouter, Form, HTTPException
+from fastapi import APIRouter, Body, Form, HTTPException
 from datetime import timedelta
 import common.auth as auth
 
 router = APIRouter(prefix="/auth", tags=["Auth"])
 
-@router.post("/login")
-async def login(username: Optional[str] = Form(default=None),
-                password: Optional[str] = Form(default=None),
-                request: Optional[auth.LoginRequest] = None):
-    
-    # Если данные пришли через Form Data
-    if username is not None and password is not None:
-        final_username = username
-        final_password = password
-    # Если данные пришли через JSON
-    elif request is not None:
-        final_username = request.username
-        final_password = request.password
-    else:
-        raise HTTPException(status_code=400, detail="Не указаны логин и пароль")
-
-    user = next((u for u in auth.USERS if u["username"] == final_username), None)
-    if not user or user["password"] != final_password:
+def authenticate_user(username: str, password: str):
+    """Проверяет, существует ли пользователь и правильный ли пароль."""
+    user = next((u for u in auth.USERS if u["username"] == username), None)
+    if not user or user["password"] != password:
         raise HTTPException(status_code=401, detail="Неверный логин или пароль")
-    
+    return user
+
+def generate_access_token(username: str):
+    """Генерирует токен для аутентифицированного пользователя."""
     access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = auth.create_access_token(
-        data={"sub": user["username"]}, expires_delta=access_token_expires
-    )
-    return {"access_token": access_token, "token_type": "bearer"}
+    return auth.create_access_token(data={"sub": username}, expires_delta=access_token_expires)
+
+async def login_common(username: str, password: str):
+    """Общий метод аутентификации."""
+    user = authenticate_user(username, password)
+    access_token = generate_access_token(user["username"])
+    return {"access_token": access_token, "token_type": "bearer"}
+
+@router.post("/login", summary="Авторизация через JSON")
+async def login_json(request: auth.LoginRequest = Body(...)):
+    """Принимает JSON-запросы."""
+    return await login_common(request.username, request.password)
+
+@router.post("/login/token", summary="Авторизация через Form-Data")
+async def login_form(username: str = Form(...), password: str = Form(...)):
+    """Принимает Form-Data (x-www-form-urlencoded)."""
+    return await login_common(username, password)