Создание API на FastAPI - IMI-DS

Привет! FastAPI — один из самых быстрых и современных фреймворков для создания API на Python. Он использует асинхронность, автоматическую документацию и Pydantic для валидации данных.

В этой статье мы создадим полноценное REST API для управления пользователями с использованием:

FastAPI и Uvicorn
Pydantic моделей
Depends для зависимостей
Аутентификации

Содержание

Что нужно знать перед началом
Основная часть
Настройка проекта
Модели данных
Заглушка базы данных
Аутентификация и JWT
CRUD эндпоинты
POST /users — регистрация
POST /token — вход в систему
GET /users/me — текущий пользователь
GET /users/{user_id} — получить пользователя по ID
GET /users — список пользователей (только для админа)
PUT /users/{user_id} — обновление пользователя
DELETE /users/{user_id} — удаление пользователя
Полный код
Задачи для закрепления
Нюансы и подводные камни
Частые ошибки и как их избежать
Ошибка 1: Забыл установить библиотеки
Ошибка 2: Неправильная обработка токена
Ошибка 3: Отсутствие проверки прав
Шпаргалка
Заключение
КВИЗ
Что дальше?

Что нужно знать перед началом

Для этого урока тебе понадобится:

Установленный FastAPI и Uvicorn
Базовое понимание Python и API

Совет: FastAPI автоматически генерирует документацию Swagger UI по адресу /docs.

Основная часть

Настройка проекта

Шаг 1. Установка зависимостей:

pip install fastapi uvicorn python-multipart

Шаг 2. Создание файла main.py:

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime, timedelta
import jwt
from passlib.context import CryptContext

app = FastAPI(title="User API", version="1.0.0")

Модели данных

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    password: str = Field(..., min_length=6)
    age: Optional[int] = Field(None, ge=0, le=120)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    age: Optional[int]
    created_at: datetime

class UserLogin(BaseModel):
    username: str
    password: str

Заглушка базы данных

users_db = {}
next_id = 1

# Хеширование паролей
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

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

Аутентификация и JWT

SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Неверный токен")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Неверный токен")
    
    user = users_db.get(username)
    if user is None:
        raise HTTPException(status_code=401, detail="Пользователь не найден")
    return user

CRUD эндпоинты

POST /users — регистрация

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    global next_id
    
    if user.username in users_db:
        raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
    
    hashed_password = hash_password(user.password)
    new_user = {
        "id": next_id,
        "username": user.username,
        "email": user.email,
        "password": hashed_password,
        "age": user.age,
        "created_at": datetime.utcnow()
    }
    users_db[user.username] = new_user
    next_id += 1
    
    return new_user

POST /token — вход в систему

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user:
        raise HTTPException(status_code=400, detail="Неверное имя пользователя или пароль")
    
    if not verify_password(form_data.password, user["password"]):
        raise HTTPException(status_code=400, detail="Неверное имя пользователя или пароль")
    
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

GET /users/me — текущий пользователь

@app.get("/users/me", response_model=UserResponse)
def get_current_user_info(current_user: dict = Depends(get_current_user)):
    return current_user

GET /users/{user_id} — получить пользователя по ID

@app.get("/users/{user_id}")
def get_user(user_id: int, current_user: dict = Depends(get_current_user)):
    for user in users_db.values():
        if user["id"] == user_id:
            return user
    raise HTTPException(status_code=404, detail="Пользователь не найден")

GET /users — список пользователей (только для админа)

def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    return current_user

@app.get("/users", dependencies=[Depends(require_admin)])
def get_users():
    return list(users_db.values())

PUT /users/{user_id} — обновление пользователя

@app.put("/users/{user_id}")
def update_user(user_id: int, user_update: UserCreate, current_user: dict = Depends(get_current_user)):
    user = None
    for u in users_db.values():
        if u["id"] == user_id:
            user = u
            break
    
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    
    if current_user["id"] != user_id and current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    
    if user_update.username != user["username"] and user_update.username in users_db:
        raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
    
    # Обновляем данные
    user["username"] = user_update.username
    user["email"] = user_update.email
    user["age"] = user_update.age
    if user_update.password:
        user["password"] = hash_password(user_update.password)
    
    return {"message": "Пользователь обновлён", "user": user}

DELETE /users/{user_id} — удаление пользователя

@app.delete("/users/{user_id}")
def delete_user(user_id: int, current_user: dict = Depends(get_current_user)):
    if current_user["id"] != user_id and current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    
    user = None
    username_to_delete = None
    for username, u in users_db.items():
        if u["id"] == user_id:
            user = u
            username_to_delete = username
            break
    
    if user is None:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    
    del users_db[username_to_delete]
    return {"message": f"Пользователь {user['username']} удалён"}

Полный код

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime, timedelta
import jwt
from passlib.context import CryptContext

app = FastAPI(title="User API", version="1.0.0")

# ========== Pydantic модели ==========
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    password: str = Field(..., min_length=6)
    age: Optional[int] = Field(None, ge=0, le=120)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    age: Optional[int]
    created_at: datetime

# ========== База данных ==========
users_db = {}
next_id = 1

# Хеширование
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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

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

# ========== JWT ==========
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Неверный токен")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Неверный токен")
    
    user = users_db.get(username)
    if user is None:
        raise HTTPException(status_code=401, detail="Пользователь не найден")
    return user

def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    return current_user

# ========== Эндпоинты ==========
@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    global next_id
    if user.username in users_db:
        raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
    
    new_user = {
        "id": next_id,
        "username": user.username,
        "email": user.email,
        "password": hash_password(user.password),
        "age": user.age,
        "created_at": datetime.utcnow()
    }
    users_db[user.username] = new_user
    next_id += 1
    return new_user

@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["password"]):
        raise HTTPException(status_code=400, detail="Неверные учётные данные")
    token = create_access_token(data={"sub": user["username"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/users/me", response_model=UserResponse)
def get_current_user_info(current_user: dict = Depends(get_current_user)):
    return current_user

@app.get("/users/{user_id}")
def get_user(user_id: int, current_user: dict = Depends(get_current_user)):
    for user in users_db.values():
        if user["id"] == user_id:
            return user
    raise HTTPException(status_code=404, detail="Пользователь не найден")

@app.get("/users", dependencies=[Depends(require_admin)])
def get_users():
    return list(users_db.values())

@app.put("/users/{user_id}")
def update_user(user_id: int, user_update: UserCreate, current_user: dict = Depends(get_current_user)):
    user = next((u for u in users_db.values() if u["id"] == user_id), None)
    if not user:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    
    if current_user["id"] != user_id and current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    
    if user_update.username != user["username"] and user_update.username in users_db:
        raise HTTPException(status_code=400, detail="Имя пользователя уже занято")
    
    user.update({
        "username": user_update.username,
        "email": user_update.email,
        "age": user_update.age,
        "password": hash_password(user_update.password) if user_update.password else user["password"]
    })
    return {"message": "Пользователь обновлён", "user": user}

@app.delete("/users/{user_id}")
def delete_user(user_id: int, current_user: dict = Depends(get_current_user)):
    if current_user["id"] != user_id and current_user["username"] != "admin":
        raise HTTPException(status_code=403, detail="Недостаточно прав")
    
    user = next((u for u in users_db.values() if u["id"] == user_id), None)
    if not user:
        raise HTTPException(status_code=404, detail="Пользователь не найден")
    
    username_to_delete = [u for u in users_db if users_db[u]["id"] == user_id][0]
    del users_db[username_to_delete]
    return {"message": f"Пользователь удалён"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Задачи для закрепления

Задача 1. Создай эндпоинт для получения списка всех пользователей.

Задача 2. Создай эндпоинт для обновления email пользователя.

Задача 3. Добавь пагинацию в эндпоинт GET /users.

Задача 4. Создай эндпоинт для изменения пароля.

Задача 5. Добавь роль «admin» и проверку прав.

Ответы:

Задача 1.

@app.get("/users")
def get_users():
    return list(users_db.values())

Задача 2.

@app.patch("/users/{user_id}/email")
def update_email(user_id: int, email: str):
    user = next((u for u in users_db.values() if u["id"] == user_id), None)
    if not user:
        raise HTTPException(status_code=404)
    user["email"] = email
    return {"message": "Email обновлён"}

Задача 3.

@app.get("/users")
def get_users(skip: int = 0, limit: int = 10):
    users = list(users_db.values())
    return users[skip:skip+limit]

Задача 4.

@app.put("/users/{user_id}/password")
def change_password(user_id: int, old_password: str, new_password: str):
    user = next((u for u in users_db.values() if u["id"] == user_id), None)
    if not user:
        raise HTTPException(status_code=404)
    if not verify_password(old_password, user["password"]):
        raise HTTPException(status_code=400, detail="Неверный пароль")
    user["password"] = hash_password(new_password)
    return {"message": "Пароль изменён"}

Задача 5.

def require_admin(current_user: dict = Depends(get_current_user)):
    if current_user["role"] != "admin":
        raise HTTPException(status_code=403)
    return current_user

Нюансы и подводные камни

FastAPI автоматически генерирует документацию по /docs и /redoc.
Pydantic модели валидируют данные автоматически.
Depends используется для внедрения зависимостей.
JWT-аутентификация требует библиотек python-jose и passlib.
Пароли хешируются с помощью passlib.
OAuth2PasswordBearer извлекает токен из заголовка.
Асинхронность позволяет обрабатывать больше запросов.

Частые ошибки и как их избежать

Ошибка 1: Забыл установить библиотеки

pip install python-jose[cryptography] passlib[bcrypt]

Ошибка 2: Неправильная обработка токена

Проверяй exp (время жизни) при декодировании.

Ошибка 3: Отсутствие проверки прав

Используй Depends(require_admin) для админских эндпоинтов.

Шпаргалка

Компонент	Пример
GET	`@app.get("/users")`
POST	`@app.post("/users")`
PUT	`@app.put("/users/{id}")`
DELETE	`@app.delete("/users/{id}")`
Depends	`Depends(get_current_user)`
Pydantic	`class User(BaseModel):`
JWT	`jwt.encode(...)`
Password	`passlib.context.CryptContext`