FastAPI: создаем высокопроизводительные REST API на Python

# Создание виртуального окружения
python -m venv venv

# Активация на Windows
venv\Scripts\activate

# Активация на Unix/MacOS
source venv/bin/activate

pip install fastapi uvicorn

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
return {"message": "Hello World"}

uvicorn main:app --reload

FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

project/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── routes/
│ │ ├── __init__.py
│ │ └── items.py
│ └── models/
│ ├── __init__.py
│ └── item.py
└── requirements.txt

from fastapi import APIRouter, HTTPException, Path, Query
from typing import List, Optional

# Временное хранилище для демонстрации
items_db = {
1: {"name": "Hammer", "price": 9.99, "count": 20, "id": 1, "description": "For hammering nails"},
2: {"name": "Screwdriver", "price": 4.99, "count": 15, "id": 2, "description": "For turning screws"}
}

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

# GET запрос для получения всех элементов
@router.get("/", response_model=List[dict])
async def get_items(skip: int = 0, limit: int = 10):
return list(items_db.values())[skip: skip + limit]

# GET запрос для получения конкретного элемента по ID
@router.get("/{item_id}", response_model=dict)
async def get_item(item_id: int = Path(..., title="ID элемента", ge=1)):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")
return items_db[item_id]

# POST запрос для создания нового элемента
@router.post("/", response_model=dict, status_code=201)
async def create_item(item: dict):
if any(i["name"] == item["name"] for i in items_db.values()):
raise HTTPException(status_code=400, detail="Элемент с таким именем уже существует")

# Генерируем новый ID
new_id = max(items_db.keys(), default=0) + 1
item["id"] = new_id
items_db[new_id] = item
return item

# PUT запрос для обновления существующего элемента
@router.put("/{item_id}", response_model=dict)
async def update_item(item: dict, item_id: int = Path(..., title="ID элемента для обновления", ge=1)):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")

# Обновляем данные, сохраняя ID
item["id"] = item_id
items_db[item_id] = item
return item

# DELETE запрос для удаления элемента
@router.delete("/{item_id}")
async def delete_item(item_id: int = Path(..., title="ID элемента для удаления", ge=1)):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")

deleted_item = items_db.pop(item_id)
return {"message": f"Элемент {deleted_item['name']} удален"}

from fastapi import FastAPI
from app.routes import items

app = FastAPI(
title="Inventory API",
description="API для управления инвентарем",
version="0.1.0"
)

# Подключаем маршруты
app.include_router(items.router)

@app.get("/")
async def root():
return {"message": "Добро пожаловать в Inventory API! Перейдите на /docs для документации."}

from pydantic import BaseModel, Field, validator
from typing import Optional
import re

class ItemBase(BaseModel):
name: str = Field(..., min_length=1, max_length=50, description="Название товара")
description: Optional[str] = Field(None, max_length=500, description="Описание товара")
price: float = Field(..., gt=0, description="Цена товара (должна быть положительной)")
count: int = Field(..., ge=0, description="Количество товаров на складе")

@validator('name')
def name_must_be_valid(cls, v):
if not re.match(r'^[a-zA-Z0-9 \-_]+$', v):
raise ValueError('Название может содержать только буквы, цифры, пробелы, дефисы и подчеркивания')
return v

@validator('price')
def price_must_be_realistic(cls, v):
if v > 1000000:
raise ValueError('Цена кажется нереалистично высокой')
return round(v, 2) # Округляем до 2 знаков после запятой

class ItemCreate(ItemBase):
# Дополнительные поля или валидации для создания товара
pass

class ItemUpdate(ItemBase):
# Все поля необязательны при обновлении
name: Optional[str] = Field(None, min_length=1, max_length=50)
price: Optional[float] = Field(None, gt=0)
count: Optional[int] = Field(None, ge=0)

class ItemResponse(ItemBase):
id: int = Field(..., description="Уникальный идентификатор товара")

class Config:
orm_mode = True # Позволяет использовать ORM-объекты напрямую

from fastapi import APIRouter, HTTPException, Path, Query, Depends
from typing import List, Optional
from app.models.item import ItemCreate, ItemUpdate, ItemResponse

# Временное хранилище
items_db = {
1: {"name": "Hammer", "price": 9.99, "count": 20, "id": 1, "description": "For hammering nails"},
2: {"name": "Screwdriver", "price": 4.99, "count": 15, "id": 2, "description": "For turning screws"}
}

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

@router.get("/", response_model=List[ItemResponse])
async def get_items(skip: int = 0, limit: int = 10):
return list(items_db.values())[skip: skip + limit]

@router.get("/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int = Path(..., title="ID элемента", ge=1)):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")
return items_db[item_id]

@router.post("/", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
if any(i["name"] == item.name for i in items_db.values()):
raise HTTPException(status_code=400, detail="Элемент с таким именем уже существует")

# Преобразуем Pydantic модель в словарь и добавляем ID
new_id = max(items_db.keys(), default=0) + 1
item_dict = item.dict()
item_dict["id"] = new_id

items_db[new_id] = item_dict
return item_dict

@router.put("/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int = Path(..., ge=1), item_update: ItemUpdate):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")

# Получаем текущий элемент
current_item = items_db[item_id]

# Обновляем только предоставленные поля
update_data = item_update.dict(exclude_unset=True)
updated_item = {**current_item, **update_data}

items_db[item_id] = updated_item
return updated_item

@router.delete("/{item_id}")
async def delete_item(item_id: int = Path(..., ge=1)):
if item_id not in items_db:
raise HTTPException(status_code=404, detail="Элемент не найден")

deleted_item = items_db.pop(item_id)
return {"message": f"Элемент {deleted_item['name']} удален"}

pip install sqlalchemy databases psycopg2-binary

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

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# Для PostgreSQL:
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

from sqlalchemy import Column, Integer, String, Float, Text
from app.database import Base

class Item(Base):
__tablename__ = "items"

id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, index=True)
description = Column(Text, nullable=True)
price = Column(Float)
count = Column(Integer)

from fastapi import FastAPI
from app.routes import items
from app.database import engine
from app.models import db_models

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

app = FastAPI(
title="Inventory API",
description="API для управления инвентарем",
version="0.1.0"
)

app.include_router(items.router)

@app.get("/")
async def root():
return {"message": "Добро пожаловать в Inventory API! Перейдите на /docs для документации."}

from fastapi import APIRouter, HTTPException, Path, Query, Depends
from sqlalchemy.orm import Session
from typing import List, Optional

from app.models.item import ItemCreate, ItemUpdate, ItemResponse
from app.models.db_models import Item
from app.database import get_db

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

@router.get("/", response_model=List[ItemResponse])
def get_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = db.query(Item).offset(skip).limit(limit).all()
return items

@router.get("/{item_id}", response_model=ItemResponse)
def get_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Элемент не найден")
return item

@router.post("/", response_model=ItemResponse, status_code=201)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
# Проверка на существующее имя
existing_item = db.query(Item).filter(Item.name == item.name).first()
if existing_item:
raise HTTPException(status_code=400, detail="Элемент с таким именем уже существует")

# Создаем модель SQLAlchemy из Pydantic модели
db_item = Item(**item.dict())

# Добавляем в базу данных
db.add(db_item)
db.commit()
db.refresh(db_item)

return db_item

@router.put("/{item_id}", response_model=ItemResponse)
def update_item(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)):
# Получаем существующий элемент
db_item = db.query(Item).filter(Item.id == item_id).first()
if db_item is None:
raise HTTPException(status_code=404, detail="Элемент не найден")

# Обновляем только предоставленные поля
item_data = item.dict(exclude_unset=True)
for key, value in item_data.items():
setattr(db_item, key, value)

db.commit()
db.refresh(db_item)

return db_item

@router.delete("/{item_id}")
def delete_item(item_id: int, db: Session = Depends(get_db)):
db_item = db.query(Item).filter(Item.id == item_id).first()
if db_item is None:
raise HTTPException(status_code=404, detail="Элемент не найден")

# Сохраняем имя перед удалением
item_name = db_item.name

# Удаляем из базы данных
db.delete(db_item)
db.commit()

return {"message": f"Элемент {item_name} удален"}

@router.get("/inventory-status", response_model=List[InventoryStatusReport])
async def get_inventory_status(db: Session = Depends(get_db)):
# Запрос к представлению вместо соединения таблиц в ORM
result = db.execute("SELECT * FROM inventory_status_view").fetchall()
return result

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

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
from pydantic import BaseModel
from typing import Optional

# Настройки безопасности
SECRET_KEY = "YOUR_SECRET_KEY" # В реальных проектах используйте переменные окружения
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Модели для аутентификации
class Token(BaseModel):
access_token: str
token_type: str

class TokenData(BaseModel):
username: Optional[str] = None

class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None

class UserInDB(User):
hashed_password: str

# Имитация БД пользователей (в реальном приложении используйте базу данных)
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "john@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"disabled": False,
}
}

# Инициализация контекста для хэширования паролей
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
return pwd_context.hash(password)

def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user

def create_access_token(data: dict, expires_delta: Optional[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

async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Невозможно проверить учетные данные",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Неактивный пользователь")
return current_user

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from datetime import timedelta

from app.security import (
Token, User, authenticate_user, create_access_token, 
get_current_active_user, ACCESS_TOKEN_EXPIRE_MINUTES, fake_users_db
)

router = APIRouter(tags=["authentication"])

@router.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Неверное имя пользователя или пароль",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}

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

from fastapi import FastAPI
from app.routes import items, auth
from app.database import engine
from app.models import db_models

db_models.Base.metadata.create_all(bind=engine)

app = FastAPI(
title="Inventory API",
description="API для управления инвентарем с поддержкой аутентификации",
version="0.1.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)

app.include_router(auth.router)
app.include_router(items.router)

@app.get("/")
async def root():
return {"message": "Добро пожаловать в Inventory API! Перейдите на /docs для документации."}

from fastapi import APIRouter, HTTPException, Path, Query, Depends
from sqlalchemy.orm import Session
from typing import List, Optional

from app.models.item import ItemCreate, ItemUpdate, ItemResponse
from app.models.db_models import Item
from app.database import get_db
from app.security import get_current_active_user, User

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

@router.get("/", response_model=List[ItemResponse])
def get_items(
skip: int = 0, 
limit: int = 100, 
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user)
):
items = db.query(Item).offset(skip).limit(limit).all()
return items

# Остальные методы также обновляются с добавлением:
# current_user: User = Depends(get_current_active_user)

from fastapi import APIRouter, HTTPException, Path, Query, Depends, Body
from sqlalchemy.orm import Session
from typing import List, Optional

from app.models.item import ItemCreate, ItemUpdate, ItemResponse
from app.models.db_models import Item
from app.database import get_db
from app.security import get_current_active_user, User

router = APIRouter(
prefix="/items", 
tags=["items"],
responses={
404: {"description": "Элемент не найден"},
400: {"description": "Некорректный запрос"},
401: {"description": "Не авторизован"}
}
)

@router.get(
"/", 
response_model=List[ItemResponse],
summary="Получить список всех элементов",
description="""
Получает список элементов инвентаря с возможностью пагинации.

- **skip**: пропустить указанное количество элементов
- **limit**: максимальное количество элементов в ответе
""",
response_description="Список элементов инвентаря"
)
def get_items(
skip: int = Query(0, description="Сколько элементов пропустить"),
limit: int = Query(100, description="Максимальное количество элементов", le=1000),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user)
):
items = db.query(Item).offset(skip).limit(limit).all()
return items

# Аналогично обновляем остальные методы с более подробной документацией