Используйте mssql-python с Flask

Flask — это лёгкий веб-фреймворк на Python, который даёт полный контроль над структурой приложения. В сочетании с mssql-python можно создавать веб-приложения и REST API на базе Microsoft SQL и База данных SQL Azure с минимальными накладными расходами.

Необходимые условия

  • Python 3.10 или более поздней версии.
  • Пакеты mssql-python и flask. Установите оба компонента с помощью pip install flask mssql-python.
  • Установите единовременные предварительные условия для операционной системы. Пользователи Windows могут пропустить этот шаг. Для полной информации о платформе см. Установить mssql-python.
    apk add libtool krb5-libs krb5-dev

Создание базы данных SQL

Создайте или подключитесь к SQL-базе данных на одной из следующих платформ:

В примерах этой статьи используется пример базы данных AdventureWorksLT, а именно таблица SalesLT.Product. Если у вас не установлен AdventureWorksLT, посмотрите примеры баз данных AdventureWorks.

Настройка проекта

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

Установите необходимые пакеты с помощью pip:

pip install flask mssql-python

структура проекта

Организуйте свой проект с отдельными модулями для настройки, управления соединениями, маршрутов и тестов:

my_app/
├── app.py            # Flask app and routes
├── config.py         # database settings
├── database.py       # connection lifecycle
├── test_app.py       # pytest tests
└── blueprints/       # optional: routes grouped into modules
    ├── __init__.py
    └── products.py

Управление подключением к базе данных

Flask не включает встроенный уровень базы данных, поэтому вы управляете соединениями напрямую. Паттерн в этом разделе хранит одно соединение на каждый запрос на объекте g Flask и автоматически закрывает его после окончания запроса.

Создайте config.py

Централизовать настройки базы данных в классе конфигурации. Переменные окружения позволяют отменять настройки по умолчанию без изменения кода.

# config.py
import os

class Config:
    """Application configuration."""
    DATABASE_SERVER = os.getenv("DB_SERVER", "<server>.database.windows.net")
    DATABASE_NAME = os.getenv("DB_NAME", "<database>")
    POOL_SIZE = int(os.getenv("DB_POOL_SIZE", "10"))

Создайте database.py

Модуль database.py управляет жизненным циклом подключения. Объект g во Flask — это пространство имён, отдельное для каждого запроса, поэтому хранение там соединения гарантирует, что каждый запрос получает собственное соединение, которое удаляется после завершения запроса.

Функция get_connection_string() строит строка подключения из конфигурации приложения. get_db() Функция создаёт соединение при первом вызове и повторно использует его для остальной части запроса. close_db() Функция запускается автоматически в конце каждого запроса, откатывая транзакцию в случае исключения и совершая иное. init_app() Функция регистрирует это поведение разбора с помощью приложения Flask.

# database.py
import mssql_python
from flask import g, current_app

def get_connection_string() -> str:
    """Build connection string from Flask app config."""
    cfg = current_app.config
    return (
        f"Server={cfg['DATABASE_SERVER']};"
        f"Database={cfg['DATABASE_NAME']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

def get_db():
    """Get a database cursor for the current request.

    The connection is stored on Flask's g object so it persists
    for the duration of the request and is reused across calls.
    """
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

def close_db(exception=None):
    """Close the database connection at the end of the request."""
    cursor = g.pop("db_cursor", None)
    conn = g.pop("db_conn", None)

    if cursor is not None:
        cursor.close()
    if conn is not None:
        if exception:
            conn.rollback()
        else:
            conn.commit()
        conn.close()

def init_app(app):
    """Register database teardown with the Flask app."""
    app.teardown_appcontext(close_db)

Замечание

ActiveDirectoryDefault использует DefaultAzureCredential, который последовательно пробует несколько поставщиков учетных данных. Первое соединение может быть медленным, потому что SDK идёт по цепочке, пока не найдёт работающего провайдера. В продакшене, если вы знаете, какой тип учетных данных использует ваша среда, укажите его напрямую (например, ActiveDirectoryMSI для управляемой идентичности), чтобы избежать цепной ходьбы. Дополнительные сведения см. в разделе проверки подлинности Microsoft Entra.

Приложение Flask

Следующий пример показывает полноценное приложение Flask с маршрутами для размещения, извлечения, создания, обновления и удаления продуктов.

Создайте app.py

Модуль приложения создаёт приложение Flask, загружает конфигурацию и регистрирует разборку базы данных. Каждая функция маршрута вызывает get_db(), чтобы получить курсор, выполняет запросы с параметризованным SQL (используя %(name)s плейсхолдеры и словарь значений) и возвращает JSON-ответы.

# app.py
from flask import Flask, jsonify, request, abort
from config import Config
from database import init_app, get_db

app = Flask(__name__)
app.config.from_object(Config)
init_app(app)

@app.route("/")
def index():
    return jsonify({"message": "Product API", "docs": "/products"})

@app.route("/products")
def list_products():
    """List products with pagination."""
    page = request.args.get("page", 1, type=int)
    page_size = request.args.get("page_size", 10, type=int)
    skip = (page - 1) * page_size

    cursor = get_db()

    cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
    total = cursor.fetchval()

    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        ORDER BY ProductID
        OFFSET %(skip)s ROWS
        FETCH NEXT %(limit)s ROWS ONLY
    """, {"skip": skip, "limit": page_size})

    items = [{
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()]

    return jsonify({
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    })

@app.route("/products/<int:product_id>")
def get_product(product_id):
    """Get a single product by ID."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        WHERE ProductID = %(id)s
    """, {"id": product_id})

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products", methods=["POST"])
def create_product():
    """Create a new product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    # OUTPUT INSERTED returns the new row's columns in the same statement,
    # so you don't need a separate SELECT to get the generated ID and defaults.
    # ProductNumber is required and unique. StandardCost and SellStartDate are
    # also NOT NULL in SalesLT.Product, so supply values for them.
    cursor.execute("""
        INSERT INTO SalesLT.Product
            (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
    """, {
        "name": data["name"],
        "product_number": data["product_number"],
        "price": data["price"],
        "color": data.get("color"),
        "size": data.get("size"),
        "category_id": data["category_id"]
    })

    row = cursor.fetchone()
    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    }), 201

@app.route("/products/<int:product_id>", methods=["PUT"])
def update_product(product_id):
    """Update an existing product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    updates = []
    params = {"id": product_id}

    for field in ("name", "product_number", "price", "color", "category_id"):
        if field in data:
            col = {"name": "Name", "product_number": "ProductNumber",
                   "price": "ListPrice", "color": "Color",
                   "category_id": "ProductCategoryID"}[field]
            updates.append(f"{col} = %({field})s")
            params[field] = data[field]

    if not updates:
        abort(400)

    cursor.execute(f"""
        UPDATE SalesLT.Product SET {', '.join(updates)}
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        WHERE ProductID = %(id)s
    """, params)

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products/<int:product_id>", methods=["DELETE"])
def delete_product(product_id):
    """Delete a product."""
    cursor = get_db()
    cursor.execute("DELETE FROM SalesLT.Product WHERE ProductID = %(id)s", {"id": product_id})
    if cursor.rowcount == 0:
        abort(404)
    return "", 204

@app.route("/health")
def health_check():
    """Check database connectivity."""
    try:
        cursor = get_db()
        cursor.execute("SELECT 1")
        return jsonify({"status": "healthy", "database": "connected"})
    except Exception as e:
        return jsonify({"status": "unhealthy", "error": str(e)}), 503

Запуск приложения

Запустите сервер разработки:

flask --app app run --debug --port 5000

Сервер прослушивает http://localhost:5000. Откройте второй терминал и обратитесь к конечным точкам с помощью curl, чтобы убедиться, что приложение взаимодействует с вашей базой данных:

# Check database connectivity
curl http://localhost:5000/health

# List the first page of products
curl "http://localhost:5000/products?page_size=5"

# Get a single product by ID
curl http://localhost:5000/products/680

Замечание

В PowerShell curl — это псевдоним для Invoke-WebRequest. Простые команды GET здесь работают нормально, но ответ возвращается в виде объекта, а не печатного JSON. Команды, использующие curl флаги вроде -X, -H, или -d (как в POST следующем примере), работают не так, как написано. В Windows curl.exe используйте для выполнения команд точно так, как показано, или используйте PowerShell Invoke-RestMethod (напримерInvoke-RestMethod http://localhost:5000/health), который также анализирует ответ JSON.

Каждая конечная точка возвращает JSON. Вы также можете открыть http://localhost:5000/products в браузере, чтобы просмотреть список с разбивкой на страницы.

Пулинг соединений

Без пула соединений каждый запрос открывает и закрывает TCP-соединение с Microsoft SQL, что добавляет задержку. Механизм пула соединений поддерживает набор незанятых соединений, готовых к повторному использованию. Чтобы включить использование пула соединений, вызовите mssql_python.pooling() один раз на уровне модуля. При включённом пуле conn.close()close_db при разборе соединение возвращается к пулу, а не закрывает его.

Включение пула подключений

Включите пулинг, вызывая mssql_python.pooling() на уровне модуля до открытия любых соединений:

# database.py with connection pooling
import mssql_python
from flask import g, current_app

# Configure pool at module level
mssql_python.pooling(max_size=20, idle_timeout=300)

def get_db():
    """Get a database cursor with connection pooling."""
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

Обработка ошибок

Flask позволяет регистрировать обработчики для определённых типов исключений. Перехват mssql_python.DatabaseError и mssql_python.IntegrityError позволяет возвращать структурированные JSON-ответы об ошибках вместо стандартных HTML-страниц ошибок.

Зарегистрировать обработчики ошибок

Добавьте эти обработчики в существующий app.py после строки app = Flask(__name__). Поскольку обработчики ссылаются на app объект, они должны появиться после создания приложения. app.py требуется import mssql_python вверху. Обработчики возвращают структурированные JSON-ответы вместо стандартных страниц ошибок в HTML:

# app.py
import mssql_python

@app.errorhandler(mssql_python.DatabaseError)
def handle_database_error(error):
    """Handle database errors."""
    return jsonify({"error": "Database error occurred"}), 500

@app.errorhandler(mssql_python.IntegrityError)
def handle_integrity_error(error):
    """Handle integrity constraint violations."""
    error_msg = str(error)
    if "UNIQUE" in error_msg:
        return jsonify({"error": "Resource already exists"}), 409
    if "FOREIGN KEY" in error_msg:
        return jsonify({"error": "Referenced resource not found"}), 400
    return jsonify({"error": "Data integrity error"}), 400

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

@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": "Bad request"}), 400

Blueprints

По мере роста приложения поддерживать все маршруты в одном файле становится сложно. Flask Blueprints позволяет группировать связанные маршруты в отдельные модули, зарегистрированные в приложении.

Организуйте маршруты с помощью чертежей

Создайте модуль чертежа для маршрутов продукта, который импортирует get_db и определяет конечные точки под общим префиксом URL:

# blueprints/products.py
from flask import Blueprint, jsonify, request, abort
from database import get_db

products_bp = Blueprint("products", __name__, url_prefix="/api/products")

@products_bp.route("/")
def list_products():
    """List all products."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product ORDER BY ProductID
    """)
    return jsonify([{
        "id": row.ProductID,
        "name": row.Name,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()])

@products_bp.route("/<int:product_id>")
def get_product(product_id):
    """Get a product by ID."""
    cursor = get_db()
    cursor.execute(
        "SELECT ProductID, Name, ListPrice, Color FROM SalesLT.Product WHERE ProductID = %(id)s",
        {"id": product_id}
    )
    row = cursor.fetchone()
    if not row:
        abort(404)
    return jsonify({"id": row.ProductID, "name": row.Name, "price": float(row.ListPrice), "color": row.Color})

Зарегистрируйте чертёж

Сохраните чертеж как blueprints/products.py, и добавьте пустой blueprints/__init__.py файл, чтобы Python рассматривал папку как пакет. Затем в app.py импортируйте план вместе с другими инструкциями импорта и зарегистрируйте его после строки app = Flask(__name__):

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Поскольку чертеж устанавливает url_prefix="/api/products", его маршруты обслуживаются под этим префиксом. Например, список маршрутов доступен по http://localhost:5000/api/products/адресу , отдельно от /products маршрутов, определённых непосредственно в app.py.

Testing

Flask предоставляет тестовый клиент, который отправляет запросы в ваше приложение без запуска настоящего HTTP-сервера. Используйте pytest фикстуры для создания клиента и повторного использования его в разных тестах.

Настройка тестов с pytest

Создайте фикстуру pytest, которая предоставляет тестовый клиент, и напишите тесты для проверки поведения маршрута:

# test_app.py
import uuid

import pytest
from app import app

@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as client:
        yield client

def test_health_check(client):
    response = client.get("/health")
    assert response.status_code == 200
    data = response.get_json()
    assert data["status"] == "healthy"

def test_list_products(client):
    response = client.get("/products")
    assert response.status_code == 200
    data = response.get_json()
    assert "items" in data
    assert "total" in data

def test_create_product(client):
    suffix = uuid.uuid4().hex[:8]
    name = f"Test Product {suffix}"
    response = client.post("/products", json={
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "category_id": 18
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data["name"] == name

def test_get_product_not_found(client):
    response = client.get("/products/99999")
    assert response.status_code == 404

Эти тесты выполняются непосредственно в вашей рабочей базе данных, а не с моками, поэтому test_create_product вставляет реальную строку в SalesLT.Product. В AdventureWorksLT оба NameProductNumber имеют уникальные ограничения, поэтому тест генерирует уникальное значение для каждого при каждом запуске. Если вместо этого жёстко прописать эти значения, тест завершится с конфликтом при повторном запуске, если сначала не удалить эту строку.

Выполнение тестов

Сохраняйте тесты в test_app.py папке проекта. С активированной виртуальной средой установите pytest и запустите её из этой папки. Установка и запуск pytest в той же виртуальной среде и flaskmssql-python гарантирует, что тесты импортируют пакеты, которые использует ваше приложение. pytest Автоматически test_app.py обнаруживает и сообщает о результатах:

pip install pytest
pytest

pytest автоматически обнаруживает test_app.py и сообщает о результатах:

==================== test session starts ====================
collected 4 items

test_app.py ....                                       [100%]

===================== 4 passed in 3.21s =====================

Связанные материалы

  • Last updated on