Gunakan mssql-python dengan Flask

Flask adalah kerangka kerja web Python ringan yang memberi Anda kendali penuh atas struktur aplikasi. Dikombinasikan dengan mssql-python, Anda dapat membangun aplikasi web dan REST API yang didukung oleh Microsoft SQL dan Azure SQL Database dengan overhead minimal.

Prasyarat

  • Python 3.10 atau yang lebih baru.
  • Paket mssql-python dan flask. Instal keduanya dengan pip install flask mssql-python.
  • Instal prasyarat khusus sistem operasi satu kali. Pengguna Windows dapat melewati langkah ini. Untuk detail platform lengkap, lihat Menginstal mssql-python.
    apk add libtool krb5-libs krb5-dev

Membuat database SQL

Buat atau sambungkan ke database SQL di salah satu platform berikut:

Contoh dalam artikel ini menggunakan database sampel AdventureWorksLT , khususnya SalesLT.Product tabel. Jika Anda belum menginstal AdventureWorksLT, lihat Database sampel AdventureWorks.

Penyusunan proyek

Pasang dependensi

Instal paket yang diperlukan dengan pip:

pip install flask mssql-python

Struktur proyek

Atur proyek Anda dengan modul terpisah untuk konfigurasi, manajemen koneksi, rute, dan pengujian:

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

Manajemen koneksi database

Flask tidak menyertakan lapisan database bawaan, sehingga Anda mengelola koneksi secara langsung. Pola di bagian ini menyimpan satu koneksi per permintaan pada objek Flask g dan menutupnya secara otomatis saat permintaan berakhir.

Buat config.py

Pusatkan pengaturan database dalam kelas konfigurasi. Variabel lingkungan memungkinkan Anda mengganti default tanpa mengubah kode.

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

Buat database.py

Modul ini database.py mengelola siklus hidup koneksi. Objek Flask g adalah namespace per permintaan, jadi menyimpan koneksi di sana memastikan setiap permintaan mendapatkan koneksinya sendiri yang dibersihkan saat permintaan selesai.

Fungsi ini get_connection_string() membuat string koneksi dari konfigurasi aplikasi. Fungsi ini get_db() membuat koneksi pada panggilan pertama dan menggunakannya kembali untuk sisa permintaan. Fungsi close_db() dijalankan secara otomatis pada akhir setiap permintaan, membatalkan transaksi jika terjadi pengecualian dan mengomitnya jika tidak. Fungsi init_app() ini mendaftarkan perilaku pembersihan ini pada aplikasi 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 menggunakan DefaultAzureCredential, yang mencoba beberapa penyedia kredensial secara berurutan. Koneksi pertama bisa lambat karena SDK menelusuri rantai hingga menemukan penyedia yang berfungsi. Dalam lingkungan produksi, jika Anda mengetahui jenis kredensial yang digunakan oleh lingkungan Anda, tentukan secara langsung (misalnya, ActiveDirectoryMSI untuk identitas terkelola) agar terhindar dari penelusuran berantai. Untuk informasi selengkapnya, lihat Autentikasi Microsoft Entra.

Aplikasi Flask

Contoh berikut menunjukkan aplikasi Flask lengkap dengan rute untuk mencantumkan, mengambil, membuat, memperbarui, dan menghapus produk.

Buat app.py

Modul aplikasi membuat aplikasi Flask, memuat konfigurasi, dan mendaftarkan pembongkaran database. Setiap fungsi rute memanggil get_db() untuk mendapatkan kursor, mengeksekusi kueri dengan SQL berparameter (menggunakan %(name)s placeholder dan kamus nilai), dan mengembalikan respons 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

Jalankan aplikasi

Mulai server pengembangan:

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

Server mendengarkan pada http://localhost:5000. Buka terminal kedua dan panggil titik akhir dengan menggunakan curl untuk mengonfirmasi aplikasi berbicara dengan database Anda:

# 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

Di PowerShell, curl adalah alias untuk Invoke-WebRequest. Perintah GET sederhana di sini berjalan dengan baik, tetapi respons yang dikembalikan berupa objek, bukan JSON yang dicetak. Perintah yang menggunakan curl bendera seperti -X, -H, atau -d (seperti POST contoh nanti) tidak berfungsi seperti yang tertulis. Di Windows, gunakan curl.exe untuk menjalankan perintah persis seperti yang ditunjukkan, atau gunakan PowerShell Invoke-RestMethod (misalnya, Invoke-RestMethod http://localhost:5000/health), yang juga mengurai respons JSON untuk Anda.

Setiap titik akhir mengembalikan JSON. Anda juga dapat membuka http://localhost:5000/products di browser untuk melihat daftar paginasi.

Pemanfaatan koneksi

Tanpa pengumpulan koneksi, setiap permintaan membuka dan menutup koneksi TCP ke Microsoft SQL, yang menambahkan latensi. Pengumpulan koneksi membuat satu set koneksi menganggur siap untuk digunakan kembali. Untuk mengaktifkan pengumpulan koneksi memanggil mssql_python.pooling() sekali di tingkat modul. Dengan pooling diaktifkan, conn.close() dalam teardown close_db mengembalikan koneksi ke pool alih-alih menutupnya.

Aktifkan pengumpulan koneksi

Aktifkan pengumpulan dengan memanggil mssql_python.pooling() di tingkat modul sebelum koneksi apa pun dibuka:

# 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

Penanganan kesalahan

Flask memungkinkan Anda mendaftarkan penanganan untuk jenis pengecualian tertentu. Menangkap mssql_python.DatabaseError dan mssql_python.IntegrityError memungkinkan Anda mengembalikan respons kesalahan JSON terstruktur, bukan halaman kesalahan HTML default.

Daftarkan penangan kesalahan

Tambahkan handler ini ke app.py yang sudah ada, setelah baris app = Flask(__name__). Karena handler mereferensikan objek app, handler tersebut harus diletakkan setelah aplikasi dibuat. app.py memerlukan import mssql_python di bagian atas. Handler mengembalikan respons JSON terstruktur, bukan halaman error HTML default:

# 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

Seiring berkembangnya aplikasi Anda, menempatkan semua rute dalam satu file menjadi sulit untuk dipelihara. Blueprint Flask memungkinkan Anda mengelompokkan rute terkait ke dalam modul terpisah yang didaftarkan pada aplikasi.

Mengatur rute dengan cetak biru

Buat modul cetak biru untuk rute produk yang mengimpor get_db dan menentukan titik akhir di bawah awalan URL bersama:

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

Daftarkan cetak biru

Simpan cetak biru sebagai blueprints/products.py, dan tambahkan file kosong blueprints/__init__.py sehingga Python memperlakukan folder sebagai paket. Kemudian, di app.py, impor blueprint bersama impor lainnya dan daftarkan setelah baris app = Flask(__name__):

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Karena blueprint menentukan url_prefix="/api/products", rute-rutenya tersedia di bawah prefiks tersebut. Misalnya, rute list tersedia di http://localhost:5000/api/products/, terpisah dari rute /products yang ditentukan langsung di app.py.

Testing

Flask menyediakan klien pengujian yang mengirim permintaan ke aplikasi Anda tanpa memulai server HTTP nyata. Gunakan fixture pytest untuk membuat klien dan menggunakannya kembali di berbagai pengujian.

Pengaturan pengujian dengan pytest

Buat perlengkapan pytest yang menyediakan klien pengujian dan tulis pengujian untuk memverifikasi perilaku rute:

# 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

Pengujian ini dijalankan pada database aktif Anda, bukan mock, jadi test_create_product menyisipkan baris sungguhan ke SalesLT.Product. Di AdventureWorksLT, keduanya Name dan ProductNumber memiliki batasan unik, sehingga pengujian menghasilkan nilai unik untuk masing-masing pada setiap eksekusi. Jika Anda menetapkan nilai-nilai tersebut secara hardcode sebagai gantinya, pengujian akan gagal karena konflik pada eksekusi kedua kecuali Anda menghapus baris tersebut terlebih dahulu.

Jalankan pengujian

Simpan pengujian seperti di test_app.py folder proyek Anda. Dengan lingkungan virtual Anda diaktifkan, instal pytest dan jalankan dari folder tersebut. Menginstal dan menjalankan pytest di dalam lingkungan virtual yang sama dengan flask dan mssql-python memastikan pengujian mengimpor paket yang digunakan aplikasi Anda. pytest secara otomatis menemukan test_app.py dan melaporkan hasilnya:

pip install pytest
pytest

pytest Menemukan test_app.py secara otomatis dan melaporkan hasilnya:

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

test_app.py ....                                       [100%]

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

Konten terkait

  • Last updated on