Verwenden Sie mssql-python mit Flask

Flask ist ein leichtes Python-Webframework, das dir volle Kontrolle über die Anwendungsstruktur gibt. In Kombination mit mssql-python kann man Webanwendungen und REST-APIs mit minimalem Aufwand auf Microsoft SQL und Azure SQL-Datenbank bauen.

Voraussetzungen

• Python 3.10 oder höher.
• Die Pakete mssql-python und flask. Installiere beide mit pip install flask mssql-python.
• Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen. Windows-Nutzer können diesen Schritt überspringen. Für vollständige Plattformdetails siehe Install mssql-python.
  • Alpin
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Erstellen einer SQL-Datenbank

Erstellen oder verbinden Sie sich mit einer SQL-Datenbank auf einer der folgenden Plattformen:

Die Beispiele in diesem Artikel verwenden die AdventureWorksLT-Beispieldatenbank , speziell die Tabelle SalesLT.Product . Wenn Sie AdventureWorksLT nicht installiert haben, sehen Sie sich die AdventureWorks-Beispieldatenbanken an.

Projektkonfiguration

Abhängigkeiten installieren

Installiere die erforderlichen Pakete mit pip:

pip install flask mssql-python

Projektstruktur

Organisieren Sie Ihr Projekt mit separaten Modulen für Konfiguration, Verbindungsmanagement, Routen und Tests:

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

Datenbank-Verbindungsverwaltung

Flask hat keine integrierte Datenbankschicht, sodass man Verbindungen direkt verwaltet. Das Muster in diesem Abschnitt speichert pro Anfrage eine Verbindung auf Flasks g Objekt und schließt sie automatisch, wenn die Anfrage endet.

Erstellen Sie config.py

Datenbankeinstellungen in einer Konfigurationsklasse zentralisieren. Umgebungsvariablen erlauben es, Standardwerte zu überschreiben, ohne den Code zu ändern.

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

Erstellen Sie database.py

Das Modul database.py verwaltet den Verbindungslebenszyklus. Flasks g-Objekt ist ein anforderungsbezogener Namespace, daher stellt das Speichern der Verbindung dort sicher, dass jede Anfrage ihre eigene Verbindung erhält, die nach Abschluss der Anfrage aufgeräumt wird.

Die get_connection_string() Funktion erstellt den Verbindungszeichenfolge aus der App-Konfiguration. Die get_db() Funktion stellt beim ersten Anruf eine Verbindung her und verwendet sie für den Rest der Anfrage wieder. Die close_db()-Funktion wird am Ende jeder Anfrage automatisch ausgeführt, macht die Transaktion rückgängig, wenn eine Ausnahme aufgetreten ist, und schreibt sie andernfalls fest. Die init_app() Funktion registriert dieses Teardown-Verhalten in der Flask-App.

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

Flask-Anwendung

Das folgende Beispiel zeigt eine vollständige Flask-Anwendung mit Routen zum Auflisten, Abrufen, Erstellen, Aktualisieren und Löschen von Produkten.

Erstellen Sie app.py

Das Anwendungsmodul erstellt die Flask-App, lädt die Konfiguration und registriert die Datenbankzerlegung. Jede Routenfunktion ruft einen Cursor auf, get_db() führt Abfragen mit parametrisiertem SQL aus (unter Verwendung %(name)s von Platzhaltern und einem Wertewörterbuch) und gibt JSON-Antworten zurück.

# 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

Ausführen der Anwendung

Starten Sie den Entwicklungsserver:

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

Der Server lauscht auf http://localhost:5000. Öffnen Sie ein zweites Terminal und rufen Sie die Endpunkte mithilfe von curl auf, um zu bestätigen, dass die App mit Ihrer Datenbank kommuniziert:

# 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

Jeder Endpunkt gibt JSON zurück. Du kannst auch im Browser die http://localhost:5000/products paginierte Liste öffnen.

Verbindungspooling

Ohne Connection Pooling öffnet und schließt jede Anfrage eine TCP-Verbindung zu Microsoft SQL, was die Latenz erhöht. Connection Pooling hält eine Reihe von Leerlaufverbindungen bereit zur Wiederverwendung. Um Connection-Pooling zu aktivieren, rufen Sie mssql_python.pooling() einmal auf Modulebene auf. Wenn Pooling aktiviert ist, gibt conn.close() beim close_db-Teardown die Verbindung an den Pool zurück, anstatt sie zu schließen.

Aktivieren von Verbindungspooling

Pooling aktivieren, indem Sie auf Modulebene aufrufen mssql_python.pooling() , bevor Verbindungen geöffnet werden:

# 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

Fehlerbehandlung

Flask ermöglicht es dir, Handler für bestimmte Ausnahmetypen zu registrieren. Das Abfangen von mssql_python.DatabaseError und mssql_python.IntegrityError ermöglicht es Ihnen, strukturierte JSON-Fehlerantworten anstelle der standardmäßigen HTML-Fehlerseiten zurückzugeben.

Fehlerhandler registrieren

Füge diese Handler nach der Zeile app.py zum bestehenden app = Flask(__name__)hinzu. Da die Handler auf das app-Objekt verweisen, müssen sie erst nach der Erstellung der App definiert werden. app.py benötigt import mssql_python oben. Die Handler geben strukturierte JSON-Antworten statt Standard-HTML-Fehlerseiten zurück:

# 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

Mit wachsender Anwendung wird es schwierig, alle Routen in einer einzigen Datei zu verwalten. Flask Blueprints ermöglicht es, verwandte Routen in separate Module zu gruppieren, die in der App registriert sind.

Routen mit Bauplänen organisieren

Erstellen Sie ein Blueprint-Modul für Produktrouten, das get_db importiert und Endpunkte unter einem gemeinsamen URL-Präfix definiert:

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

Registrieren Sie den Bauplan

Speichere den Blueprint als blueprints/products.py, und füge eine leere blueprints/__init__.py Datei hinzu, damit Python den Ordner als Paket behandelt. Importiere dann in app.py, den Bauplan zusammen mit deinen anderen Importen und registriere ihn nach der Zeile app = Flask(__name__) :

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Da das Blueprint auf url_prefix="/api/products" gesetzt ist, werden seine Routen unter diesem Präfix bereitgestellt. Zum Beispiel ist die Listenroute bei http://localhost:5000/api/products/verfügbar, getrennt von den direkt in /productsdefinierten app.py Routen.

Testen

Flask stellt einen Testclient bereit, der Anfragen an Ihre Anwendung sendet, ohne einen echten HTTP-Server starten zu müssen. Nutze pytest Fixtures, um den Client zu erstellen und ihn in verschiedenen Tests wiederzuverwenden.

Testaufbau mit pytest

Erstellen Sie eine pytest-Fixture, die einen Testclient bereitstellt, und schreiben Sie Tests, um das Routenverhalten zu überprüfen:

# 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

Diese Tests werden gegen deine Live-Datenbank und nicht gegen Mocks ausgeführt, daher fügt test_create_product eine echte Zeile in SalesLT.Product ein. In AdventureWorksLT haben beide Name und ProductNumber einzigartige Einschränkungen, sodass der Test für jeden bei jedem Durchlauf einen einzigartigen Wert erzeugt. Wenn du diese Werte stattdessen fest kodierst, scheitert der Test mit einem Konflikt beim zweiten Durchlauf, es sei denn, du löschst zuerst die Zeile.

Ausführen der Tests

Speichere die Tests als test_app.py in deinem Projektordner. Mit aktivierter virtueller Umgebung installiere pytest und führe es aus diesem Ordner aus. Die Installation und Ausführung pytest in derselben virtuellen Umgebung wie flask und mssql-python stellt sicher, dass die Tests die Pakete importieren, die Ihre App verwendet. pytest erkennt test_app.py automatisch und meldet die Ergebnisse:

pip install pytest
pytest

pytest entdeckt test_app.py automatisch und meldet die Ergebnisse:

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

test_app.py ....                                       [100%]

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

Verwandte Inhalte

• Verbindungsverwaltung
• Verbindungspooling
• Fehlerbehandlung
• FastAPI-Integration
• Microsoft Entra-Authentifizierung