JWT авторизация в FastAPI: от теории до работающего кода

При разработке современных веб-приложений и API вопрос безопасности и аутентификации пользователей встаёт одним из первых. Как сделать так, чтобы пользователь мог войти один раз и получать доступ к защищённым ресурсам без постоянного ввода пароля? Как организовать систему, которая легко масштабируется и не требует хранения состояния сессии на сервере?

В этой статье я разберу подход, основанный на JWT (JSON Web Tokens), и покажу, как реализовать полноценную авторизацию в FastAPI — одном из самых быстрых и современных фреймворков для Python. Мы пройдём путь от архитектуры приложения до готового кода, который можно использовать в реальных проектах.

JWT (JSON Web Token) — это компактный и самодостаточный способ передачи информации между сторонами в виде JSON-объекта. Токен подписан цифровой подписью, что гарантирует его подлинность.

Токен состоит из трёх частей, разделённых точками:

textheader.payload.signature

Header (заголовок) — содержит тип токена и алгоритм подписи:

json{"alg":"HS256","typ":"JWT"}

Payload (полезная нагрузка) — содержит утверждения (claims): информация о пользователе, время выпуска, срок действия и другие данные:

json{"sub":"user_id_123","exp":1735689600,"iat":1735603200}

Signature (подпись) — создаётся путём шифрования header и payload с секретным ключом.

В реальных проектах редко ограничиваются одним токеном. Обычно используют пару:

Этот подход повышает безопасность: если access токен скомпрометирован, он будет действителен недолго, а refresh токен можно хранить в защищённом месте (например, HttpOnly cookie).

Прежде чем переходить к коду, важно понять, как организовать проект. Грамотное разделение на слои делает код поддерживаемым и тестируемым.

textproject/├── app/│ ├── core/│ │ ├── config.py # Настройки приложения│ │ ├── security.py # Функции для работы с JWT│ │ └── dependencies.py # Зависимости FastAPI│ ├── models/│ │ └── user.py # Модели базы данных│ ├── schemas/│ │ ├── user.py # Pydantic-схемы для пользователя│ │ └── token.py # Pydantic-схемы для токенов│ ├── routers/│ │ ├── auth.py # Эндпоинты аутентификации│ │ └── users.py # Защищённые эндпоинты пользователя│ ├── services/│ │ └── user.py # Бизнес-логика│ └── main.py # Точка входа├── requirements.txt└── .env

Зачем такое разделение?

Начнём с файла core/config.py, где будут храниться чувствительные настройки:

pythonfrom pydantic_settings import BaseSettingsclassSettings(BaseSettings): SECRET_KEY:str="your-secret-key-change-in-production" ALGORITHM:str="HS256" ACCESS_TOKEN_EXPIRE_MINUTES:int=30 REFRESH_TOKEN_EXPIRE_DAYS:int=7classConfig: env_file =".env"settings = Settings()

В core/security.py создадим функции создания и проверки токенов:

pythonfrom datetime import datetime, timedeltafrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom.config import settingspwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")defverify_password(plain_password:str, hashed_password:str)->bool:"""Проверка пароля"""return pwd_context.verify(plain_password, hashed_password)defget_password_hash(password:str)->str:"""Хэширование пароля"""return pwd_context.hash(password)defcreate_access_token(data:dict, expires_delta: timedelta =None)->str:"""Создание access токена""" to_encode = data.copy()if expires_delta: expire = datetime.utcnow()+ expires_deltaelse: expire = datetime.utcnow()+ timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)return encoded_jwtdefcreate_refresh_token(data:dict)->str:"""Создание refresh токена""" to_encode = data.copy() expire = datetime.utcnow()+ timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)return encoded_jwtdefdecode_token(token:str)->dict:"""Декодирование токена"""try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])return payloadexcept JWTError:returnNone

Определим схемы для входящих и исходящих данных.

schemas/user.py:

pythonfrom pydantic import BaseModel, EmailStrclassUserCreate(BaseModel): email: EmailStr password:str full_name:strclassUserResponse(BaseModel):id:int email: EmailStr full_name:strclassConfig: from_attributes =True

schemas/token.py:

pythonfrom pydantic import BaseModelclassToken(BaseModel): access_token:str refresh_token:str token_type:str="bearer"classTokenData(BaseModel): user_id:int

Это ключевой компонент, который будет использоваться для защиты эндпоинтов.

core/dependencies.py:

pythonfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom jose import JWTErrorfrom sqlalchemy.orm import Sessionfrom..database import get_dbfrom..models.user import Userfrom..schemas.token import TokenDatafrom.security import decode_tokenoauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")asyncdefget_current_user( token:str= Depends(oauth2_scheme), db: Session = Depends(get_db))-> User:""" Зависимость, которая извлекает текущего пользователя из токена. Используется для защиты эндпоинтов. """ credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate":"Bearer"},) payload = decode_token(token)if payload isNone:raise credentials_exception user_id = payload.get("sub")if user_id isNone:raise credentials_exception token_data = TokenData(user_id=int(user_id)) user = db.query(User).filter(User.id== token_data.user_id).first()if user isNone:raise credentials_exceptionreturn user

routers/auth.py:

pythonfrom fastapi import APIRouter, Depends, HTTPException, statusfrom sqlalchemy.orm import Sessionfrom datetime import timedeltafrom..database import get_dbfrom..models.user import Userfrom..schemas.user import UserCreate, UserResponsefrom..schemas.token import Tokenfrom..core.security import( verify_password, create_access_token, create_refresh_token, get_password_hash)from..core.config import settingsrouter = APIRouter(prefix="/api/auth", tags=["authentication"])@router.post("/register", response_model=UserResponse)defregister(user_data: UserCreate, db: Session = Depends(get_db)):"""Регистрация нового пользователя""" existing_user = db.query(User).filter(User.email == user_data.email).first()if existing_user:raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered") hashed_password = get_password_hash(user_data.password) new_user = User( email=user_data.email, hashed_password=hashed_password, full_name=user_data.full_name) db.add(new_user) db.commit() db.refresh(new_user)return new_user@router.post("/login", response_model=Token)deflogin( form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):""" Аутентификация пользователя. Возвращает access и refresh токены. """ user = db.query(User).filter(User.email == form_data.username).first()ifnot user ornot verify_password(form_data.password, user.hashed_password):raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate":"Bearer"},) access_token = create_access_token(data={"sub":str(user.id)}) refresh_token = create_refresh_token(data={"sub":str(user.id)})return{"access_token": access_token,"refresh_token": refresh_token,"token_type":"bearer"}@router.post("/refresh", response_model=Token)defrefresh_token( refresh_token:str, db: Session = Depends(get_db)):"""Обновление access токена с помощью refresh токена""" payload = decode_token(refresh_token)if payload isNone:raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token") user_id = payload.get("sub")if user_id isNone:raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token") user = db.query(User).filter(User.id==int(user_id)).first()if user isNone:raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found") new_access_token = create_access_token(data={"sub":str(user.id)})return{"access_token": new_access_token,"refresh_token": refresh_token,"token_type":"bearer"}

routers/users.py:

pythonfrom fastapi import APIRouter, Dependsfrom..schemas.user import UserResponsefrom..core.dependencies import get_current_userfrom..models.user import Userrouter = APIRouter(prefix="/api/users", tags=["users"])@router.get("/me", response_model=UserResponse)defget_current_user_info(current_user: User = Depends(get_current_user)):""" Получение информации о текущем пользователе. Эндпоинт защищён: требуется валидный access токен. """return current_user

main.py:

pythonfrom fastapi import FastAPIfrom.routers import auth, usersapp = FastAPI(title="FastAPI JWT Auth", version="1.0.0")app.include_router(auth.router)app.include_router(users.router)@app.get("/")defroot():return{"message":"Welcome to FastAPI JWT Auth API"}

Чтобы наглядно представить, как все компоненты взаимодействуют, ниже приведена схема двух основных процессов.

После запуска приложения (uvicorn app.main:app --reload) документация будет доступна по адресу http://localhost:8000/docs.

Проверка работы:

При развёртывании приложения обязательно:

Мы разобрали, как устроена JWT-авторизация, начиная от теории и заканчивая практической реализацией на FastAPI. Ключевые выводы: