Как создать CRUD REST API с Flask и PostgreSQL

Это пошаговое руководство по созданию простого CRUD REST API на Flask с базой данных PostgreSQL. Вы узнаете, как настроить ElephantSQL, создать виртуальное окружение, подключиться к БД, реализовать POST/GET/PUT/DELETE-роуты и протестировать их. В конце — чек-листы, критерии приёмки, тест-кейсы и рекомендации по безопасности и миграции.

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

В этом материале показано, как реализовать простой REST API с четырьмя CRUD-операциями (Create, Read, Update, Delete) с использованием Flask — лёгкого Python-фреймворка — и PostgreSQL.

О чём этот материал

• Подготовка облачной базы данных PostgreSQL (ElephantSQL).
• Создание и активация виртуального окружения Python.
• Настройка Flask-приложения и подключение к базе через psycopg2.
• Реализация маршрутов POST, GET, PUT, DELETE.
• Примеры запросов curl и тест-кейсы.
• Рекомендации по безопасности, тестированию и миграции на FastAPI/ORM.

В начале добавлены примеры команд и готовые сниппеты, затем — чек-листы для ролей и итоговое краткое резюме.

Подготовка базы данных PostgreSQL (ElephantSQL)

Если вы хотите использовать управляемый облачный экземпляр PostgreSQL, зайдите на ElephantSQL, зарегистрируйтесь и создайте новую инстанцию.

Нажмите кнопку “Create New Instance” (Создать новый экземпляр) и выберите бесплатный план (или подходящий тариф). Укажите имя инстанса и регион размещения.

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

Важно: URL содержит логин, пароль, хост и название БД в одном строчном формате. Храните его в защищённом месте (переменные окружения, секретный менеджер).

Настройка Flask-сервера

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

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

python --version
pip install virtualenv
virtualenv venv

Активируйте виртуальное окружение:

# На Windows:
.\venv\Scripts\activate
# На Unix / macOS:
source venv/bin/activate

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

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

flask
python-dotenv
psycopg2-binary

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

pip install -r requirements.txt

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

Создайте файл .env и вставьте туда строку подключения к БД (DATABASE_URL):

DATABASE_URL=ваш_URL_базы_данных

Создание app.py (минимальная конфигурация)

Создайте файл 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"

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

Пояснение: этот код создаёт глобальное подключение к БД и простой маршрут /, возвращающий строку. В продакшне рекомендуется использовать пул подключений и не держать глобальное подключение открытым между процессами (см. раздел безопасность).

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

Ниже — полная логика для работы с таблицей users. Для компактности используются SQL-строки и psycopg2. Альтернатива — ORM (SQLAlchemy, Peewee) или использование FastAPI.

Создание таблицы users

В app.py добавьте SQL-операцию создания таблицы (однократно при запуске или при миграции):

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 и name.

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

Добавьте в app.py код для вставки пользователя и возврата его ID:

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

Пример запроса curl:

curl -X POST http://localhost:5000/api/user -H "Content-Type: application/json" -d '{"name":"Alice"}'

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": "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

Примеры curl:

curl http://localhost:5000/api/user
curl http://localhost:5000/api/user/1

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

Добавьте SQL-константу и маршрут для обновления:

UPDATE_USER_BY_ID = "UPDATE users SET name = %s WHERE id = %s"

@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."})

Пример curl:

curl -X PUT http://localhost:5000/api/user/1 -H "Content-Type: application/json" -d '{"name":"Bob"}'

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

Добавьте SQL-константу и маршрут удаления:

DELETE_USER_BY_ID = "DELETE FROM users WHERE id = %s"

@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."})

Пример curl:

curl -X DELETE http://localhost:5000/api/user/1

Рекомендации по улучшению и безопасность

Important: следующий список содержит практики, которые стоит внедрить для продакшн-приложений.

• Используйте пул подключений (psycopg2.pool.SimpleConnectionPool или pgbouncer) вместо одного глобального соединения.
• Обрабатывайте исключения (psycopg2.Error) и возвращайте корректные коды ответа.
• Валидируйте входные данные (schema validation: marshmallow, pydantic).
• Используйте HTTPS и храните секреты в секрет-менеджере или переменных окружения.
• Ограничьте права пользователя БД (минимальные привилегии для приложения).
• Настройте логирование запросов и ошибок, не выводите секреты в логи.
• Регулярно делайте бэкапы БД и проверяйте restore-процедуры.

Безопасность запросов

В приведённом коде используются параметризованные запросы (placeholders %s) — это защита от SQL-инъекций. Не формируйте строки SQL через форматирование Python.

Производительность и масштабирование

• Для высокой нагрузки рассмотрите пул подключений и асинхронный сервер (uvicorn + FastAPI).
• ORM (SQLAlchemy) даёт удобство миграций и моделей, но добавляет слой абстракции.

Альтернативные подходы и миграция

• FastAPI — современная асинхронная альтернатива Flask с автогенерацией OpenAPI и лучшей производительностью в I/O-операциях.
• SQLAlchemy (ORM) или Alembic (миграции) для управления схемой и миграциями.
• Докеризация приложения и деплой в контейнерах для повторяемости окружения.
• Если нужна схема документов — MongoDB как альтернатива реляционной модели.

Тонкая миграция с Flask на FastAPI: перевод маршрутов на async, замена Flask-Request/Response на Starlette-интерфейс, замена WSGI-сервера на ASGI (uvicorn/gunicorn+uvicorn workers).

Тесты и критерии приёмки

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

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

Тест-кейсы (пример)

1. Создание пользователя
   • Вход: POST с телом {“name”:”Test”}
   • Ожидается: 201, тело содержит id и name = Test
2. Получение списка пользователей
   • Вход: GET /api/user
   • Ожидается: 200 и массив с ранее созданным пользователем
3. Обновление некорректного id
   • Вход: PUT /api/user/9999
   • Ожидается: 404
4. Удаление пользователя
   • Вход: DELETE /api/user/{id}
   • Ожидается: 200 и подтверждение удаления

Playbook: развёртывание и откат

Минимальный SOP для релиза:

1. Создать образ (Dockerfile) и прогнать локальные интеграционные тесты.
2. Деплой на staging, прогнать smoke tests (POST/GET/PUT/DELETE).
3. Переключить трафик на новую версию по шагам (канареечный деплой).
4. Мониторить логи и метрики 5–15 минут.
5. Если критическая ошибка — выполнить rollback на предыдущий образ и восстановить БД из бэкапа при необходимости.

Критерии отката: ошибки 5xx > 1% запросов или падение ключевых сценариев (создание/чтение).

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

Разработчик

• Локально протестировать все эндпоинты.
• Добавить unit/integration тесты.
• Обеспечить обработку ошибок.

DevOps

• Настроить пул подключений/pgbouncer.
• Добавить секрет-менеджер и CI/CD.
• Настроить бэкапы и мониторинг.

QA

• Проверить happy path и негативные сценарии.
• Проверить обработку ошибок БД.
• Выполнить нагрузочное тестирование (см. рекомендации по масштабированию).

Сравнение: Flask vs FastAPI (кратко)

| Подход | Синхронный, минималистичный | ASGI, ориентирован на async и валидацию | Автогенерация OpenAPI | Нет (требуется расширение) | Да (встроено) | Производительность | Хорошая для умеренной нагрузки | Лучше при высоком I/O | Простота | Прост для начала | Немного круче кривая, но мощнее для API

Ментальные модели и рекомендации

• Модель “слой ответственности”: маршруты — обработка ввода/вывода, сервисы — бизнес-логика, репозитории — доступ к данным.
• “Fail fast”: валидируйте данные как можно раньше и возвращайте понятные ошибки.
• “Least privilege”: выдавайте приложению минимум прав в БД.

Маленькая шпаргалка (cheat sheet)

• Запуск сервера: python app.py (или через gunicorn для продакшена).
• Просмотр переменных окружения: print(os.getenv(“DATABASE_URL”)) — только для отладки.
• Сбор логов ошибок: использовать logging.exception в блоках except.

Короткий словарь (1 строка)

• CRUD — Create, Read, Update, Delete.
• Flask — лёгкий фреймворк для веб-приложений на Python.
• psycopg2 — адаптер PostgreSQL для Python.
• ElephantSQL — managed PostgreSQL в облаке.

FAQ

Как запустить это приложение локально?

1. Создайте виртуальное окружение и активируйте его.
2. Установите зависимости: pip install -r requirements.txt.
3. Вставьте DATABASE_URL в .env.
4. Запустите python app.py и тестируйте эндпоинты curl или Postman.

Подходит ли этот код для production?

Шаблон полезен для прототипа или небольших проектов. Для production: добавьте пул подключений, миграции, обработку ошибок, логирование, HTTPS и ограничение прав БД.

Итог

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

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

• Начните с безопасного хранения DATABASE_URL.
• Используйте параметризованные SQL-запросы.
• Автоматизируйте тесты и деплой.

Спасибо за чтение. Удачи при разработке вашего API!