Gebruik mssql-python met Flask

Flask is een lichtgewicht Python-webframework dat je volledige controle geeft over de applicatiestructuur. In combinatie met mssql-python kun je webapplicaties en REST API's bouwen die worden ondersteund door Microsoft SQL en Azure SQL Database met minimale overhead.

Prerequisites

  • Python 3.10 of hoger.
  • De pakketten mssql-python en flask. Installeer beide met pip install flask mssql-python.
  • Installeer eenmalige vereisten voor het besturingssysteem. Windows-gebruikers kunnen deze stap overslaan. Voor volledige platformdetails, zie Install mssql-python.
    apk add libtool krb5-libs krb5-dev

Een SQL-database maken

Maak een SQL-database aan of maak verbinding met een van de volgende platforms:

De voorbeelden in dit artikel gebruiken de voorbeelddatabase van AdventureWorksLT , specifiek de SalesLT.Product tabel. Als je AdventureWorksLT niet hebt geïnstalleerd, zie dan de voorbeelddatabases van AdventureWorks.

Projectopstelling

Afhankelijkheden installeren

Installeer de benodigde pakketten met pip:

pip install flask mssql-python

Projectstructuur

Organiseer je project met aparte modules voor configuratie, verbindingsbeheer, routes en 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

Databaseverbindingsbeheer

Flask heeft geen ingebouwde databaselaag, dus je beheert verbindingen direct. Het patroon in deze sectie slaat één verbinding per verzoek op op het g object van Flask en sluit het automatisch wanneer het verzoek eindigt.

Maak het bestand config.py

Centraliseer database-instellingen in een configuratieklasse. Omgevingsvariabelen laten je defaults overschrijven zonder de code te wijzigen.

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

Maak het bestand database.py

De database.py module beheert de levenscyclus van de verbinding. Het g object van Flask is een naamruimte per verzoek, dus het opslaan van de verbinding daar zorgt ervoor dat elk verzoek zijn eigen verbinding krijgt die wordt opgeschoond zodra het verzoek is afgerond.

De get_connection_string() functie bouwt de verbindingsreeks op vanuit de app-configuratie. De get_db() functie maakt een verbinding aan bij het eerste gesprek en gebruikt deze opnieuw voor de rest van het verzoek. De functie close_db() wordt automatisch uitgevoerd aan het einde van elk verzoek, waarbij de transactie wordt teruggedraaid als er een uitzondering is opgetreden en anders wordt gecommit. De init_app() functie registreert dit demontagegedrag met de 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)

Opmerking

ActiveDirectoryDefault gebruikt DefaultAzureCredential, dat meerdere credentialproviders achter elkaar probeert. De eerste verbinding kan traag zijn omdat de SDK de keten doorloopt totdat hij een werkende provider vindt. In productie, als je weet welk type inloggegevens je omgeving gebruikt, specificeer het dan direct (bijvoorbeeld ActiveDirectoryMSI voor managed identity) om de chain walk te voorkomen. Zie Microsoft Entra-verificatie voor meer informatie.

Flask-applicatie

Het volgende voorbeeld toont een complete Flask-applicatie met routes voor het vermelden, ophalen, aanmaken, bijwerken en verwijderen van producten.

Maak het bestand app.py

De applicatiemodule maakt de Flask-app aan, laadt de configuratie en registreert de databasedemontage. Elke routefunctie roept get_db() een cursor aan, voert queries uit met geparametriseerde SQL (met behulp van %(name)s placeholders en een woordenboek van waarden), en geeft JSON-antwoorden terug.

# 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

De toepassing uitvoeren

Start de ontwikkelserver:

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

De server luistert op http://localhost:5000. Open een tweede terminal en roep de eindpunten aan met curl om te bevestigen dat de app met je database communiceert:

# 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

Opmerking

In PowerShell curl is dit een alias voor Invoke-WebRequest. De eenvoudige GET-commando's hier werken prima, maar de respons komt terug als een object in plaats van als geprinte JSON. Commando's die vlaggen gebruiken curl zoals -X, -H, of -d (zoals het POST voorbeeld later) werken niet zoals geschreven. Op Windows kun je curl.exe gebruiken om de opdrachten precies uit te voeren zoals getoond, of de Invoke-RestMethod van PowerShell gebruiken (bijvoorbeeld Invoke-RestMethod http://localhost:5000/health), die ook de JSON-respons voor je parseert.

Elk eindpunt geeft JSON terug. Je kunt ook in een browser de http://localhost:5000/products gepagineerde lijst bekijken.

Groepsgewijze verbindingen

Zonder connection pooling opent en sluit elk verzoek een TCP-verbinding met Microsoft SQL, wat de latentie toevoegt. Verbindingspooling houdt een verzameling inactieve verbindingen gereed voor hergebruik. Om verbindingspooling in te schakelen, roept u mssql_python.pooling() eenmaal aan op moduleniveau. Met ingeschakelde pooling retourneert conn.close() in de close_db teardown de verbinding naar de pool in plaats van deze te sluiten.

Groepsgewijze verbindingen inschakelen

Schakel pooling in door op moduleniveau aan te roepen mssql_python.pooling() voordat er verbindingen worden geopend:

# 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

Foutafhandeling

Flask laat je handlers registreren voor specifieke uitzonderingstypes. Catching mssql_python.DatabaseError en mssql_python.IntegrityError laat je gestructureerde JSON-foutantwoorden teruggeven in plaats van standaard HTML-foutpagina's.

Registreer foutafhandelaren

Voeg deze handlers toe aan de bestaande app.py, na de app = Flask(__name__) lijn. Omdat de handlers naar het app object verwijzen, moeten ze pas komen nadat de app is gemaakt. app.py heeft import mssql_python bovenaan nodig. De handlers geven gestructureerde JSON-antwoorden terug in plaats van standaard HTML-foutpagina's:

# 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

Naarmate je applicatie groeit, wordt het moeilijk om alle routes in één bestand te beheren. Flask Blueprints laat je gerelateerde routes groeperen in aparte modules die bij de app geregistreerd zijn.

Organiseer routes met blauwdrukken

Maak een blueprintmodule voor productroutes die eindpunten importeert get_db en definieert onder een gedeelde URL-prefix:

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

Registreer de blauwdruk

Sla de blauwdruk op als blueprints/products.py, en voeg een leeg blueprints/__init__.py bestand toe zodat Python de map als een pakket behandelt. Importeer vervolgens in app.py, de blauwdruk met je andere importen en registreer deze na de app = Flask(__name__) regel:

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Omdat de blueprint url_prefix="/api/products" instelt, zijn de routes beschikbaar onder dat voorvoegsel. Bijvoorbeeld, de lijstroute is beschikbaar op http://localhost:5000/api/products/, los van de /products routes die direct in gedefinieerd zijn app.py.

Testing

Flask biedt een testclient die verzoeken naar je applicatie stuurt zonder een echte HTTP-server te starten. Gebruik pytest fixtures om de client te maken en hergebruik deze in tests.

Testopstelling met pytest

Maak een pytest-fixture aan die een testclient levert en schrijf tests om het routegedrag te verifiëren:

# 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

Deze tests worden uitgevoerd op je live-database in plaats van op mockobjecten, dus voegt test_create_product een echte rij toe aan SalesLT.Product. In AdventureWorksLT hebben beide Name en ProductNumber unieke beperkingen, dus de test genereert bij elke run een unieke waarde voor elk. Als je die waarden hardcodeert, faalt de test met een conflict bij de tweede run, tenzij je eerst de rij verwijdert.

De tests uitvoeren

Sla de tests op als test_app.py in je projectmap. Met je virtuele omgeving geactiveerd, installeer pytest en voer het uit vanuit die map. Installeren en draaien pytest in dezelfde virtuele omgeving als flask en mssql-python zorgen ervoor dat de tests de pakketten importeren die je app gebruikt. pytest Ontdekt en rapporteert test_app.py automatisch de resultaten:

pip install pytest
pytest

pytest ontdekt test_app.py automatisch en rapporteert de resultaten:

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

test_app.py ....                                       [100%]

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

Verwante onderwerpen

  • Last updated on