Använd mssql-python med Flask

Flask är ett lättviktigt Python-webbramverk som ger dig full kontroll över applikationsstrukturen. I kombination med mssql-python kan du bygga webbapplikationer och REST-API:er med Microsoft SQL och Azure SQL Database med minimal överhead.

Förutsättningar

Python 3.10 eller senare.
paketen mssql-python och flask. Installera båda med pip install flask mssql-python.

Installera operativsystemsspecifika engångsförutsättningar. Windows-användare kan hoppa över detta steg. För fullständiga plattformsdetaljer, se Installera mssql-python.

apk add libtool krb5-libs krb5-dev

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

dnf install -y libtool-ltdl krb5-libs

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

zypper install -y libltdl7

brew install openssl

Skapa en SQL-databas

Skapa eller koppla till en SQL-databas på en av följande plattformar:

Skapa en SQL-databas på några minuter med hjälp av Azure-portalen

Exemplen i denna artikel använder AdventureWorksLT :s exempeldatabas, specifikt tabellen SalesLT.Product . Om du inte har AdventureWorksLT installerat, se AdventureWorks exempeldatabaser.

Projektinställningar

Installera beroenden

Installera de nödvändiga paketen med pip:

pip install flask mssql-python

Projekt-struktur

Organisera ditt projekt med separata moduler för konfiguration, anslutningshantering, rutter och tester:

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

Databasanslutningshantering

Flask har inget inbyggt databaslager, så du hanterar anslutningar direkt. Mönstret i denna sektion lagrar en anslutning per förfrågan på Flasks g objekt och stänger det automatiskt när förfrågan avslutas.

Skapa config.py

Centralisera databasinställningar i en konfigurationsklass. Miljövariabler låter dig åsidosätta standardinställningar utan att ändra koden.

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

Skapa database.py

Modulen database.py hanterar anslutningens livscykel. Flasks g-objekt är en namnrymd för varje begäran, så om anslutningen lagras där säkerställs att varje begäran får sin egen anslutning som rensas upp när begäran har slutförts.

Funktionen get_connection_string() bygger reťazec pripojenia från appens konfiguration. Funktionen get_db() skapar en anslutning vid första samtalet och återanvänder den för resten av förfrågan. Funktionen close_db() körs automatiskt i slutet av varje begäran och återställer transaktionen om ett undantag inträffar, och genomför den annars. Funktionen init_app() registrerar detta demonteringsbeteende i Flask-appen.

# 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 använder DefaultAzureCredential, som provar flera leverantörer av autentiseringsuppgifter i följd. Den första anslutningen kan vara långsam eftersom SDK:n går igenom kedjan tills den hittar en fungerande leverantör. I produktion, om du vet vilken typ av behörighet din miljö använder, ange det direkt (till exempel ActiveDirectoryMSI för managed identity) för att undvika kedjevandring. Mer information finns i Microsoft Entra-autentisering.

Flaskapplikation

Följande exempel visar en komplett Flask-applikation med rutter för listning, hämtning, skapande, uppdatering och borttagning av produkter.

Skapa app.py

Applikationsmodulen skapar Flask-appen, laddar konfigurationen och registrerar databasnedtagningen. Varje ruttfunktion anropar get_db() för att hämta en markör, utför frågor med parameteriserad SQL (med hjälp av %(name)s platshållare och en ordbok med värden) och returnerar JSON-svar.

# 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

Starta programmet

Starta utvecklingsservern:

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

Servern lyssnar på http://localhost:5000. Öppna en andra terminal och anropa slutpunkterna genom att bekräfta curl att appen kommunicerar med din databas:

# 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

I PowerShell curl är ett alias för Invoke-WebRequest. De enkla GET-kommandona här fungerar bra, men svaret kommer tillbaka som ett objekt istället för utskriven JSON. Kommandon som använder curl flaggor som -X, -H, eller -d (som exemplet POST senare) fungerar inte som skrivet. På Windows, använd curl.exe för att köra kommandona exakt som visat, eller använd PowerShell Invoke-RestMethod (till exempel Invoke-RestMethod http://localhost:5000/health), som också tolkar JSON-svaret åt dig.

Varje slutpunkt returnerar JSON. Du kan också öppna http://localhost:5000/products i en webbläsare för att se den paginerade listan.

Anslutningspoolning

Utan anslutningspooling öppnar och stänger varje förfrågan en TCP-anslutning till Microsoft SQL, vilket ökar latensen. Anslutningspoolning innebär att en uppsättning lediga anslutningar hålls redo för återanvändning. För att aktivera anslutningspoolning anropar du mssql_python.pooling() en gång på modulnivå. När poolning är aktiverat returnerar conn.close() i close_db teardown anslutningen till poolen i stället för att stänga den.

Aktivera anslutningspooler

Aktivera pooling genom att anropa mssql_python.pooling() på modulnivå innan några anslutningar öppnas:

# 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

Felhantering

Flask låter dig registrera hanterare för specifika undantagstyper. Genom att fånga mssql_python.DatabaseError och mssql_python.IntegrityError kan du returnera strukturerade JSON-felsvar i stället för HTML-felsidor som standard.

Registrera felhanterare

Lägg till dessa hanterare till den befintliga app.py, efter raden app = Flask(__name__) . Eftersom hanterarna refererar till objektet app måste de komma efter att appen har skapats. app.py behöver import mssql_python längst upp. Handlarna returnerar strukturerade JSON-svar istället för standard HTML-felsidor:

# 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

När din applikation växer blir det svårt att samla alla rutter i en enda fil. Flask Blueprints låter dig gruppera relaterade rutter i separata moduler som registreras i appen.

Organisera rutter med ritningar

Skapa en blueprint-modul för produktrutter som importerar get_db och definierar slutpunkter under ett delat 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})

Registrera ritningen

Spara blueprinten som blueprints/products.py, och lägg till en tom blueprints/__init__.py fil så att Python behandlar mappen som ett paket. Sedan, i app.py, importera blueprinten tillsammans med dina andra importer och registrera den efter raden app = Flask(__name__):

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Eftersom blueprinten anger url_prefix="/api/products", nås dess rutter under det prefixet. Till exempel är listrutten tillgänglig vid http://localhost:5000/api/products/, separat från de /products rutter som definieras direkt i app.py.

Testing

Flask tillhandahåller en testklient som skickar förfrågningar till din applikation utan att starta en riktig HTTP-server. Använd pytest fixtures för att skapa klienten och återanvänd den i tester.

Testuppställning med pytest

Skapa en pytest-fixtur som tillhandahåller en testklient och skriver tester för att verifiera ruttbeteende:

# 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

Dessa tester körs mot din aktiva databas i stället för mockobjekt, så test_create_product lägger in en verklig rad i SalesLT.Product. I AdventureWorksLT har både Name och ProductNumber unika begränsningar, så testet genererar ett unikt värde för varje körning. Om du hårdkodar dessa värden istället misslyckas testet med en konflikt vid andra körningen om du inte tar bort raden först.

Kör testerna

Spara testerna som test_app.py i din projektmapp. Med din virtuella miljö aktiverad, installera pytest och kör den från den mappen. Att installera och köra pytest i samma virtuella miljö som flask och mssql-python säkerställer att testerna importerar de paket som din app använder. pytest Upptäcker och rapporterar test_app.py automatiskt resultaten:

pip install pytest
pytest

pytest Upptäcker test_app.py automatiskt och rapporterar resultaten:

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

test_app.py ....                                       [100%]

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

Feedback

Var den här sidan till hjälp?

Last updated on 2026-07-22