Használd az mssql-pythont a Flask-szal

A Flask egy könnyű Python webkeretrendszer, amely teljes irányítást ad az alkalmazásstruktúra felett. Az mssql-pythonnal kombinálva minimális költségköltséggel építhetsz webalkalmazásokat és REST API-kat Microsoft SQL és Azure SQL Database támogatásával.

Prerequisites

  • Python 3.10 vagy újabb verzió.
  • A mssql-python és flask csomagok. Telepítsd mindkettőt a(z) pip install flask mssql-python használatával.
  • Egyszeri operációs rendszerspecifikus előfeltételek telepítése. A Windows felhasználók ezt a lépést kihagyhatják. A platform teljes részleteiért lásd: Install mssql-python.
    apk add libtool krb5-libs krb5-dev

SQL-adatbázis létrehozása

Létrehozni vagy csatlakozni SQL adatbázishoz az alábbi platformok egyikén:

A cikkben szereplő példák az AdventureWorksLT mintaadatbázist, konkrétan a SalesLT.Product táblát használják. Ha nincs telepítve az AdventureWorksLT, nézd meg az AdventureWorks mintaadatbázisokat.

Projekt beállítás

Függőségek telepítése

Telepítsd a szükséges csomagokat pip-jel:

pip install flask mssql-python

A projekt szerkezete

Szervezze meg a projektjét külön modulokkal a konfigurációhoz, kapcsolatkezeléshez, útvonalakhoz és tesztekhez:

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

Adatbázis-kapcsolatkezelés

A Flask nem tartalmaz beépített adatbázis-réteget, így közvetlenül kezeled a kapcsolatokat. Ebben a szakaszban szereplő minta minden kérés után egy kapcsolatot tárol a Flask g objektumán, és automatikusan lezárja a kérés végén.

Készíts config.py

Központosítsd az adatbázis beállításait egy konfigurációs osztályban. A környezeti változók lehetővé teszik, hogy az alapértelmezett beállításokat felülírd anélkül, hogy kódot változtatsz.

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

Készíts database.py

A database.py modul kezeli a kapcsolati életciklust. A Flask g objektuma kérelmenként külön névtér, így ha ott tároljuk a kapcsolatot, az biztosítja, hogy minden kérés saját kapcsolatot kapjon, amely a kérés végén felszabadul.

A get_connection_string() függvény az alkalmazás konfigurációjából építi fel a kapcsolati karakterlánc-et. A get_db() függvény az első hívásnál létrehoz egy kapcsolatot, majd újrahasználja azt a kérés többi részéhez. A close_db() függvény automatikusan lefut minden kérés végén: ha kivétel történt, visszagörgeti a tranzakciót, ellenkező esetben pedig véglegesíti azt. A init_app() funkció ezt a lebontási viselkedést a Flask alkalmazással regisztrálja.

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

Megjegyzés:

ActiveDirectoryDefault használ DefaultAzureCredential, amely több hitelesítésszolgáltatót próbál egymás után. Az első kapcsolat lassú lehet, mert az SDK végigjárja a láncot, amíg meg nem talál egy működő szolgáltatót. A termelésben, ha tudod, melyik hitelesítéstípust használja a környezeted, közvetlenül megadd (például ActiveDirectoryMSI menedzselt identitásnál), hogy elkerüld a láncos sétát. További információ: Microsoft Entra-hitelesítés.

Flask-alkalmazás

Az alábbi példa egy teljes Flask alkalmazást mutat be, amelyekben útvonalak találhatók a termékek listázására, lekérésére, létrehozására, frissítésére és törlésére.

Készíts app.py

Az alkalmazásmodul létrehozza a Flask alkalmazást, betölti a konfigurációt, és regisztrálja az adatbázis lebontását. Minden útvonalfüggvény meghívja a get_db() függvényt egy kurzor lekéréséhez, paraméterezett SQL-lel hajt végre lekérdezéseket (%(name)s helyőrzők és egy értékszótár használatával), és JSON-válaszokat ad vissza.

# 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

Az alkalmazás futtatása

Indítsa el a fejlesztői kiszolgálót:

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

A szerver a(z) http://localhost:5000 címen figyel. Nyiss egy második terminált, és hívd meg a végpontokat azzal, hogy curl megerősítsd, hogy az alkalmazás kommunikál az adatbázisoddal:

# 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

Megjegyzés:

A PowerShellben a curl egy alias a Invoke-WebRequest számára. Az egyszerű GET parancsok itt jól működnek, de a válasz objektumként jelenik meg, nem pedig nyomtatott JSON-ként. Olyan parancsok, amelyek curl zászlókat használnak, mint -X, -H, vagy -d (például a POST későbbi példa), nem úgy működnek, ahogy leírták. Windows rendszeren a parancsok pontosan a megadott módon történő futtatásához használd a curl.exe eszközt, vagy a PowerShell Invoke-RestMethod elemét (például: Invoke-RestMethod http://localhost:5000/health), amely a JSON-választ is feldolgozza helyetted.

Minden végpont JSON-t ad vissza. A lapokra osztott lista megtekintéséhez böngészőben is megnyithatod a(z) http://localhost:5000/products elemet.

Kapcsolatmegosztás

Kapcsolati pooling nélkül minden kérés megnyitja és zár egy TCP kapcsolatot a Microsoft SQL-hez, ami késleltetést növel. A kapcsolat-összevonás a tétlen kapcsolatok egy készletét készen tartja az újrafelhasználásra. A kapcsolat-újrafelhasználás engedélyezéséhez modulszinten egyszer hívja meg a(z) mssql_python.pooling() elemet. Ha a pooling engedélyezve, conn.close() a close_db lebontásban visszaadja a kapcsolatot a poolhoz, nem zárná be.

Kapcsolatkészletezés engedélyezése

A poolingolás engedélyezése modul szinten hívással mssql_python.pooling() a csatlakozások megnyitása előtt:

# 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

Hibakezelés

A Flask lehetővé teszi, hogy regisztráld a kezelőket speciális kivételtípusokra. A mssql_python.DatabaseError és mssql_python.IntegrityError elkapása lehetővé teszi, hogy strukturált JSON hibaválaszokat adjon vissza az alapértelmezett HTML hibaoldalak helyett.

Hibakezelők regisztrálása

Ezeket a kezelőket adjuk hozzá a meglévő app.py, a app = Flask(__name__) sor után. Mivel a kezelők hivatkoznak az app objektumra, az alkalmazás létrehozása után kell megjelenniük. app.py tetejére import mssql_python szükséges. A kezelők strukturált JSON válaszokat adnak vissza alapértelmezett HTML hibaoldalak helyett:

# 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

Ahogy az alkalmazás növekszik, az összes útvonal egyetlen fájlba való elhelyezése nehezebb lesz fenntartani. A Flask Blueprints lehetővé teszi, hogy a kapcsolódó útvonalakat külön modulokba csoportosítsd, amelyek regisztrálva vannak az alkalmazásban.

Szervezd meg az útvonalakat tervrajzokkal

Hozz létre egy tervrajz modult a termékútvonalakhoz, amely importálja get_db és definiálja a végpontokat egy közös URL előtag alatt:

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

Regisztrálja a tervrajzot

Mentsd el a blueprintet , blueprints/products.pyés adj hozzá egy üres blueprints/__init__.py fájlt, hogy a Python csomagként kezelje a mappát. Ezután a(z) app.py fájlban importáld a blueprintet a többi importtal együtt, és regisztráld a(z) app = Flask(__name__) sor után:

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Mivel a blueprint beállítja a(z) url_prefix="/api/products" értéket, az útvonalai ezen előtag alatt érhetők el. Például a listaútvonal a(z) http://localhost:5000/api/products/ címen érhető el, elkülönülve a közvetlenül a(z) app.py fájlban definiált /products útvonalaktól.

Testing

A Flask egy tesztklienst biztosít, amely kéréseket küld az alkalmazásodnak anélkül, hogy valódi HTTP szervert indítana. Használj pytest fixtúrákat a kliens létrehozásához, és újrahasznosítsd a tesztek során.

Test beállítása pytest-szel

Hozz létre egy pytest fixtúrát, amely tesztklienst biztosít, és írj teszteket az útvonal viselkedésének ellenőrzésére:

# 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

Ezek a tesztek az élő adatbázisodon futnak, nem pedig mock objektumokon, így a test_create_product egy valódi sort szúr be a SalesLT.Product-be. Az AdventureWorksLT-ben mindkettőnek NameProductNumber egyedi korlátai vannak, így a teszt minden futáskor egyedi értéket generál mindegyikhez. Ha ehelyett hardcode-olod ezeket az értékeket, a teszt a második futtatáskor ütközés miatt sikertelen lesz, hacsak előtte nem törlöd a sort.

A tesztek futtatása

Mentsd a teszteket test_app.py néven a projektmappádba. Miután aktiváltad a virtuális környezetet, telepítsd a(z) pytest elemet, majd futtasd ebből a mappából. A(z) pytest ugyanabban a virtuális környezetben való telepítése és futtatása, mint a(z) flask és mssql-python, biztosítja, hogy a tesztek az alkalmazás által használt csomagokat importálják. pytest automatikusan test_app.py felfedezi és jelenti az eredményeket:

pip install pytest
pytest

pytest automatikusan test_app.py felfedezi és jelenti az eredményeket:

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

test_app.py ....                                       [100%]

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

Kapcsolódó tartalom

  • Last updated on