REST API на FastAPI: разработка с нуля, деплой на сервер

Что такое FastAPI и почему его выбирают?

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

В 2025 году FastAPI становится всё более популярным для создания микросервисов и высоконагруженных приложений. Он работает примерно в два раза быстрее Flask и предоставляет гораздо более удобный API для разработчика.

Установка зависимостей

Для создания полнофункционального REST API нам потребуются следующие пакеты:

pip install fastapi uvicorn sqlalchemy pydantic python-dotenv

Разберёмся, что нужно каждый пакет:

• fastapi — сам фреймворк
• uvicorn — ASGI сервер для запуска FastAPI приложений
• sqlalchemy — ORM для работы с базами данных
• pydantic — валидация данных (уже идёт с FastAPI)
• python-dotenv — загрузка переменных окружения

Создание первого FastAPI приложения

Начнём с простого "Hello World" и постепенно будем усложнять функциональность:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(
    title='My API',
    description='REST API на FastAPI',
    version='1.0.0'
)

# Модель данных для запросов
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    is_available: bool = True

# GET запрос
@app.get('/')
async def root():
    """Корневой endpoint"""
    return {'message': 'Добро пожаловать в API!'}

# GET с параметрами пути
@app.get('/items/{item_id}')
async def get_item(item_id: int):
    """Получить товар по ID"""
    return {'item_id': item_id, 'name': 'Пример товара'}

# POST запрос
@app.post('/items/')
async def create_item(item: Item):
    """Создать новый товар"""
    return {'created': item, 'status': 'success'}

# Запуск сервера
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

Запустите приложение:

python main.py

Автоматическая интерактивная документация будет доступна по адресу http://localhost:8000/docs

Работа с базой данных через SQLAlchemy

Для реальных приложений нужна база данных. Вот как организовать работу с SQLAlchemy:

Структура проекта

project/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
└── .env

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv(
    'DATABASE_URL',
    'sqlite:///./test.db'  # SQLite для разработки
)

engine = create_engine(
    DATABASE_URL,
    connect_args={'check_same_thread': False} if 'sqlite' in DATABASE_URL else {}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
    """Dependency для получения сессии БД"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base

class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String)
    price = Column(Float)
    is_available = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

schemas.py (Pydantic модели)

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    is_available: bool = True

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    is_available: Optional[bool] = None

class Item(ItemBase):
    id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True  # Для работы с ORM моделями

crud.py

from sqlalchemy.orm import Session
from models import Item as ItemModel
from schemas import ItemCreate, ItemUpdate

def get_items(db: Session, skip: int = 0, limit: int = 10):
    """Получить список товаров с пагинацией"""
    return db.query(ItemModel).offset(skip).limit(limit).all()

def get_item(db: Session, item_id: int):
    """Получить товар по ID"""
    return db.query(ItemModel).filter(ItemModel.id == item_id).first()

def create_item(db: Session, item: ItemCreate):
    """Создать новый товар"""
    db_item = ItemModel(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

def update_item(db: Session, item_id: int, item: ItemUpdate):
    """Обновить товар"""
    db_item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if db_item:
        update_data = item.dict(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_item, key, value)
        db.commit()
        db.refresh(db_item)
    return db_item

def delete_item(db: Session, item_id: int):
    """Удалить товар"""
    db_item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if db_item:
        db.delete(db_item)
        db.commit()
    return db_item

main.py

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from database import Base, engine, get_db
import models, schemas, crud

# Создаём таблицы
Base.metadata.create_all(bind=engine)

app = FastAPI(title='Items API')

@app.get('/items/', response_model=list[schemas.Item])
async def read_items(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db)
):
    """Получить список товаров"""
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

@app.get('/items/{item_id}', response_model=schemas.Item)
async def read_item(item_id: int, db: Session = Depends(get_db)):
    """Получить товар по ID"""
    item = crud.get_item(db, item_id=item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Товар не найден'
        )
    return item

@app.post('/items/', response_model=schemas.Item)
async def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    """Создать новый товар"""
    return crud.create_item(db=db, item=item)

@app.put('/items/{item_id}', response_model=schemas.Item)
async def update_item(
    item_id: int,
    item: schemas.ItemUpdate,
    db: Session = Depends(get_db)
):
    """Обновить товар"""
    db_item = crud.update_item(db, item_id=item_id, item=item)
    if not db_item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Товар не найден'
        )
    return db_item

@app.delete('/items/{item_id}')
async def delete_item(item_id: int, db: Session = Depends(get_db)):
    """Удалить товар"""
    item = crud.delete_item(db, item_id=item_id)
    if not item:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail='Товар не найден'
        )
    return {'status': 'Товар удалён'}

Аутентификация и авторизация

Для защиты API используйте JWT токены. Установите дополнительный пакет:

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

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET_KEY = 'your-secret-key-keep-it-safe'
ALGORITHM = 'HS256'

pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

def hash_password(password: str):
    """Хешировать пароль"""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str):
    """Проверить пароль"""
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Создать JWT токен"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(hours=24)

    to_encode.update({'exp': expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async 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)
    except JWTError:
        raise HTTPException(status_code=401)
    return username

@app.post('/token')
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """Получить токен доступа"""
    # Проверьте пользователя в БД
    access_token_expires = timedelta(hours=24)
    access_token = create_access_token(
        data={'sub': form_data.username},
        expires_delta=access_token_expires
    )
    return {'access_token': access_token, 'token_type': 'bearer'}

@app.get('/users/me')
async def read_users_me(current_user: str = Depends(get_current_user)):
    """Получить информацию о текущем пользователе"""
    return {'username': current_user}

Обработка ошибок и валидация

FastAPI автоматически валидирует данные через Pydantic и возвращает понятные ошибки:

from pydantic import BaseModel, Field, validator

class CreateItemRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    quantity: int = Field(default=0, ge=0)

    @validator('name')
    def name_must_contain_space(cls, v):
        if ' ' not in v:
            raise ValueError('Название должно содержать пробел')
        return v

@app.post('/items/validate')
async def create_item_with_validation(item: CreateItemRequest):
    """Создать товар с валидацией"""
    return {'created': item}

CORS и безопасность

Для работы с фронтенд приложениями нужно настроить CORS:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:3000', 'https://example.com'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

# Для разработки можно разрешить все источники
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

Логирование и мониторинг

Добавьте логирование для отладки и мониторинга:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.get('/items/')
async def read_items(db: Session = Depends(get_db)):
    """Получить список товаров"""
    logger.info('Запрос списка товаров')
    try:
        items = crud.get_items(db)
        logger.info(f'Возвращено {len(items)} товаров')
        return items
    except Exception as e:
        logger.error(f'Ошибка при получении товаров: {e}')
        raise HTTPException(status_code=500)

Деплой FastAPI приложения на сервер

Используя Gunicorn и Nginx

# Установка зависимостей на сервере
pip install gunicorn
pip install -r requirements.txt

# Запуск приложения через Gunicorn
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Конфигурация Nginx

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Используя Docker

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t fastapi-app .
docker run -p 8000:8000 fastapi-app

Тестирование FastAPI приложений

pip install pytest httpx

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_items():
    response = client.get('/items/')
    assert response.status_code == 200
    assert isinstance(response.json(), list)

def test_create_item():
    response = client.post('/items/', json={
        'name': 'Test Item',
        'price': 100.0
    })
    assert response.status_code == 200
    assert response.json()['name'] == 'Test Item'

# Запуск тестов
pytest

Лучшие практики FastAPI разработки

• Используйте type hints для автоматической валидации и документации
• Разделяйте код на модули (models, schemas, crud, main)
• Всегда возвращайте понятные HTTP статус коды
• Документируйте endpoints с помощью docstring'ов
• Используйте Depends для внедрения зависимостей
• Не логируйте чувствительные данные
• Используйте переменные окружения для конфигурации
• Пишите тесты для всех критических функций
• Используйте асинхронные функции для I/O операций

Заключение

FastAPI — это мощный и современный фреймворк для создания REST API на Python. Он обеспечивает высокую производительность, удобную разработку и автоматическую генерацию документации. В этом гайде мы рассмотрели основы создания полнофункционального API с базой данных, аутентификацией и развёртыванием на продакшене.

Ключевые поисковые запросы: FastAPI REST API, создание API на Python, FastAPI с SQLAlchemy, JWT аутентификация, Pydantic валидация, FastAPI деплой, микросервисы на FastAPI, асинхронный Python.