Utilisez mssql-python avec Flask

Flask est un framework web Python léger qui vous donne un contrôle total sur la structure de l’application. Combiné à mssql-python, vous pouvez créer des applications web et des API REST soutenues par Microsoft SQL et Azure SQL Database avec un minimum de surcharge.

Logiciels requis

  • Python 3.10 ou version ultérieure.
  • Packages mssql-python et flask. Installez les deux avec pip install flask mssql-python.
  • Installez les prérequis ponctuels spécifiques au système d'exploitation. Les utilisateurs Windows peuvent sauter cette étape. Pour tous les détails sur la plateforme, voir Installer mssql-python.
    apk add libtool krb5-libs krb5-dev

Créer une base de données SQL

Créer ou connecter une base de données SQL sur l’une des plateformes suivantes :

Les exemples de cet article utilisent la base de données d’exemples AdventureWorksLT , en particulier la SalesLT.Product table. Si vous n’avez pas AdventureWorksLT installé, consultez les bases de données d’exemple AdventureWorks.

Configuration du projet

Installer des dépendances

Installez les packages requis avec pip :

pip install flask mssql-python

Structure du projet

Organisez votre projet avec des modules séparés pour la configuration, la gestion des connexions, les itinéraires et les 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

Gestion des connexions à la base de données

Flask n’inclut pas de couche de base de données intégrée, donc vous gérez directement les connexions. Le motif de cette section stocke une connexion par requête sur l’objet de g Flask et le ferme automatiquement à la fin de la requête.

Créez config.py

Centralisez les paramètres de la base de données dans une classe de configuration. Les variables d’environnement permettent de contourner les paramètres par défaut sans changer de code.

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

Créez database.py

Le database.py module gère le cycle de vie de la connexion. L’objet g de Flask est un espace de noms propre à chaque requête ; y stocker la connexion garantit donc que chaque requête dispose de sa propre connexion, correctement libérée à la fin de la requête.

La get_connection_string() fonction construit la chaîne de connexion à partir de la configuration de l’application. La get_db() fonction crée une connexion dès le premier appel et la réutilise pour le reste de la requête. La fonction close_db() s’exécute automatiquement à la fin de chaque requête, annule la transaction en cas d’exception et la valide dans le cas contraire. La init_app() fonction détecte ce comportement de démontage avec l’application 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 utilise DefaultAzureCredential, qui tente successivement d’utiliser plusieurs fournisseurs d’informations d’identification. La première connexion peut être lente car le SDK parcourt la chaîne jusqu’à ce qu’il trouve un fournisseur fonctionnel. En production, si vous savez quel type d’identifiant votre environnement utilise, spécifiez-le directement (par exemple, ActiveDirectoryMSI pour l’identité gérée) afin d’éviter la marche en chaîne. Pour plus d’informations, consultez Authentification Microsoft Entra.

Application Flask

L’exemple suivant montre une application Flask complète avec des itinéraires pour lister, récupérer, créer, mettre à jour et supprimer des produits.

Créez app.py

Le module application crée l’application Flask, charge la configuration et enregistre le démantèlement de la base de données. Chaque fonction de routage appelle get_db() pour obtenir un curseur, exécute des requêtes avec du SQL paramétré (en utilisant les espaces réservés %(name)s et un dictionnaire de valeurs), et renvoie des réponses 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

Exécuter l’application

Démarrez le serveur de développement :

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

Le serveur écoute sur http://localhost:5000. Ouvrez un second terminal et appelez les points de terminaison en utilisant curl pour confirmer que l’application communique avec votre base de données :

# 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

Dans PowerShell, curl est un alias pour Invoke-WebRequest. Les commandes GET simples fonctionnent bien ici, mais la réponse revient sous forme d’objet plutôt qu’en JSON imprimé. Les commandes qui utilisent des options comme curl, -X, -H ou -d (comme l’exemple POST plus loin) ne fonctionnent pas telles quelles. Sous Windows, utilisez curl.exe pour exécuter les commandes exactement comme indiquées, ou utilisez celles de Invoke-RestMethod PowerShell (par exemple, Invoke-RestMethod http://localhost:5000/health), qui analysent également la réponse JSON pour vous.

Chaque terminaison retourne du JSON. Vous pouvez aussi ouvrir http://localhost:5000/products dans un navigateur pour consulter la liste paginationnée.

Regroupement de connexions

Sans pooling de connexions, chaque requête ouvre et ferme une connexion TCP vers Microsoft SQL, ce qui ajoute de la latence. Le pooling de connexions maintient un ensemble de connexions inactives prêtes à être réutilisées. Pour activer le pool de connexions, appelez mssql_python.pooling() une fois au niveau du module. Avec le pooling activé, conn.close() dans la phase de démontage de close_db renvoie la connexion au pool au lieu de la fermer.

Activer le regroupement de connexions

Activez le pooling en appelant mssql_python.pooling() au niveau du module avant l’ouverture de toute connexion :

# 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

Gestion des erreurs

Flask permet d’enregistrer des gestionnaires pour des types d’exceptions spécifiques. En interceptant mssql_python.DatabaseError et mssql_python.IntegrityError, vous pouvez renvoyer des réponses d’erreur JSON structurées au lieu des pages d’erreur HTML par défaut.

Enregistrer des gestionnaires d’erreurs

Ajoutez ces manipulateurs à l’existant app.py, après la app = Flask(__name__) ligne. Étant donné que les gestionnaires font référence à l’objet app, ils doivent être définis après la création de l’application. app.py nécessite import mssql_python en haut. Les gestionnaires renvoient des réponses JSON structurées au lieu de pages d’erreur HTML par défaut :

# 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

À mesure que votre application grandit, il devient difficile de maintenir toutes les routes dans un seul fichier. Flask Blueprints permet de regrouper les routes liées en modules distincts enregistrés dans l’application.

Organiser les itinéraires avec des plans

Créez un module blueprint pour les routes de produit qui importe get_db et définit les points de terminaison sous un préfixe URL partagé :

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

Inscrivez le plan

Sauvegardez le blueprint sous forme blueprints/products.pyde , et ajoutez un fichier vide blueprints/__init__.py pour que Python traite le dossier comme un paquet. Ensuite, dans app.py, importez le blueprint avec vos autres importations et enregistrez-le après la app = Flask(__name__) ligne :

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Comme le blueprint définit url_prefix="/api/products", ses routes sont accessibles sous ce préfixe d’URL. Par exemple, la route de liste est disponible à http://localhost:5000/api/products/, distincte des /products routes définies directement dans app.py.

Testing

Flask fournit un client de test qui envoie des requêtes à votre application sans lancer un véritable serveur HTTP. Utilisez pytest fixtures pour créer le client et le réutiliser dans l’ensemble des tests.

Configuration du test avec pytest

Créer un accessoire pytest qui fournit un client de test et écrire des tests pour vérifier le comportement des routes :

# 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

Ces tests s’exécutent sur votre base de données en direct plutôt que sur des mocks, donc test_create_product insère une vraie ligne dans SalesLT.Product. Dans AdventureWorksLT, Name et ProductNumber ont des contraintes uniques, donc le test génère une valeur unique pour chacun à chaque exécution. Si vous codez ces valeurs en dur à la place, le test échoue avec un conflit lors de la deuxième exécution, sauf si vous supprimez d’abord la ligne.

Exécuter les tests

Enregistrez les tests sous test_app.py dans le dossier de votre projet. Avec votre environnement virtuel activé, installez-le pytest et exécutez-le depuis ce dossier. Installer et exécuter pytest dans le même environnement virtuel que flask et mssql-python garantit que les tests importent les paquets utilisés par votre application. pytest découvre test_app.py et rapporte automatiquement les résultats :

pip install pytest
pytest

pytest découvre test_app.py automatiquement et rapporte les résultats :

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

test_app.py ....                                       [100%]

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

Contenu connexe

  • Last updated on