Разработка RESTful API на Python: пошаговое руководство от Flask до деплоя

mkdir bookstore_api
cd bookstore_api
python -m venv venv
source venv/bin/activate # На Windows: venv\Scripts\activate
pip install flask flask-restful flask-sqlalchemy marshmallow

bookstore_api/
├── venv/
├── app.py # Основной файл приложения
├── models.py # Модели данных
├── resources.py # API ресурсы и эндпоинты
├── schemas.py # Схемы сериализации/валидации
└── config.py # Конфигурация приложения

from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy

# Инициализация приложения
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///bookstore.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'your-secret-key' # В продакшене использовать переменную окружения!

# Инициализация расширений
db = SQLAlchemy(app)
api = Api(app)

# Импортируем ресурсы после инициализации db
from resources import BookResource, BookListResource

# Регистрация маршрутов
api.add_resource(BookListResource, '/api/books')
api.add_resource(BookResource, '/api/books/<int:book_id>')

if __name__ == '__main__':
# Создаем таблицы при запуске
with app.app_context():
db.create_all()
app.run(debug=True)

from app import db
from datetime import datetime

class Book(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
author = db.Column(db.String(50), nullable=False)
description = db.Column(db.Text)
price = db.Column(db.Float, nullable=False)
in_stock = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)

def __repr__(self):
return f'<Book {self.title}>'

from marshmallow import Schema, fields, validate

class BookSchema(Schema):
id = fields.Int(dump_only=True) # Только для вывода, не для ввода
title = fields.Str(required=True, validate=validate.Length(min=1, max=100))
author = fields.Str(required=True, validate=validate.Length(min=1, max=50))
description = fields.Str()
price = fields.Float(required=True, validate=validate.Range(min=0))
in_stock = fields.Bool(default=True)
created_at = fields.DateTime(dump_only=True)

from flask import request
from flask_restful import Resource
from marshmallow import ValidationError
from app import db
from models import Book
from schemas import BookSchema

book_schema = BookSchema()
books_schema = BookSchema(many=True)

class BookListResource(Resource):
def get(self):
books = Book.query.all()
return books_schema.dump(books), 200

def post(self):
try:
# Валидация входящих данных
book_data = book_schema.load(request.json)

# Создание новой книги
new_book = Book(
title=book_data['title'],
author=book_data['author'],
description=book_data.get('description', ''),
price=book_data['price'],
in_stock=book_data.get('in_stock', True)
)

# Сохранение в базу данных
db.session.add(new_book)
db.session.commit()

return book_schema.dump(new_book), 201

except ValidationError as err:
return {"message": "Validation error", "errors": err.messages}, 400

class BookResource(Resource):
def get(self, book_id):
book = Book.query.get_or_404(book_id)
return book_schema.dump(book), 200

python app.py

curl -X POST http://localhost:5000/api/books \
-H "Content-Type: application/json" \
-d '{"title": "Python для профессионалов", "author": "Лучано Рамальо", "description": "Продвинутые возможности языка Python", "price": 45.99}'

class BookResource(Resource):
def get(self, book_id):
book = Book.query.get_or_404(book_id, 
description=f"Книга с идентификатором {book_id} не найдена")
return book_schema.dump(book), 200

def put(self, book_id):
book = Book.query.get_or_404(book_id)

try:
# Полное обновление ресурса
book_data = book_schema.load(request.json)

book.title = book_data['title']
book.author = book_data['author']
book.description = book_data.get('description', '')
book.price = book_data['price']
book.in_stock = book_data.get('in_stock', book.in_stock)

db.session.commit()
return book_schema.dump(book), 200

except ValidationError as err:
return {"message": "Validation error", "errors": err.messages}, 400

def patch(self, book_id):
book = Book.query.get_or_404(book_id)

# Для частичного обновления не требуем все поля
if 'title' in request.json:
book.title = request.json['title']
if 'author' in request.json:
book.author = request.json['author']
if 'description' in request.json:
book.description = request.json['description']
if 'price' in request.json:
book.price = request.json['price']
if 'in_stock' in request.json:
book.in_stock = request.json['in_stock']

# Валидация после изменения
try:
book_schema.dump(book)
db.session.commit()
return book_schema.dump(book), 200
except ValidationError as err:
db.session.rollback()
return {"message": "Validation error", "errors": err.messages}, 400

def delete(self, book_id):
book = Book.query.get_or_404(book_id)
db.session.delete(book)
db.session.commit()
return {"message": f"Книга {book_id} успешно удалена"}, 204

from marshmallow import Schema, fields, validate, validates, ValidationError
import re

class BookSchema(Schema):
id = fields.Int(dump_only=True)
title = fields.Str(required=True, validate=validate.Length(min=1, max=100))
author = fields.Str(required=True, validate=validate.Length(min=1, max=50))
description = fields.Str()
price = fields.Float(required=True, validate=validate.Range(min=0))
in_stock = fields.Bool(default=True)
created_at = fields.DateTime(dump_only=True)

# Собственный валидатор для проверки формата имени автора
@validates('author')
def validate_author(self, value):
if not re.match(r'^[A-ZА-Я][a-zа-я]+ [A-ZА-Я][a-zа-я]+$', value):
raise ValidationError('Имя автора должно быть в формате "Имя Фамилия"')

# Комплексная валидация нескольких полей
@validates_schema
def validate_price_for_premium_books(self, data, **kwargs):
if 'title' in data and 'price' in data:
if 'премиум' in data['title'].lower() and data['price'] < 30:
raise ValidationError('Премиум книги должны иметь цену от 30')

@app.errorhandler(404)
def resource_not_found(e):
return {"message": str(e)}, 404

@app.errorhandler(400)
def bad_request(e):
return {"message": "Неверный формат запроса"}, 400

@app.errorhandler(500)
def internal_server_error(e):
return {"message": "Внутренняя ошибка сервера"}, 500

class BookListResource(Resource):
def get(self):
# Получаем параметры из query string
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
author = request.args.get('author')
min_price = request.args.get('min_price', type=float)
max_price = request.args.get('max_price', type=float)

# Базовый запрос
query = Book.query

# Применяем фильтры, если они указаны
if author:
query = query.filter(Book.author.ilike(f'%{author}%'))
if min_price is not None:
query = query.filter(Book.price >= min_price)
if max_price is not None:
query = query.filter(Book.price <= max_price)

# Пагинация
paginated_books = query.paginate(page=page, per_page=per_page)

# Формируем результат с метаданными
result = {
"books": books_schema.dump(paginated_books.items),
"meta": {
"page": page,
"per_page": per_page,
"total_pages": paginated_books.pages,
"total_items": paginated_books.total
}
}

return result, 200

pip install flask-jwt-extended

from flask_jwt_extended import JWTManager, create_access_token, jwt_required

# Инициализация JWT
app.config['JWT_SECRET_KEY'] = 'super-secret-key' # В продакшене использовать переменные окружения
jwt = JWTManager(app)

from flask_jwt_extended import create_access_token

class UserLoginResource(Resource):
def post(self):
# В реальном приложении здесь была бы проверка по базе данных
if not request.is_json:
return {"message": "Отсутствует JSON в теле запроса"}, 400

username = request.json.get('username', None)
password = request.json.get('password', None)

if not username or not password:
return {"message": "Отсутствуют учетные данные"}, 400

# Простая проверка (в реальном API использовать безопасную аутентификацию!)
if username != 'admin' or password != 'password':
return {"message": "Неверное имя пользователя или пароль"}, 401

access_token = create_access_token(identity=username)
return {"access_token": access_token}, 200

from flask_jwt_extended import jwt_required

api.add_resource(UserLoginResource, '/api/login')

# И модифицируем методы создания/изменения в BookResource:
@jwt_required() # Требуем токен для создания
def put(self, book_id):
# ...код метода...

@jwt_required()
def patch(self, book_id):
# ...код метода...

@jwt_required()
def delete(self, book_id):
# ...код метода...

import unittest
import json
from app import app, db
from models import Book

class BookAPITestCase(unittest.TestCase):
def setUp(self):
# Настраиваем тестовое окружение
app.config['TESTING'] = True
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
self.client = app.test_client()

# Создаем контекст приложения и таблицы БД
with app.app_context():
db.create_all()

# Добавляем тестовые данные
test_book = Book(
title='Тестовая книга',
author='Тест Тестов',
description='Книга для тестирования API',
price=19.99
)
db.session.add(test_book)
db.session.commit()

def tearDown(self):
# Очищаем после каждого теста
with app.app_context():
db.session.remove()
db.drop_all()

def test_get_books(self):
response = self.client.get('/api/books')
self.assertEqual(response.status_code, 200)
data = json.loads(response.data)
self.assertEqual(len(data['books']), 1)
self.assertEqual(data['books'][0]['title'], 'Тестовая книга')

def test_get_book(self):
response = self.client.get('/api/books/1')
self.assertEqual(response.status_code, 200)
data = json.loads(response.data)
self.assertEqual(data['title'], 'Тестовая книга')

def test_create_book(self):
# Получаем токен аутентификации
auth_response = self.client.post('/api/login',
json={'username': 'admin', 'password': 'password'})
token = json.loads(auth_response.data)['access_token']

new_book = {
'title': 'Новая книга',
'author': 'Автор Авторов',
'description': 'Описание новой книги',
'price': 29.99
}

response = self.client.post(
'/api/books',
json=new_book,
headers={'Authorization': f'Bearer {token}'}
)

self.assertEqual(response.status_code, 201)
data = json.loads(response.data)
self.assertEqual(data['title'], 'Новая книга')

# Проверяем, что книга действительно создана
response = self.client.get('/api/books')
data = json.loads(response.data)
self.assertEqual(len(data['books']), 2)

if __name__ == '__main__':
unittest.main()

pip install flask-restx

from flask import Flask
from flask_restx import Api, Resource, fields
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///bookstore.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)
api = Api(app, version='1.0', title='Bookstore API',
description='RESTful API для управления книжным магазином',
doc='/api/docs') # Swagger UI будет доступен по этому пути

from app import app

if __name__ == "__main__":
app.run()

bind = "0.0.0.0:8000"
workers = 4 # Рекомендуется: (2 x количество ядер) + 1
worker_class = "gthread"
threads = 2
timeout = 30
keepalive = 5

FROM python:3.9-slim

WORKDIR /app

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

COPY . .

ENV FLASK_APP=app.py
ENV FLASK_ENV=production
ENV SQLALCHEMY_DATABASE_URI=sqlite:///bookstore.db

RUN python -c "from app import db; db.create_all()"

EXPOSE 8000

CMD ["gunicorn", "--config", "gunicorn_config.py", "wsgi:app"]

version: '3'

services:
api:
build: .
ports:
- "8000:8000"
volumes:
- ./data:/app/data
environment:
- FLASK_ENV=production
- JWT_SECRET_KEY=${JWT_SECRET_KEY}
restart: always

flask==2.0.1
flask-restful==0.3.9
flask-sqlalchemy==2.5.1
flask-jwt-extended==4.3.1
flask-restx==0.5.1
marshmallow==3.13.0
gunicorn==20.1.0

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

location / {
return 301 https://$host$request_uri;
}
}

server {
listen 443 ssl;
server_name api.example.com;

ssl_certificate /etc/letsencrypt/live/api.example.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/api.example.com/privkey.pem;

location / {
proxy_pass http://localhost: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;
}
}