Пользовательская система — регистрация, вход, JWT-аутентификация
В этой главе вы изучите схему JWT-аутентификации FastAPI и поймёте её отличия от сессионной аутентификации Django/Flask.
JWT против сессионной аутентификации
Django и Flask используютсессионную аутентификациюСервер хранит состояние пользователя, а в cookie хранится только Session ID.
FastAPI рекомендуетJWT-аутентификацию: После входа пользователь получает токен. Последующие запросы могут передавать токен для аутентификации.
| Особенности | Сессия（Django/Flask） | JWT（FastAPI） |
|---|---|---|
| Место хранения | На стороне сервера (память/база данных) | На стороне клиента (Cookie/LocalStorage) |
| Расширяемость | Серверу необходимо совместно использовать сессию | Без состояния, естественно поддерживает распределённые системы |
| Применимые сценарии | Веб-сайты с серверным рендерингом | API + SPA / мобильные приложения |
| Контроль срока действия | Сервер может аннулировать его в любое время | Действителен в течение всего срока действия токена |
Установите зависимости
(venv) $ pip install passlib python-jose python-multipart
- passlib: Хеширование паролей (аналогично werkzeug.security)
- python-jose: Генерация и проверка JWT
- python-multipart: Обработка отправки формы (форма входа OAuth2)
Определите модель User
Пример
# Путь к файлу: models.py, новые дополнения
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Хешировать пароль"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Проверка пароля"""
return pwd_context.verify(password, self.hashed_password)
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
hashed_password = Column(String(256), nullable=False)
def set_password(self, password: str):
"""Хешировать пароль"""
self.hashed_password = pwd_context.hash(password)
def verify_password(self, password: str) -> bool:
"""Проверка пароля"""
return pwd_context.verify(password, self.hashed_password)
Сгенерировать и выполнить миграции:
(venv) $ alembic revision --autogenerate -m "新增 User 模型" (venv) $ alembic upgrade head
Вспомогательные функции JWT
Пример
# Путь к файлу: auth.py (новый файл)
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Секретный ключ (в продакшене читается из переменных окружения)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Срок действия токена: 24 часа
# OAuth2PasswordBearer сообщает FastAPI: из заголовков fromRequest Authorization: Bearer <token> Mediummention извлекает JWT
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Сгенерировать JWT-токен"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | Отсутствует:
"""Проверить и декодировать JWT-токен"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
# Секретный ключ (в продакшене читается из переменных окружения)
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # Срок действия токена: 24 часа
# OAuth2PasswordBearer сообщает FastAPI: из заголовков fromRequest Authorization: Bearer <token> Mediummention извлекает JWT
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict) -> str:
"""Сгенерировать JWT-токен"""
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | Отсутствует:
"""Проверить и декодировать JWT-токен"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
Регистрация маршрута
Пример
# Путь к файлу: routers/users.py
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["User"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""Регистрация пользователя"""
# Проверить, существуют ли уже username и email
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=Электронная почта уже зарегистрирована)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import User
from schemas import UserCreate, UserResponse
from auth import create_access_token
router = APIRouter(prefix="/users", tags=["User"])
templates = Jinja2Templates(directory="templates")
@router.post("/register", response_model=UserResponse)
def register(
username: str = Form(...),
email: str = Form(...),
password: str = Form(...),
db: Session = Depends(get_db)
):
"""Регистрация пользователя"""
# Проверить, существуют ли уже username и email
if db.query(User).filter(User.username == username).first():
raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
if db.query(User).filter(User.email == email).first():
raise HTTPException(status_code=400, detail=Электронная почта уже зарегистрирована)
user = User(username=username, email=email)
user.set_password(password)
db.add(user)
db.commit()
db.refresh(user)
return user
Маршрут входа (выдача JWT)
Пример
# Добавить в файл: routers/users.py
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Войти и получить токен
Используйте OAuth2PasswordRequestForm: имена полей — username и password (формат формы).
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="Неверное имя пользователя или пароль",
headers={"WWW-Authenticate": "Bearer"}
)
# Выпуск JWT: содержит только user_id (не должен содержать конфиденциальную информацию)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
from fastapi.security import OAuth2PasswordRequestForm
from auth import create_access_token
@router.post("/token")
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
Войти и получить токен
Используйте OAuth2PasswordRequestForm: имена полей — username и password (формат формы).
"""
user = db.query(User).filter(User.username == form_data.username).first()
if not user or not user.verify_password(form_data.password):
raise HTTPException(
status_code=401,
detail="Неверное имя пользователя или пароль",
headers={"WWW-Authenticate": "Bearer"}
)
# Выпуск JWT: содержит только user_id (не должен содержать конфиденциальную информацию)
access_token = create_access_token(data={"sub": str(user.id)})
return {"access_token": access_token, "token_type": "bearer"}
В JWT-токене
subЭто стандартное поле subject, хранящее идентификатор пользователя. Сам токен не шифруется (только кодируется Base64), поэтомуНи в коем случаене помещайте в него конфиденциальную информацию, такую как пароли. Если вы хотите шифрование, используйте JWE вместо JWT.
Использование защиты JWT в маршрутах
Пример
# Функция зависимости для получения текущего пользователя
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Извлечь текущего пользователя из JWT-токена"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Токен недействителен или истёк")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=“Пользователь не существует”)
return user
# Внедрить в маршруты, требующие входа
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Получить информацию о текущем пользователе (требуется вход)"""
return current_user
from auth import oauth2_scheme, decode_access_token
def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
"""Извлечь текущего пользователя из JWT-токена"""
payload = decode_access_token(token)
if payload is None:
raise HTTPException(status_code=401, detail="Токен недействителен или истёк")
user_id = payload.get("sub")
user = db.query(User).filter(User.id == int(user_id)).first()
if user is None:
raise HTTPException(status_code=401, detail=“Пользователь не существует”)
return user
# Внедрить в маршруты, требующие входа
@router.get("/me", response_model=UserResponse)
def read_current_user(current_user: User = Depends(get_current_user)):
"""Получить информацию о текущем пользователе (требуется вход)"""
return current_user
Для SSR-страниц JWT обычно хранится в куки (а не в заголовке Authorization), и в шаблонных маршрутах токен извлекается из куки для проверки.
Итоги главы
В этой главе вы реализовали JWT-аутентификацию в FastAPI: passlib для хеширования паролей, python-jose для выпуска и проверки JWT, OAuth2PasswordBearer для извлечения токенов и Depends для внедрения зависимостей аутентификации.
В отличие от аутентификации на основе сессий в Django/Flask, JWT — это решение без сохранения состояния, более подходящее для сценариев разделения фронтенда и бэкенда.
Другие расширения