我正在为一个宠物项目编写 API fastapi
,问题如下 - 我正在 swagger 中并通过 post 请求测试授权:注册、授权工作。然后,我添加了工作JWT
,以及依赖注入(我认为这就是它的名字)并且它再次通过 api 工作。当我尝试通过下面的按钮通过 swagger 登录时,问题就开始了:
它要求输入登录名和密码。问题在于数据库中的用户登录名和密码不匹配。相反,会返回一个错误:
我尝试将端点从 /auth/login 更改为 /login,现在收到错误:
神经网络表示验证似乎没有通过(但它确实通过了请求)。接下来是代码
依赖项.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
身份验证.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
身份验证路由器.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"}
用户方案:
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