JWT autoryzacja FastAPI: kod krok po kroku

Implementacja uwierzytelniania JWT w FastAPI: kompletny przewodnik dla programistów

JWT (JSON Web Token) to zwarty token służący do przekazywania oświadczeń między stronami. Token składa się z trzech części zakodowanych w Base64: nagłówka, ładunku i podpisu, oddzielonych kropkami.

Nagłówek zawiera algorytm podpisu (np. HS256) i typ tokenu (JWT):

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

Ładunek przechowuje oświadczenia: sub (identyfikator użytkownika), iat (czas wystawienia), exp (czas wygaśnięcia):

{"sub": "alex", "name": "Alex Tomchok", "iat": 1516239022, "exp": 1516242622}

Podpis jest obliczany jako HMAC-SHA256 z (nagłówek.ładunek + klucz tajny). Serwer weryfikuje podpis podczas walidacji: każda zmiana ładunku unieważnia token. Dane są czytelne, ale chronione przed modyfikacją.

Tokeny dostępu mają krótki czas życia (15–30 minut), tokeny odświeżania — długi (dni/miesiące). W przykładzie używany jest tylko token dostępu.

Przygotowanie projektu i zależności

Utwórz strukturę:

fastapi-jwt-auth/
├── app/
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── schemas/
│   │   ├── token.py
│   │   └── user.py
│   ├── api/
│   │   └── v1/
│   │       ├── endpoints/
│   │       │   ├── auth.py
│   │       │   └── users.py
│   │       └── dependencies.py
│   └── services/
│       └── user_service.py
└── requirements.txt

requirements.txt:

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pydantic[email]==2.5.0

Zainstaluj: pip install -r requirements.txt.

Konfiguracja i schematy Pydantic

W core/config.py zdefiniuj ustawienia za pomocą Pydantic:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-super-secret-key-change-this-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

Schematy w schemas/user.py:

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool = True

    class Config:
        from_attributes = True

class UserInDB(User):
    hashed_password: str

schemas/token.py:

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: str | None = None

Bezpieczeństwo: haszowanie i operacje JWT

W core/security.py zaimplementuj haszowanie haseł (bcrypt) i funkcje JWT:

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    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_jwt

def decode_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

Serwis użytkowników i fałszywa baza danych

services/user_service.py z pamięciowym magazynem:

from app.schemas.user import UserInDB
from app.core.security import verify_password, get_password_hash

fake_users_db = {
    "alex": {
        "id": 1,
        "username": "alex",
        "email": "[email protected]",
        "hashed_password": get_password_hash("secret123"),
        "is_active": True,
    }
}

def get_user_by_username(username: str) -> UserInDB | None:
    if username in fake_users_db:
        return UserInDB(**fake_users_db[username])
    return None

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user_by_username(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

Zależności dla autoryzacji

W api/v1/dependencies.py użyj OAuth2PasswordBearer:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.security import decode_access_token
from app.services.user_service import get_user_by_username
from app.schemas.token import TokenData
from app.schemas.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Nie można zweryfikować poświadczeń",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user = get_user_by_username(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Nieaktywny użytkownik")
    return current_user

Endpointy: logowanie i chronione trasy

api/v1/endpoints/auth.py:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas.token import Token
from app.services.user_service import authenticate_user
from app.core.security import create_access_token

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nieprawidłowa nazwa użytkownika lub hasło",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token, token_type="bearer")

api/v1/endpoints/users.py:

from fastapi import APIRouter, Depends
from app.schemas.user import User
from app.api.v1.dependencies import get_current_active_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

Podłącz routery w main.py i uruchom uvicorn app.main:app.

Co jest ważne

• Ładunek JWT nie jest szyfrowany, tylko podpisywany — używaj tylko danych niewrażliwych.
• Przechowuj SECRET_KEY w zmiennych środowiskowych, generuj o długości 32+ znaków.
• W produkcji dodaj tokeny odświeżania i bazę danych (SQLAlchemy + Alembic).
• Zależności FastAPI umożliwiają ponowne wykorzystanie logiki autoryzacji bez duplikacji.
• Testuj endpointy przez Swagger UI: /docs.

— Editorial Team