Flask ile mssql-python kullanın

Flask, uygulama yapısı üzerinde tam kontrol sağlayan hafif bir Python web framework'tür. mssql-python ile birleştiğinde, Microsoft SQL ve Azure SQL Veritabanı destekli web uygulamaları ve REST API'leri minimum yükle oluşturabilirsiniz.

Prerequisites

  • Python 3.10 veya üzeri.
  • mssql-python ve flask paketleri. Her ikisini de . pip install flask mssql-pythonile kur.
  • Tek seferlik işletim sistemine özgü önkoşulları yükleyin. Windows kullanıcıları bu adımı atlayabilir. Platformun tam detayları için mssql-python'u Install sayfasına bakınız.
    apk add libtool krb5-libs krb5-dev

SQL veritabanı oluşturma

Aşağıdaki platformlardan birinde bir SQL veritabanı oluşturun veya bağlanın:

Bu makaledeki örnekler AdventureWorksLT örnek veritabanını, özellikle SalesLT.Product tabloyu kullanır. Eğer AdventureWorksLT'niz yüklü değilse, AdventureWorks örnek veritabanlarına bakabilirsiniz.

Proje kurulumu

Bağımlılıkları yükleme

Gerekli paketleri pip ile kurun:

pip install flask mssql-python

Proje yapısı

Projenizi yapılandırma, bağlantı yönetimi, rotalar ve testler için ayrı modüllerle organize edin:

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

Veritabanı bağlantı yönetimi

Flask yerleşik veritabanı katmanı içermiyor, bu yüzden bağlantıları doğrudan yönetiyorsunuz. Bu bölümdeki desen, Flask g nesnesinde her istek için bir bağlantı saklar ve istek bittiğinde otomatik olarak kapatılır.

config.py oluştur

Veritabanı ayarlarını bir yapılandırma sınıfında merkezileştirin. Ortam değişkenleri, kodu değiştirmeden varsayılan ayarları geçersiz kılmanıza olanak tanır.

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

database.py oluştur

Modül, database.py bağlantı yaşam döngüsünü yönetir. Flask'ın nesnesi g istek başına bir isim alanıdır, bu yüzden bağlantıyı orada saklamak, her isteğin kendi bağlantısını almasını sağlar ve istek bittiğinde temizlenir.

Bu fonksiyon, get_connection_string() uygulama yapılandırmasından bağlantı dizesi'i oluşturur. Fonksiyon, get_db() ilk çağrıda bağlantı oluşturur ve talebin geri kalanında tekrar kullanır. close_db() fonksiyonu, her isteğin sonunda otomatik olarak çalışır; bir istisna oluşursa işlemi geri alır, aksi takdirde onaylar. Bu init_app() işlevi, bu temizleme davranışını Flask uygulamasına kaydeder.

# 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 DefaultAzureCredential, kullanır ve bu da birden fazla kimlik doğrulama sağlayıcısını sırayla dener. İlk bağlantı yavaş olabilir çünkü SDK çalışan bir sağlayıcı bulana kadar zincirde yürür. Üretimde, ortamınızın hangi kimlik bilgisi türünü kullandığını biliyorsanız, zincirde dolaşmayı önlemek için bunu doğrudan belirtin (örneğin, yönetilen kimlik için ActiveDirectoryMSI). Daha fazla bilgi için bkz . Microsoft Entra kimlik doğrulaması.

Flask uygulaması

Aşağıdaki örnek, ürün listeleme, elde etme, oluşturma, güncelleme ve silme rotalarıyla tam bir Flask uygulamasını göstermektedir.

app.py oluştur

Uygulama modülü Flask uygulamasını oluşturur, yapılandırmayı yükler ve veritabanı sökülmesini kaydeder. Her rota fonksiyonu bir imleç almak için çağrı yapar get_db() , parametreli SQL ile sorguları (yer tutucular ve değer sözlüğü kullanarak %(name)s ) çalıştırır ve JSON yanıtlarını döndürür.

# 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

Uygulamayı çalıştırma

Geliştirme sunucusunu başlatın:

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

Sunucu, http://localhost:5000 üzerinde dinliyor. İkinci bir terminal açın ve son noktaları arayın, uygulamanın veritabanınızla iletişim kurduğunu doğrulamak için kullanın curl :

# 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

PowerShell'de, curl için Invoke-WebRequestbir diğer addır. Buradaki basit GET komutları sorunsuz çalışıyor, ancak yanıt JSON olarak yazdırılmak yerine bir nesne olarak döndürülüyor. curl gibi, -X, -H veya -d bayraklarını kullanan komutlar (örneğin, daha sonra verilen POST örneği) yazıldıkları gibi çalışmaz. Windows'ta, komutları tam gösterildiği gibi çalıştırmak için curl.exe kullanın veya PowerShell'in Invoke-RestMethod komutunu kullanın (örneğin, Invoke-RestMethod http://localhost:5000/health); bu da JSON yanıtını sizin için ayrıştırır.

Her uç nokta JSON döndürür. Ayrıca sayfalı listeyi görmek için tarayıcıda açabilirsiniz http://localhost:5000/products .

Bağlantı havuzlama

Bağlantı havuzu olmadan, her istek Microsoft SQL'e bir TCP bağlantısını açıp kapatır ve bu da gecikme ekler. Bağlantı havuzu, boştaki bağlantıları yeniden kullanılmak üzere hazır tutar. Bağlantı havuzunu etkinleştirmek için mssql_python.pooling() bileşenini modül düzeyinde bir kez çağırın. Havuzlama etkinleştirildiğinde, close_db teardown işlemindeki conn.close(), bağlantıyı kapatmak yerine havuza geri döndürür.

Bağlantı havuzunu etkinleştirme

Herhangi bir bağlantı açılmadan önce modül seviyesinde çağrı mssql_python.pooling() yaparak havuzlama etkinleştirin:

# 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

Hata yönetimi

Flask, belirli istisna türleri için işleyicileri kaydetmenize olanak tanır. Yakalama mssql_python.DatabaseError ve mssql_python.IntegrityError varsayılan HTML hata sayfaları yerine yapılandırılmış JSON hata yanıtları döndürmenize olanak tanır.

Hata işleyicilerini kaydedin.

Bu işleyicileri, mevcut app.py öğesine, app = Flask(__name__) satırından sonra ekleyin. Yöneticiler nesneye app referans verdiği için, uygulama oluşturulduktan sonra gelmeliler. app.py üstte import mssql_python gerektirir. Yöneticiler, varsayılan HTML hata sayfaları yerine yapılandırılmış JSON yanıtlarını döndürür:

# 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

Uygulamanız büyüdükçe, tüm yolları tek bir dosyada toplamak zorlaşıyor. Flask Blueprints, ilgili rotaları uygulamaya kayıtlı ayrı modüller halinde gruplamanıza olanak tanıyor.

Güzergahları planlarla düzenleyin

Ürün rotaları için get_db öğesini içe aktaran ve uç noktaları ortak bir URL öneki altında tanımlayan bir blueprint modülü oluşturun:

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

Planı kaydet

Blueprint'i , blueprints/products.pyolarak kaydedin ve Python'un klasörü bir paket olarak ele alması için boş blueprints/__init__.py bir dosya ekleyin. Ardından, app.py içinde blueprint'i diğer içe aktarmalarınızla birlikte içe aktarın ve app = Flask(__name__) satırından sonra kaydedin:

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

Blueprint url_prefix="/api/products" ayarladığı için, rotaları bu önek altında sunulur. Örneğin, liste rotası, app.py içinde doğrudan tanımlanan /products rotalarından ayrı olarak http://localhost:5000/api/products/ konumunda kullanılabilir.

Testing

Flask, gerçek bir HTTP sunucusu başlatmadan uygulamanıza istekler gönderen bir test istemcisi sağlar. İstemciyi oluşturmak ve testler arasında yeniden kullanmak için pytest fikstürlerini kullanın.

pytest ile test kurulumu

Test istemcisi sağlayan bir pytest fikstürü oluşturun ve rota davranışını doğrulamak için testler yazın:

# 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

Bu testler, mocklar yerine canlı veritabanınıza karşı çalışır; bu nedenle test_create_product, SalesLT.Product içine gerçek bir satır ekler. AdventureWorksLT'de her ikisinin NameProductNumber de benzersiz kısıtlamaları vardır, bu yüzden test her koşuda her biri için benzersiz bir değer üretir. Bunun yerine bu değerleri sabit olarak kodlarsanız, önce satırı silmediğiniz sürece test ikinci çalıştırmada bir çakışma nedeniyle başarısız olur.

Testleri çalıştırma

Testleri proje klasörünüzdeki gibi test_app.py kaydedin. Sanal ortamınız etkinleştiğinde, o klasörden kurup pytest çalıştırın. flask ve mssql-python ile aynı sanal ortamda pytest kurmak ve çalıştırmak, testlerin uygulamanızın kullandığı paketleri içe aktarmasını sağlar. pytest Sonuçları otomatik olarak keşfeder test_app.py ve bildirir:

pip install pytest
pytest

pytest Otomatik olarak bulur test_app.py ve sonuçları bildirir:

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

test_app.py ....                                       [100%]

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

İlgili içerik

  • Last updated on