Создание CRUD REST API на Flask и PostgreSQL

Введение

Application Programming Interfaces (API) — ключевой компонент для связи между системами: они позволяют обмениваться данными и вызывать действия между приложениями и внешними сервисами. Backend-разработка включает в себя не только написание API, но и серверную бизнес-логику, проектирование архитектуры, работу с БД и наблюдаемостью.

Ниже — практическая инструкция по созданию простого CRUD REST API на Flask (легковесный Python-фреймворк) и PostgreSQL. Руководство охватывает настройку базы в облаке, поднятие локального сервера, реализацию маршрутов и рекомендации по безопасности и развертыванию.

Что вы получите

• Рабочую структуру Flask-приложения, подключённого к PostgreSQL.
• Примеры SQL-запросов и HTTP-маршрутов для CRUD.
• Чек-листы для разработчика и DevOps.
• Настройки безопасности и критерии приёмки.

Использование Flask для создания API

Flask — минималистичный Python-фреймворк, удобный для написания backend API и интеграции с фронтендами на React, Angular и другими. Исходные файлы примера доступны в репозитории проекта (ссылка в оригинале).

Это руководство реализует REST API с четырьмя операциями CRUD для управления данными пользователей в PostgreSQL.

Настройка базы данных PostgreSQL

Для быстрого старта можно использовать ElephantSQL — облачный провайдер PostgreSQL. Зарегистрируйтесь и создайте инстанс.

Нажмите кнопку “Create New Instance”, чтобы создать новый инстанс для приложения.

Укажите имя инстанса, выберите бесплатный тариф и регион хоста, затем завершите создание.

После создания откройте страницу настроек инстанса и скопируйте DATABASE URL — он понадобится для подключения.

Подготовка сервера Flask

В терминале создайте папку проекта и перейдите в неё. Убедитесь, что у вас установлен Python 3.6+:

python --version

Установите virtualenv и создайте виртуальную среду:

pip install virtualenv
virtualenv venv

Активируйте виртуальную среду:

# On Windows:   
.\venv\Scripts\activate  
# On Unix or MacOS:   
source venv/bin/activate

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

Создайте файл requirements.txt в корне проекта и добавьте:

flask  
python-dotenv  
psycopg2-binary

Установите зависимости:

pip install -r requirements.txt

Библиотека psycopg2-binary — адаптер Python ↔ PostgreSQL, используемый для установки соединения и выполнения SQL-запросов.

Создайте файл .env и сохраните в нём строку подключения:

DATABASE_URL= your database URL

Создание файла сервера

Создайте app.py и добавьте базовую инициализацию приложения и подключение к базе:

import os   
import psycopg2  
from dotenv import load_dotenv  
from flask import Flask, request, jsonify  
  
load_dotenv()  
  
app = Flask(__name__)  
url = os.getenv("DATABASE_URL")  
connection = psycopg2.connect(url)  
  
  
@app.get("/")  
def home():  
    return "hello world"

Этот код инициализирует экземпляр Flask, подключается к базе по URL и добавляет маршрут “/“ для проверки работоспособности.

Создание REST API с операциями CRUD

Далее реализуем таблицу пользователей и четыре основных маршрута: POST, GET (все и по id), PUT и DELETE.

Создание таблицы пользователей

В app.py выполните SQL для создания таблицы users, если её ещё нет:

CREATE_USERS_TABLE = "CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name TEXT);"  
  
with connection:  
    with connection.cursor() as cursor:  
        cursor.execute(CREATE_USERS_TABLE)

• Таблица users содержит id (SERIAL) и поле name (TEXT).
• Используется connection.cursor() для выполнения SQL-запросов через psycopg2.

1. Метод POST — создание пользователя

Маршрут для добавления нового пользователя:

INSERT_USER_RETURN_ID = "INSERT INTO users (name) VALUES (%s) RETURNING id;"  
@app.route("/api/user", methods=["POST"])  
def create_user():  
    data = request.get_json()  
    name = data["name"]  
    with connection:  
        with connection.cursor() as cursor:  
            cursor.execute(INSERT_USER_RETURN_ID, (name,))  
            user_id = cursor.fetchone()[0]  
    return {"id": user_id, "name": name, "message": f"User {name} created."}, 201  

• SQL-запрос вставляет имя в таблицу и возвращает id нового пользователя.
• Ответ содержит id, name и сообщение об успешном создании.

2. Метод GET — получение данных

Два маршрута: получение всех пользователей и получение пользователя по id.

SELECT_ALL_USERS = "SELECT * FROM users;"  
  
@app.route("/api/user", methods=["GET"])  
def get_all_users():  
    with connection:  
        with connection.cursor() as cursor:  
            cursor.execute(SELECT_ALL_USERS)  
            users = cursor.fetchall()  
            if users:  
                result = []  
                for user in users:  
                    result.append({"id": user[0], "name": user[1]})  
                return jsonify(result)  
            else:  
                return jsonify({"error": f"Users not found."}), 404  
  
@app.route("/api/user/", methods=["GET"])  
def get_user(user_id):  
    with connection:  
        with connection.cursor() as cursor:  
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))            
            user = cursor.fetchone()  
            if user:  
                return jsonify({"id": user[0], "name": user[1]})  
            else:  
                return jsonify({"error": f"User with ID {user_id} not found."}), 404

• Первый маршрут возвращает список всех пользователей в JSON.
• Второй возвращает пользователя по id или 404, если не найден.

3. Метод PUT — обновление пользователя

Маршрут для обновления имени пользователя по id:

@app.route("/api/user/", methods=["PUT"])  
def update_user(user_id):  
    data = request.get_json()  
    name = data["name"]  
    with connection:  
        with connection.cursor() as cursor:  
            cursor.execute(UPDATE_USER_BY_ID, (name, user_id))  
            if cursor.rowcount == 0:  
                return jsonify({"error": f"User with ID {user_id} not found."}), 404  
    return jsonify({"id": user_id, "name": name, "message": f"User with ID {user_id} updated."})

• Если операция обновления не изменила ни одной строки, возвращается 404.

4. Метод DELETE — удаление пользователя

Маршрут для удаления пользователя по id:

@app.route("/api/user/", methods=["DELETE"])  
def delete_user(user_id):  
    with connection:  
        with connection.cursor() as cursor:  
            cursor.execute(DELETE_USER_BY_ID, (user_id,))  
            if cursor.rowcount == 0:  
                return jsonify({"error": f"User with ID {user_id} not found."}), 404  
    return jsonify({"message": f"User with ID {user_id} deleted."})

• Возвращается сообщение об удалении или 404, если пользователь не найден.

Практические советы и лучшие практики

Критерии приёмки

• Все CRUD-эндпоинты отвечают ожидаемыми кодами: 201 для создания, 200 для успешной операции, 404 при отсутствии ресурса.
• Валидация входных данных (например, непустое имя) выполнена до выполнения SQL.
• SQL-инъекции предотвращены с помощью параметризованных запросов (psycopg2 использует подстановку %s).
• Логирование ошибок и метрик включено, есть минимум юнит-тестов для каждого маршрута.

Мини-методология для вывода в продакшн

1. Разделите конфигурации: используйте переменные окружения и секреты (Vault, AWS Secrets Manager).
2. Перенесите соединение с DB в пул соединений (psycopg2.pool) — для высокой нагрузки.
3. Добавьте миграции схемы (Alembic) вместо вручную выполняемых CREATE TABLE.
4. Настройте CI/CD: тесты → построение артефакта → деплой в staging → smoke-тесты → прод.
5. Мониторинг и алертинг: SLO/SLI для доступности / latency.

Чек-листы по ролям

• Developer:

  • Проверить, что все маршруты покрыты юнит/интеграционными тестами.
  • Использовать parameterized queries и проверки входа.
  • Обрабатывать и логировать исключения.

• DevOps / SRE:

  • Настроить пул соединений и health-check endpoint.
  • Настроить автоскейлинг и мониторинг (CPU, connection count, p95 latency).
  • Управлять секретами и ротацией ключей.

Сниппет: полезные SQL- и Python-моменты

• Примеры SQL-запросов для переменных мест:

-- Создание таблицы
CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name TEXT);

-- Вставка и возврат id
INSERT INTO users (name) VALUES (%s) RETURNING id;

-- Получение по id
SELECT * FROM users WHERE id = %s;

-- Обновление
UPDATE users SET name = %s WHERE id = %s;

-- Удаление
DELETE FROM users WHERE id = %s;

• Рекомендуемый шаблон ответа JSON:
  • Успех: {“id”: …, “name”: …, “message”: “…”}
  • Ошибка: {“error”: “описание”}

Безопасность и харднинг

• Всегда используйте параметризованные запросы (Prepared statements) — в примере это %s в psycopg2.
• Не храните DATABASE_URL в репозитории; используйте .env только для локальной разработки.
• Ограничьте права пользователя БД: используйте отдельную роль с минимальными привилегиями.
• Включите TLS/SSL для соединения с базой, если провайдер поддерживает.
• Ограничьте rate для API, чтобы предотвратить DDoS и перебор ресурсов.

Когда такой подход не подходит

• Высокая нагрузка и требование низкой задержки: подумайте о пуле соединений, оптимизации запросов или использовании более производительного интерфейса (async).
• Нужна строгая валидация и автогенерация схемы API: FastAPI даёт более строгую типизацию и автодокументацию (OpenAPI).
• Если нужна схема без SQL и горизонтальное масштабирование по документам: MongoDB может быть предпочтительнее.

Альтернативные подходы

• FastAPI: высокопроизводительный асинхронный фреймворк с автогенерацией схем и лёгкой валидацией (pydantic).
• Django REST Framework: когда нужно быстро получить комплексные CRUD-интерфейсы и аутентификацию.
• Базы: PostgreSQL для реляционной структуры; MongoDB для схем с гибкой структурой.

Сравнение (схематично)

| MongoDB | Гибкая схема, документы | Быстро менять структуру | Нет транзакций уровня RDBMS (есть multi-doc транзакции)

Тест-кейсы и приёмка

• POST /api/user с валидным именем → 201, тело содержит id и name.
• POST /api/user с пустым именем → 400 и сообщение об ошибке валидации.
• GET /api/user → 200 и массив пользователей.
• GET /api/user/{id} при существующем id → 200 и объект пользователя.
• GET /api/user/{id} при несуществующем id → 404.
• PUT /api/user/{id} с корректными данными → 200 и обновлённый объект.
• DELETE /api/user/{id} при существующем id → 200 и сообщение об удалении.

Быстрая галерея крайних случаев

• Конкурентная вставка с одинаковыми уникальными полями — обработайте ошибку и добавьте повторную попытку при необходимости.
• Падение подключения к БД — приложение должно корректно возвращать 5xx и логировать причину.

Глоссарий (одной строкой)

• REST — архитектурный стиль для веб-API, использующий HTTP-методы как операции над ресурсами.
• CRUD — четыре базовые операции над данными: Create, Read, Update, Delete.
• psycopg2 — драйвер Python для PostgreSQL.

Итог

Это руководство показало, как создать базовый CRUD REST API на Flask и PostgreSQL: от настройки облачной БД до реализации маршрутов и списков практик для продакшна. Для реальных проектов добавьте валидацию, пул соединений, миграции (Alembic), логирование и мониторинг — это минимальный набор для стабильной эксплуатации.

Важно: Flask отлично подходит для лёгких и средних по сложности проектов. Для высоконагруженных систем рассмотрите асинхронные фреймворки и оптимизацию пулов подключений.

Краткое резюме:

• Flask + psycopg2 — быстрый путь к рабочему API.
• Используйте parameterized queries и переменные окружения.
• Автоматизируйте миграции и тестирование перед деплоем.

Дополнительно: можно рассмотреть FastAPI + asyncpg для асинхронности и более строгой валидации, либо Django REST Framework для сложных проектов со встроенной auth и ORM.