Używaj mssql-python z Flaskiem

Flask to lekki framework Python, który daje pełną kontrolę nad strukturą aplikacji. W połączeniu z mssql-python możesz tworzyć aplikacje internetowe i interfejsy API REST oparte na Microsoft SQL i Azure SQL Database, przy minimalnym narzucie.

Wymagania wstępne

  • Python 3.10 lub nowszy.
  • Pakiety mssql-python i flask. Zainstaluj oba za pomocą pip install flask mssql-python.
  • Zainstaluj jednorazowe wymagania wstępne dotyczące systemu operacyjnego. Użytkownicy Windows mogą pominąć ten krok. Pełne szczegóły dotyczące platformy można znaleźć w artykule Install mssql-python.
    apk add libtool krb5-libs krb5-dev

Tworzenie bazy danych SQL

Stwórz lub połącz się z bazą danych SQL na jednej z następujących platform:

Przykłady w tym artykule korzystają z przykładowej bazy danych AdventureWorksLT , a konkretnie z tabeli SalesLT.Product . Jeśli nie masz zainstalowanego AdventureWorksLT, zobacz przykładowe bazy danych AdventureWorks.

Konfiguracja projektu

Instalowanie zależności

Zainstaluj wymagane pakiety za pomocą pip:

pip install flask mssql-python

struktura projektu

Zorganizuj swój projekt z osobnymi modułami do konfiguracji, zarządzania połączeniami, tras i testów:

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

Zarządzanie połączeniami bazy danych

Flask nie ma wbudowanej warstwy bazy danych, więc połączenia zarządzasz bezpośrednio. Wzorzec w tej sekcji przechowuje jedno połączenie dla każdego żądania na obiekcie g Flask i automatycznie zamyka je po zakończeniu żądania.

Stwórz config.py

Scentralizuj ustawienia bazy danych w klasie konfiguracyjnej. Zmienne środowiskowe pozwalają nadpisać domyślne ustawienia bez zmiany kodu.

# 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"))

Stwórz database.py

Moduł database.py zarządza cyklem życia połączenia. Obiekt g we Flasku jest przestrzenią nazw przypisaną do pojedynczego żądania, więc przechowywanie tam połączenia zapewnia, że każde żądanie otrzymuje własne połączenie, które zostanie zamknięte po zakończeniu obsługi żądania.

Funkcja get_connection_string() buduje parametry połączenia z konfiguracji aplikacji. Funkcja get_db() tworzy połączenie przy pierwszym wywołaniu i wykorzystuje je ponownie do reszty żądania. Funkcja close_db() jest uruchamiana automatycznie na końcu każdego żądania, wycofując transakcję w przypadku wystąpienia wyjątku, a w przeciwnym razie ją zatwierdzając. Funkcja init_app() rejestruje tę obsługę czyszczenia w aplikacji 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)

Note

ActiveDirectoryDefault używa DefaultAzureCredential, który testuje kolejno wielu dostawców poświadczeń. Pierwsze połączenie może być wolne, ponieważ SDK sprawdza kolejnych dostawców w łańcuchu, aż znajdzie takiego, który działa poprawnie. W produkcji, jeśli wiesz, jakiego typu poświadczenia używa twoje środowisko, określ go bezpośrednio (na przykład ActiveDirectoryMSI dla tożsamości zarządzanej), aby uniknąć tzw. chain walk. Aby uzyskać więcej informacji, zobacz Microsoft Entra authentication (Uwierzytelnianie w usłudze Microsoft Entra).

Aplikacja Flask

Poniższy przykład pokazuje kompletną aplikację Flask z trasami do listowania, pobierania, tworzenia, aktualizacji i usuwania produktów.

Stwórz app.py

Moduł aplikacyjny tworzy aplikację Flesk, ładuje konfigurację i rejestruje rozbiórkę bazy danych. Każda funkcja trasy wywołuje get_db(), aby uzyskać kursor, wykonuje zapytania za pomocą sparametryzowanego języka SQL (używając %(name)s jako symboli zastępczych oraz słownika wartości) i zwraca odpowiedzi 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

Uruchamianie aplikacji

Uruchom serwer deweloperów:

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

Serwer nasłuchuje na http://localhost:5000. Otwórz drugi terminal i wywołaj punkty końcowe za pomocą curl, aby potwierdzić, że aplikacja komunikuje się z bazą danych:

# 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

Note

W programie PowerShell curl jest aliasem dla programu Invoke-WebRequest. Proste polecenia GET działają tu bez problemu, ale odpowiedź wraca jako obiekt, a nie jako wydrukowany JSON. Polecenia wykorzystujące curl flagi takie jak -X, -H, lub -d (jak w POST późniejszym przykładzie) nie działają zgodnie z napisanym wzorem. W systemie Windows użyj curl.exe, aby uruchamiać polecenia dokładnie tak, jak pokazano, albo użyj polecenia cmdlet Invoke-RestMethod programu PowerShell (na przykład Invoke-RestMethod http://localhost:5000/health), które automatycznie przetwarza odpowiedź JSON.

Każdy punkt końcowy zwraca JSON. Możesz też otworzyć http://localhost:5000/products w przeglądarce, aby zobaczyć listę paginowaną.

Buforowanie połączeń

Bez puli połączeń każde żądanie otwiera i zamyka połączenie TCP z Microsoft SQL, co zwiększa opóźnienia. Pula połączeń utrzymuje zestaw bezczynnych połączeń gotowych do ponownego użycia. Aby włączyć pulę połączeń, wywołaj mssql_python.pooling() raz na poziomie modułu. Po włączonym poolingu, conn.close() podczas close_db rozbiórki połączenie wraca do puli zamiast go zamykać.

Włącz buforowanie połączeń

Włącz pulowanie, wywołując mssql_python.pooling() na poziomie modułu przed otwarciem jakichkolwiek połączeń:

# 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

Obsługa błędów

Flask umożliwia rejestrowanie programów obsługi dla określonych typów wyjątków. Przechwytywanie mssql_python.DatabaseError i mssql_python.IntegrityError umożliwia zwracanie ustrukturyzowanych odpowiedzi błędów w formacie JSON zamiast domyślnych stron błędów HTML.

Zarejestruj procedury obsługi błędów

Dodaj te procedury obsługi do istniejącego app.py, po wierszu app = Flask(__name__). Ponieważ procedury obsługi odwołują się do obiektu app, muszą znajdować się dopiero po utworzeniu aplikacji. app.py potrzebuje import mssql_python na górze. Handlery zwracają ustrukturyzowane odpowiedzi JSON zamiast domyślnych stron błędów 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

W miarę rozwoju aplikacji przechowywanie wszystkich tras w jednym pliku staje się trudne do utrzymania. Flask Blueprints pozwala grupować powiązane trasy w osobne moduły zarejestrowane w aplikacji.

Organizuj trasy za pomocą planów

Stwórz moduł blueprint dla tras produktów, który importuje get_db i definiuje punkty końcowe pod wspólnym prefiksem 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})

Zarejestruj plan

Zapisz blueprint jako blueprints/products.py, i dodaj pusty blueprints/__init__.py plik, aby Python traktował folder jak pakiet. Następnie w app.py, zaimportuj blueprint wraz z innymi importami i zarejestruj go po linii app = Flask(__name__) :

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Ponieważ blueprint ustawia url_prefix="/api/products", jego trasy są udostępniane pod tym prefiksem. Na przykład trasa listy jest dostępna pod adresem http://localhost:5000/api/products/, oddzielnie od tras /products, zdefiniowanych bezpośrednio w app.py.

Testing

Flask udostępnia klienta testowego, który wysyła żądania do Twojej aplikacji bez uruchamiania prawdziwego serwera HTTP. Użyj pytest fixtures do tworzenia klienta i ponownego wykorzystania go w testach.

Konfiguracja testu z pytest

Stwórz zestaw pytest, który udostępni klienta testowego i napisz testy do weryfikacji zachowania trasy:

# 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

Te testy działają na działającej bazie danych, a nie na mockach, więc test_create_product wstawia prawdziwy wiersz do SalesLT.Product. W AdventureWorksLT zarówno Name, jak i ProductNumber mają ograniczenia unikalności, więc test generuje dla każdego z nich unikalną wartość przy każdym uruchomieniu. Jeśli zamiast tego zakodujesz te wartości na stałe, test kończy się konfliktem przy drugim uruchomieniu, chyba że najpierw usuniesz ten wiersz.

Uruchamianie testów

Zapisz testy w test_app.py folderze projektu. Z aktywowanym środowiskiem wirtualnym zainstaluj pytest i uruchom go z tego folderu. Instalacja i uruchamianie pytest w tym samym środowisku wirtualnym, co flask i mssql-python, zapewnia, że testy importują pakiety używane przez Twoją aplikację. pytest automatycznie wykrywa test_app.py i raportuje wyniki:

pip install pytest
pytest

pytest automatycznie wykrywa test_app.py i raportuje wyniki:

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

test_app.py ....                                       [100%]

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

Treści powiązane

  • Last updated on