Použijte mssql-python s Flaskom

Flask je lehký Python webový framework, který vám dává plnou kontrolu nad strukturou aplikace. Ve spojení s mssql-python můžete vytvářet webové aplikace a REST API podporovaná Microsoft SQL a Azure SQL Database s minimálními zátěžemi.

Předpoklady

  • Python 3.10 nebo novější.
  • Balíčky mssql-python a flask. Nainstalujte obě pomocí pip install flask mssql-python.
  • Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace mssql-python.
    apk add libtool krb5-libs krb5-dev libstdc++

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Příklady v tomto článku využívají databázi AdventureWorksLT , konkrétně tabulku SalesLT.Product . Pokud nemáte AdventureWorksLT nainstalovaný, podívejte se na ukázkové databáze AdventureWorks.

Nastavení projektu

Nainstalujte závislosti

Nainstalujte požadované balíčky pomocí pip:

pip install flask mssql-python

Struktura projektu

Organizujte svůj projekt do samostatných modulů pro konfiguraci, správu spojení, trasy a testy:

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

Správa připojení k databázi

Flask neobsahuje vestavěnou databázovou vrstvu, takže spravujete spojení přímo. Vzor v této části ukládá jedno připojení pro každý požadavek do objektu g frameworku Flask a po skončení požadavku je automaticky uzavře.

Vytvořte config.py

Centralizujte nastavení databáze v konfigurační třídě. Proměnné prostředí vám umožní přepsat výchozí nastavení bez změny kódu.

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

Vytvořte database.py

Modul database.py řídí životní cyklus spojení. Objekt g ve Flask je jmenný prostor pro jednotlivé požadavky, takže ukládání spojení tam zajišťuje, že každý požadavek získá své vlastní spojení, které se po dokončení požadavku vyčistí.

Funkce get_connection_string() vytváří připojovací řetězec z konfigurace aplikace. Funkce get_db() vytvoří spojení při prvním volání a znovu jej použije pro zbytek požadavku. Funkce close_db() se automaticky spustí na konci každého požadavku, a pokud došlo k výjimce, vrátí transakci zpět; v opačném případě ji potvrdí. Funkce init_app() tuto demontáž zaznamenává v aplikaci 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 používá DefaultAzureCredential, která zkouší více poskytovatelů přihlašovacích údajů postupně. První spojení může být pomalé, protože SDK prochází řetězec, dokud nenajde funkčního poskytovatele. V produkci, pokud víte, jaký typ přihlašovacích údajů vaše prostředí používá, zadejte ho přímo (například ActiveDirectoryMSI pro spravovanou identitu), abyste se vyhnuli tzv. chain walk. Další informace naleznete v tématu ověřování Microsoft Entra.

Aplikace Flask

Následující příklad ukazuje kompletní aplikaci Flask s trasami pro zavádění, získávání, vytváření, aktualizaci a mazání produktů.

Vytvořte app.py

Aplikační modul vytvoří aplikaci Flask, načte konfiguraci a zaregistruje rozebírání databáze. Každá funkce pro jednotlivé trasy volá get_db(), aby získala kurzor, provádí dotazy pomocí parametrizovaného SQL (s využitím zástupných symbolů %(name)s a slovníku hodnot) a vrací odpovědi ve formátu 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

Spuštění aplikace

Spusťte vývojový server:

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

Server naslouchá na http://localhost:5000. Otevřete druhý terminál a volejte koncové body pomocí curl, abyste ověřili, že aplikace komunikuje s vaší databází:

# 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

V PowerShellu curl je alias pro Invoke-WebRequest. Jednoduché příkazy GET zde fungují bez problémů, ale odpověď se vrací jako objekt místo tištěného JSON. Příkazy používající curl příznaky jako -X, -H, nebo -d (jako v pozdějším příkladu POST ) nefungují tak, jak byly napsány. V systému Windows použijte curl.exe ke spuštění příkazů přesně tak, jak jsou uvedeny, nebo použijte rutinu Invoke-RestMethod prostředí PowerShell (například Invoke-RestMethod http://localhost:5000/health), která za vás také zpracuje odpověď JSON.

Každý koncový bod vrací JSON. Můžete také otevřít http://localhost:5000/products v prohlížeči a zobrazit stránkovaný seznam.

Sdílení připojení

Poolování připojení je ve výchozím nastavení zapnuté a udržuje nevyužitá spojení připravená k opětovnému použití. Před otevřením spojení zavolejte mssql_python.pooling() jednou na úrovni modulu pro změnu velikosti poolu nebo časového limitu nečinnosti. Při zapnutém poolingu conn.close() v teardownu close_db vrací připojení do poolu, místo aby fyzické připojení uzavřel.

Povolení sdružování připojení

Nakonfigurujte pooling voláním mssql_python.pooling() na úrovni modulu před otevřením jakýchkoli spojení:

# 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(), timeout=30)
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

S výchozím poskytovatelem msodbcsql18 nastaví timeout=30 časový limit přihlášení a během ověřování připojení ve fondu poskytne dočasný nativní časový limit operace připojení, v sekundách. Pokud ověření selže, ovladač spojení zahodí a pokusí se o náhradu. Více validačních operací, čištění a výměna může trvat déle než nastavená hodnota. Toto nastavení není celkový termín pro checkout připojení ani časový limit požadavku a nenastavuje časový limit dotazu.

U poskytovatele s opt-in mssql-odbc není zaručeno, že resetování pooled connection bude omezeno časovým limitem přihlášení. Pro více informací viz Omezení poskytovatelů.

Zpracování chyb

Flask umožňuje registrovat obslužné pracovníky pro konkrétní typy výjimek. Zachytí mssql_python.DatabaseError a mssql_python.IntegrityError umožní vám vracet strukturované JSON chybové odpovědi místo výchozích HTML chybových stránek.

Zaregistrujte obslužné rutiny chyb

Přidejte tyto handlery k existujícímu app.py, za řádkem app = Flask(__name__) . Protože obslužné rutiny odkazují na objekt app, musí následovat až po vytvoření aplikace. app.py potřebuje import mssql_python nahoře. Handlery vracejí strukturované JSON odpovědi místo výchozích HTML chybových stránek:

# 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

Jak vaše aplikace roste, je obtížné udržovat všechny trasy v jednom souboru. Flask Blueprints umožňují seskupit související trasy do samostatných modulů, které jsou registrované v aplikaci.

Organizujte trasy pomocí plánů

Vytvořte modul blueprint pro produktové trasy, který importuje get_db a definuje koncové body pod sdíleným prefixem URL:

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

Zaregistrujte plán

Uložte blueprint jako blueprints/products.py, a přidejte prázdný blueprints/__init__.py soubor, aby Python považoval složku za balíček. Pak v , app.pyimportujte blueprint spolu s ostatními importy a registrujte ho za řádkem app = Flask(__name__) :

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Protože blueprint nastavuje url_prefix="/api/products", jsou jeho trasy obsluhovány pod tímto prefixem. Například trasa seznamu je k dispozici na adrese http://localhost:5000/api/products/, odděleně od tras /products, které jsou definovány přímo v app.py.

Testing

Flask poskytuje testovacího klienta, který posílá požadavky do vaší aplikace bez spuštění skutečného HTTP serveru. Pomocí pytest fixture vytvořte klienta a znovu ho používejte ve všech testech.

Nastavení testu pomocí pytest

Vytvořte pytest fixture, který poskytuje testovacího klienta, a napište testy pro ověření chování trasy:

# 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

Tyto testy běží na vaší živé databázi místo mocků, takže test_create_product vloží skutečný řádek do SalesLT.Product. V AdventureWorksLT mají Name i ProductNumber jedinečná omezení, takže test při každém spuštění generuje pro každý z nich jedinečnou hodnotu. Pokud tyto hodnoty místo toho zakódujete napevno, test při druhém spuštění s konfliktem selže, pokud nejdřív řádek nesmažete.

Spuštění testů

Uložte testy do test_app.py složky projektu. Po aktivaci virtuálního prostředí nainstalujte pytest a spusťte jej v této složce. Instalace a spouštění pytest ve stejném virtuálním prostředí jako flask a mssql-python zajistí, že testy importují balíčky, které vaše aplikace používá. pytest automaticky zjišťuje test_app.py a hlásí výsledky:

pip install pytest
pytest

pytest automaticky zjišťuje test_app.py a hlásí výsledky:

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

test_app.py ....                                       [100%]

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

Související obsah

  • Poslední aktualizace v