REST API на Flask

REST API на Flask

Базовое решение

from flask import Flask, request, jsonify
from datetime import datetime
from typing import Dict, List, Tuple, Any, Optional

app = Flask(__name__)

# Хранилище задач (в памяти)
tasks_db: Dict[int, Dict[str, Any]] = {}
next_id = 1

@app.route('/tasks', methods=['GET'])
def get_tasks() -> Tuple[Dict[str, List], int]:
    """Получить список всех задач"""
    return jsonify(list(tasks_db.values())), 200

@app.route('/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id: int) -> Tuple[Dict[str, Any], int]:
    """Получить задачу по ID"""
    task = tasks_db.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    return jsonify(task), 200

@app.route('/tasks', methods=['POST'])
def create_task() -> Tuple[Dict[str, Any], int]:
    """Создать новую задачу"""
    global next_id
    
    data = request.get_json()
    if not data or 'title' not in data:
        return jsonify({"error": "Title is required"}), 400
    
    task = {
        "id": next_id,
        "title": data['title'],
        "done": data.get('done', False),
        "created_at": datetime.utcnow().isoformat()
    }
    tasks_db[next_id] = task
    next_id += 1
    
    return jsonify(task), 201

@app.route('/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id: int) -> Tuple[Dict[str, Any], int]:
    """Обновить задачу"""
    if task_id not in tasks_db:
        return jsonify({"error": "Task not found"}), 404
    
    data = request.get_json()
    task = tasks_db[task_id]
    
    if 'title' in data:
        task['title'] = data['title']
    if 'done' in data:
        task['done'] = data['done']
    
    task['updated_at'] = datetime.utcnow().isoformat()
    
    return jsonify(task), 200

@app.route('/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id: int) -> Tuple[Dict[str, Any], int]:
    """Удалить задачу"""
    if task_id not in tasks_db:
        return jsonify({"error": "Task not found"}), 404
    
    deleted = tasks_db.pop(task_id)
    return jsonify(deleted), 200

if __name__ == '__main__':
    app.run(debug=True)

Улучшенное решение с валидацией

from flask import Flask, request, jsonify
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Dict, Optional
import uuid

app = Flask(__name__)

@dataclass
class Task:
    """Модель задачи"""
    id: str
    title: str
    done: bool = False
    created_at: str = None
    updated_at: Optional[str] = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.utcnow().isoformat()
    
    def to_dict(self):
        return asdict(self)

class TaskStore:
    """Хранилище задач"""
    def __init__(self):
        self.tasks: Dict[str, Task] = {}
    
    def create(self, title: str, done: bool = False) -> Task:
        task_id = str(uuid.uuid4())
        task = Task(id=task_id, title=title, done=done)
        self.tasks[task_id] = task
        return task
    
    def get(self, task_id: str) -> Optional[Task]:
        return self.tasks.get(task_id)
    
    def get_all(self) -> list[Task]:
        return list(self.tasks.values())
    
    def update(self, task_id: str, title: Optional[str] = None, done: Optional[bool] = None) -> Optional[Task]:
        task = self.get(task_id)
        if not task:
            return None
        
        if title is not None:
            task.title = title
        if done is not None:
            task.done = done
        task.updated_at = datetime.utcnow().isoformat()
        
        return task
    
    def delete(self, task_id: str) -> Optional[Task]:
        return self.tasks.pop(task_id, None)

store = TaskStore()

def validate_task_data(data: dict) -> Tuple[bool, Optional[str]]:
    """Валидация данных задачи"""
    if not data:
        return False, "Request body is empty"
    
    if 'title' not in data:
        return False, "Field 'title' is required"
    
    title = data['title']
    if not isinstance(title, str) or not title.strip():
        return False, "Title must be a non-empty string"
    
    if 'done' in data and not isinstance(data['done'], bool):
        return False, "Field 'done' must be boolean"
    
    return True, None

@app.route('/tasks', methods=['GET'])
def get_tasks():
    """Получить все задачи"""
    tasks = [task.to_dict() for task in store.get_all()]
    return jsonify(tasks), 200

@app.route('/tasks/<task_id>', methods=['GET'])
def get_task(task_id: str):
    """Получить задачу по ID"""
    task = store.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    return jsonify(task.to_dict()), 200

@app.route('/tasks', methods=['POST'])
def create_task():
    """Создать задачу"""
    data = request.get_json()
    
    is_valid, error_message = validate_task_data(data)
    if not is_valid:
        return jsonify({"error": error_message}), 400
    
    task = store.create(
        title=data['title'].strip(),
        done=data.get('done', False)
    )
    
    return jsonify(task.to_dict()), 201

@app.route('/tasks/<task_id>', methods=['PUT'])
def update_task(task_id: str):
    """Обновить задачу"""
    task = store.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    
    data = request.get_json()
    if not data:
        return jsonify({"error": "Request body is empty"}), 400
    
    # Валидируем, что if there's title or done
    if 'title' in data:
        if not isinstance(data['title'], str) or not data['title'].strip():
            return jsonify({"error": "Title must be a non-empty string"}), 400
    
    if 'done' in data and not isinstance(data['done'], bool):
        return jsonify({"error": "Field 'done' must be boolean"}), 400
    
    updated_task = store.update(
        task_id,
        title=data.get('title'),
        done=data.get('done')
    )
    
    return jsonify(updated_task.to_dict()), 200

@app.route('/tasks/<task_id>', methods=['DELETE'])
def delete_task(task_id: str):
    """Удалить задачу"""
    task = store.delete(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    
    return jsonify(task.to_dict()), 200

@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Endpoint not found"}), 404

@app.errorhandler(500)
def server_error(error):
    return jsonify({"error": "Internal server error"}), 500

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

Тестирование API

import pytest
from flask import Flask

@pytest.fixture
def client():
    """Fixture для тестирования"""
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client

def test_create_task(client):
    """Тест создания задачи"""
    response = client.post('/tasks', json={
        'title': 'Test task',
        'done': False
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data['title'] == 'Test task'
    assert data['done'] is False
    assert 'id' in data

def test_get_tasks(client):
    """Тест получения списка задач"""
    # Создаем задачу
    client.post('/tasks', json={'title': 'Task 1'})
    client.post('/tasks', json={'title': 'Task 2'})
    
    # Получаем список
    response = client.get('/tasks')
    assert response.status_code == 200
    tasks = response.get_json()
    assert len(tasks) >= 2

def test_update_task(client):
    """Тест обновления задачи"""
    # Создаем
    create_resp = client.post('/tasks', json={'title': 'Original'})
    task_id = create_resp.get_json()['id']
    
    # Обновляем
    response = client.put(f'/tasks/{task_id}', json={
        'title': 'Updated',
        'done': True
    })
    assert response.status_code == 200
    data = response.get_json()
    assert data['title'] == 'Updated'
    assert data['done'] is True

def test_delete_task(client):
    """Тест удаления задачи"""
    create_resp = client.post('/tasks', json={'title': 'To delete'})
    task_id = create_resp.get_json()['id']
    
    response = client.delete(f'/tasks/{task_id}')
    assert response.status_code == 200
    
    # Проверяем что удалена
    get_resp = client.get(f'/tasks/{task_id}')
    assert get_resp.status_code == 404

def test_invalid_data(client):
    """Тест валидации данных"""
    response = client.post('/tasks', json={'done': True})
    assert response.status_code == 400

Запуск и тестирование

# Установка зависимостей
pip install flask pytest

# Запуск приложения
python app.py

# Тестирование с curl
curl -X GET http://localhost:5000/tasks
curl -X POST http://localhost:5000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "Купить молоко", "done": false}'

curl -X PUT http://localhost:5000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"done": true}'

curl -X DELETE http://localhost:5000/tasks/1

# Тестирование с pytest
pytest test_app.py -v

Production-ready версия

Для production нужны улучшения:

1. База данных — вместо памяти
2. Аутентификация — JWT токены
3. Логирование — структурированные логи
4. Rate limiting — защита от DDoS
5. CORS — для фронтенда
6. Swagger документация — для API

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from flasgger import Swagger

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/tasks_db'
db = SQLAlchemy(app)
CORS(app)
limiter = Limiter(app=app, key_func=get_remote_address)
swagger = Swagger(app)

# Далее использовать SQLAlchemy модели вместо памяти