Usa mssql-python con Flask

Flask è un framework web Python leggero che ti dà il pieno controllo sulla struttura dell'applicazione. Combinato con mssql-python, puoi costruire applicazioni web e API REST supportate da Microsoft SQL e database SQL di Azure con un overhead minimo.

Prerequisiti

• Python 3.10 o versione successiva.
• Pacchetti mssql-python e flask. Installa entrambi con pip install flask mssql-python.
• Installare prerequisiti specifici del sistema operativo monouso. Gli utenti Windows possono saltare questo passaggio. Per dettagli completi sulla piattaforma, vedi Installa mssql-python.
  • Alpino
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Creare un database SQL

Crea o collegati a un database SQL su una delle seguenti piattaforme:

Gli esempi in questo articolo utilizzano il database di esempio AdventureWorksLT , in particolare la SalesLT.Product tabella. Se non hai installato AdventureWorksLT, consulta i database di esempio di AdventureWorks.

Configurazione del progetto

Installa le dipendenze

Installa i pacchetti necessari con pip:

pip install flask mssql-python

Struttura del progetto

Organizza il tuo progetto con moduli separati per configurazione, gestione delle connessioni, rotte e test:

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

Gestione delle connessioni al database

Flask non include un livello database integrato, quindi gestisci direttamente le connessioni. Il pattern in questa sezione memorizza una connessione per ogni richiesta sull'oggetto di g Flask e la chiude automaticamente quando la richiesta termina.

Crea config.py

Centralizza le impostazioni del database in una classe di configurazione. Le variabili di ambiente ti permettono di sovrascrivere i valori predefiniti senza cambiare codice.

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

Crea database.py

Il database.py modulo gestisce il ciclo di vita della connessione. L'oggetto di g Flask è uno spazio per richiesta, quindi memorizzare la connessione lì garantisce che ogni richiesta riceva la propria connessione che viene ripulita quando la richiesta termina.

La get_connection_string() funzione costruisce la stringa di connessione dalla configurazione dell'app. La get_db() funzione crea una connessione alla prima chiamata e la riutilizza per il resto della richiesta. La funzione close_db() viene eseguita automaticamente alla fine di ogni richiesta, annullando la transazione se si verifica un'eccezione e confermandola in caso contrario. La init_app() funzione registra questo comportamento di teardown con l'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)

Applicazione Flask

Il seguente esempio mostra un'applicazione Flask completa con percorsi per elencare, recuperare, creare, aggiornare e cancellare prodotti.

Crea app.py

Il modulo applicativo crea l'app Flask, carica la configurazione e registra il dismantellamento del database. Ogni funzione di routing chiama get_db() per ottenere un cursore, esegue query con SQL parametrizzato (utilizzando i segnaposto %(name)s e un dizionario di valori) e restituisce risposte 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

Eseguire l'applicazione

Avviare il server di sviluppo:

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

Il server ascolta su http://localhost:5000. Apri un secondo terminale e chiama gli endpoint utilizzando curl per confermare che l'app si stia connettendo al database:

# 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

Ogni endpoint restituisce JSON. Puoi anche aprire http://localhost:5000/products in un browser per visualizzare la lista paginata.

Pool di connessioni

Senza il pool di connessioni, ogni richiesta apre e chiude una connessione TCP a Microsoft SQL, che aggiunge latenza. Il pool di connessione mantiene un insieme di connessioni inattive pronte per il riutilizzo. Per abilitare il pooling delle connessioni, chiama mssql_python.pooling() una sola volta a livello di modulo. Con il pooling attivato, conn.close() nel close_db teardown la connessione viene restituita al pool invece di chiuderlo.

Abilitare il pool di connessioni

Abilita il pooling chiamando mssql_python.pooling() a livello di modulo prima che vengano aperte eventuali connessioni:

# 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

Gestione degli errori

Flask ti permette di registrare i gestori per tipi specifici di eccezione. Intercettare mssql_python.DatabaseError e mssql_python.IntegrityError consente di restituire risposte di errore JSON strutturate invece delle pagine di errore HTML predefinite.

Registrare i gestori degli errori

Aggiungi questi handler all'esistente app.py, dopo la app = Flask(__name__) linea. Poiché gli handler fanno riferimento all'oggetto app, devono venire dopo che l'app è stata creata. app.py richiede import mssql_python in alto. I gestori restituiscono risposte JSON strutturate invece delle pagine di errore HTML predefinite:

# 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

Man mano che la tua applicazione cresce, mettere tutte le route in un unico file diventa difficile da mantenere. Flask Blueprints ti permette di raggruppare percorsi correlati in moduli separati registrati nell'app.

Organizza i percorsi con i modelli

Crea un modulo blueprint per le rotte di prodotto che importi get_db e definisce endpoint sotto un prefisso URL condiviso:

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

Registra il progetto

Salva il blueprint come blueprints/products.py, e aggiungi un file vuoto blueprints/__init__.py così Python tratta la cartella come un pacchetto. Poi, in app.py, importa il blueprint insieme alle altre importazioni e registralo dopo la riga app = Flask(__name__):

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Poiché il blueprint imposta url_prefix="/api/products", le sue rotte vengono esposte sotto quel prefisso. Ad esempio, la rotta elenco è disponibile in http://localhost:5000/api/products/, separata dalle rotte /products definite direttamente in app.py.

Testing

Flask fornisce un client di test che invia richieste alla tua applicazione senza avviare un vero server HTTP. Usa pytest i fixture per creare il client e riutilizzalo tra i test.

Configurazione del test con pytest

Crea un fix pytest che fornisca un client di test e scrivi test per verificare il comportamento delle route:

# 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

Questi test si eseguono sul tuo database live invece che sulle mock, quindi test_create_product inserisce una riga reale in SalesLT.Product. In AdventureWorksLT, sia Name che ProductNumber hanno vincoli unici, quindi il test genera un valore unico per ciascuno a ogni esecuzione. Se invece codifica quei valori in modo rigido, il test fallisce con un conflitto alla seconda esecuzione, a meno che tu non elimini prima la riga.

Esegui i test

Salva i test come test_app.py nella cartella del progetto. Con il tuo ambiente virtuale attivato, installalo pytest ed eseguilo da quella cartella. Installare e girare pytest all'interno dello stesso ambiente virtuale di flask e mssql-python garantisce che i test importino i pacchetti che la tua app utilizza. pytest scopre test_app.py e riporta automaticamente i risultati:

pip install pytest
pytest

pytest scopre test_app.py automaticamente e riporta i risultati:

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

test_app.py ....                                       [100%]

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

Contenuti correlati

• Gestione delle connessioni
• Pool di connessioni
• Gestione degli errori
• Integrazione FastAPI
• Autenticazione Microsoft Entra