- ВКонтакте
- РћРТвЂВВВВВВВВнокласснРСвЂВВВВВВВВРєРСвЂВВВВВВВВ
- РњРѕР№ Р В Р’В Р РЋРЎв„ўР В Р’В Р РЋРІР‚ВВВВВВВВРЎР‚
- Viber
- Skype
- Telegram
Не работает авторизация в swagger
Пишу апи на fastapi
для пет проекта, проблема следующая - тестирую в swagger авторизацию и через post запросы: регистрация, авторизация работает. Затем, я добавил работу с JWT
, а также инъекцию зависимостей (вроде так называется) и опять же через api работает. Проблемы начинаются когда я пытаюсь авторизироваться через swagger, через кнопку снизу:
Там просит ввести логин и пароль. Проблема в том, что логин и пароль пользователей из бд не подходят. Вместо этого выдаётся ошибка:
Попробовал сменить эндпоинт с /auth/login на просто /login и теперь выдаёт ошибку:
Нейронки говорят что вроде как валидация не проходит (но через запросы же идёт). Далее код
dependencies.py:
from fastapi import HTTPException, Security
from fastapi.security import OAuth2PasswordBearer
from utils.auth import verify_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")
# проверка на вшивость (авторизован чел или нет)
def get_current_user(token: str = Security(oauth2_scheme)):
# получение текущего пользователя на основе JWT
payload = verify_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Invalid credentials")
return payload
auth.py:
from datetime import datetime, timedelta
from jose import JWTError, jwt
from dotenv import load_dotenv
import os
load_dotenv()
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE = os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES")
# создание JWT токена
def create_access_token(data: dict, expires_delta: timedelta = None):
"""Create JWT token"""
to_encode = data.copy()
if expires_delta:
expire = datetime.now() + expires_delta
else:
expire = datetime.now() + timedelta(minutes=ACCESS_TOKEN_EXPIRE)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# проверка JWT токена
def verify_token(token: str):
"""Check JWT token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
auth_router.py:
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from database.session import SessionLocal
from database.crud import create_user, get_user_by_username
from schemas.user_scheme import UserCreate, UserLogin
from utils.security import hash_password, verify_password
from utils.auth import create_access_token
from datetime import timedelta
router = APIRouter()
async def get_session():
async with SessionLocal() as session:
yield session
@router.post("/register", response_model=UserCreate, tags=["Authentication"])
async def register_user(user: UserCreate,
session: AsyncSession = Depends(get_session)):
existing_user = await get_user_by_username(session, user.username)
if existing_user:
raise HTTPException(status_code=400,
detail="Username already registered")
hashed_password = hash_password(user.hashed_password)
new_user = await create_user(session,
user.username,
hashed_password,
user.email)
return new_user
@router.post("/login", tags=["Authentication"])
async def login_user(user: UserLogin,
session: AsyncSession = Depends(get_session)):
db_user = await get_user_by_username(session, user.username)
if not db_user:
raise HTTPException(status_code=400, detail="User does not exist")
if not verify_password(user.password, db_user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect password")
# JWT
access_token_expires = timedelta(minutes=30)
access_token = create_access_token(
data={"sub": db_user.username},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
user_scheme:
from pydantic import BaseModel, EmailStr
# база для внесения в бд
class UserBase(BaseModel):
username: str
email: EmailStr
# создание пользователя /crud.py create_user
class UserCreate(UserBase):
hashed_password: str | int
# обновление данных пользователя /crud.py update_user
class UserUpdate(UserBase):
hashed_password: str | int
# удаление пользователя по id /crud.py delete_user
class UserDelete(BaseModel):
detail: str
# вход в аккаунт /auth_router.py login_user
class UserLogin(BaseModel):
username: str
password: str
# модель для ответа на запросы по
# получению всех пользователей /crud_router.py get_users
class User(UserBase):
id: int
class Config:
orm_mode = True
ну и общая структура файлов:
│ database.db
│ main.py
│ __init__.py
│
├───database
│ │ crud.py
│ │ session.py
│ │ __init__.py
│ │
│ └───__pycache__
│ crud.cpython-311.pyc
│ session.cpython-311.pyc
│ __init__.cpython-311.pyc
│
├───models
│ │ base.py
│ │ portfolio.py
│ │ __init__.py
│ │
│ └───__pycache__
│ base.cpython-311.pyc
│ portfolio.cpython-311.pyc
│ __init__.cpython-311.pyc
│
├───routers
│ │ auth_router.py
│ │ crud_router.py
│ │ index_router.py
│ │ __init__.py
│ │
│ └───__pycache__
│ auth_router.cpython-311.pyc
│ crud_router.cpython-311.pyc
│ endpoints.cpython-311.pyc
│ index_router.cpython-311.pyc
│ __init__.cpython-311.pyc
│
├───schemas
│ │ user_scheme.py
│ │ __init__.py
│ │
│ └───__pycache__
│ user_scheme.cpython-311.pyc
│ __init__.cpython-311.pyc
│
├───static
│ .gitkeep
│
├───templates
│ │ .gitkeep
│ │
│ └───authorisation
│ autorisation.html
│
├───tests
│ test_jwt.py
│
├───utils
│ │ auth.py
│ │ dependencies.py
│ │ security.py
│ │ __init__.py
│ │
│ └───__pycache__
│ auth.cpython-311.pyc
│ dependencies.cpython-311.pyc
│ security.cpython-311.pyc
│ __init__.cpython-311.pyc
│
└───__pycache__
main.cpython-311.pyc
__init__.cpython-311.pyc
Ответы (1 шт):
Проблема оказалась в том что эндпоинт /login
принимал запрос в формате JSON
, а у меня судя по всему должен приниматься в формате application/x-www-form-urlencoded
. Решил проблему с помощью замены в эндпоинте /login
формата с user
, на Form
из fastapi
для логина и пароля.
Было:
@router.post("/login", tags=["Authentication"])
async def login_user(user: UserLogin,
session: AsyncSession = Depends(get_session)):
db_user = await get_user_by_username(session, user.username)
if not db_user:
raise HTTPException(status_code=400, detail="User does not exist")
if not verify_password(user.password, db_user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect password")
# JWT
access_token_expires = timedelta(minutes=30)
access_token = create_access_token(
data={"sub": db_user.username},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
Стало:
@router.post("/login", tags=["Authentication"])
async def login_user(username: str = Form(), # здесь замена на Form
password: str = Form(), # здесь замена на Form
session: AsyncSession = Depends(get_session)):
db_user = await get_user_by_username(session, username)
if not db_user:
raise HTTPException(status_code=400, detail="User does not exist")
if not verify_password(password, db_user.hashed_password):
raise HTTPException(status_code=400, detail="Incorrect password")
# JWT
access_token_expires = timedelta(minutes=30)
access_token = create_access_token(
data={"sub": db_user.username},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}