FastAPI uygulamalarını mssql-python ile test edin ve dağıtın

mssql-python ile bir FastAPI uygulaması oluşturduktan sonra, onu dağıtım, bağlantı yeniden kullanımı, hata işleme, kimlik doğrulama ve otomatik test için yapılandırın.

Prerequisites

• FastAPI ile mssql-python'u tam kullanın veya AdventureWorksLT örnek veritabanını kullanan eşdeğer bir FastAPI uygulaması edin. Bu makaledeki kimlik doğrulama bağımlılığı, SalesLT.Customer’yi sorgular.

• Üretim ve test bağımlılıklarını kurun:

  pip install pydantic-settings pyjwt pytest httpx
  

Dağıtım ayarlarını yapılandırma

Ortam değişkenlerinden dağıtıma özgü değerleri yüklemek için Pydantic Settings kullanın. Bu yaklaşım, sırları kaynak koddan uzak tutar ve her ortama kendi veritabanı, havuzu ve kimlik doğrulama yapılandırması sağlar.

Oluştur config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_server: str
    database_name: str
    pool_size: int = 20
    pool_idle_timeout: int = 300
    jwt_secret: str


settings = Settings()


def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

DATABASE_SERVER, DATABASE_NAME ve JWT_SECRET öğelerini dağıtım ortamında ayarlayın. Pydantic Settings, büyük harfli ortam değişken isimlerini otomatik olarak okur.

Note

ActiveDirectoryDefault Birden fazla kimlik doğrulama sağlayıcısını ardışık olarak deniyor. Üretimde, kimlik bilgisi zincirini dolaşmaktan kaçınmak için, yönetilen kimlikte ActiveDirectoryMSI gibi dağıtılan kimlik için kimlik doğrulama modunu belirtin. Mevcut modlar için mssql-python ile Microsoft Entra doğrulama bölümüne bakınız.

Bağlantı havuzunu yapılandırma

MSSQL-Python, varsayılan olarak bağlantı havuzlamasını etkinleştirir. Havuzu bir kez yapılandırın, uygulama ilk bağlantısını oluşturmadan önce. Uygulamanın beklenen eşzamanlı veritabanı çalışması ve veritabanı hizmet katmanı için havuzu boyutlandırın.

database.py öğesini dağıtım ayarlarını kullanacak şekilde güncelleyin:

from collections.abc import Generator

import mssql_python

from config import get_connection_string, settings


mssql_python.pooling(
    max_size=settings.pool_size,
    idle_timeout=settings.pool_idle_timeout,
)


def get_db_dependency() -> Generator:
    with mssql_python.connect(get_connection_string()) as conn:
        with conn.cursor() as cursor:
            yield cursor

Bağlantı bağlam yöneticisi, istek başarıyla işlendiğinde işlemi kalıcı hâle getirir, istek işleme sırasında bir istisna oluştuğunda geri alır ve bağlantıyı kapatır. Bağlantıyı kapatınca havuza geri döner. Havuz anahtarları, boyutlandırma, kimlik yalıtımı ve havuzun tükenmesine ilişkin yönergeler için mssql-python ile bağlantı havuzu oluşturma bölümüne bakın.

Veritabanı hatalarını ele alın

İstisna işleyicilerini kaydedin ki veritabanı hataları bağlantı detaylarını, sorguları veya sunucu hata metnini açığa çıkarmadan tutarlı yanıtlar döndürsün.

main.pyiçinde app = FastAPI(...) sonrasına işleyicileri ekleyin:

import mssql_python
from fastapi import Request
from fastapi.responses import JSONResponse


@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(
    request: Request,
    exc: mssql_python.IntegrityError,
):
    return JSONResponse(
        status_code=409,
        content={
            "detail": "The request conflicts with existing data.",
            "type": "integrity_error",
        },
    )


@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(
    request: Request,
    exc: mssql_python.DatabaseError,
):
    return JSONResponse(
        status_code=500,
        content={
            "detail": "A database operation failed.",
            "type": "database_error",
        },
    )

İstisnayı, yanıtı geri almadan önce uygulamanızın korumalı telemetri boru hattı üzerinden kaydedin. İstisna hiyerarşisi ve SQLSTATE işlemesi için mssql-python için Hata işleme ve SQLSTATE kodlarına bakınız.

Kimlik doğrulama bağımlılıkları ekleyin

FastAPI bağımlılıklarını zincirleyerek bir JSON Web Token'ı (JWT) doğrulamak, eşleşen AdventureWorksLT müşterisini yüklemek ve o müşteriyi korumalı rotalara erişilebilir kılmak. Veritabanı bağlantısı edinmeden önce tokenı doğrulayın ki geçersiz bir token havuzlu bağlantı kullanmasın.

Oluştur auth.py:

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from config import settings
from database import get_db_dependency


security = HTTPBearer()


def get_customer_id(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> int:
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.jwt_secret,
            algorithms=["HS256"],
        )
        customer_id = int(payload["sub"])
    except (KeyError, TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    return customer_id


def get_current_customer(
    customer_id: int = Depends(get_customer_id),
    cursor = Depends(get_db_dependency),
):
    cursor.execute(
        """
        SELECT CustomerID, FirstName, LastName
        FROM SalesLT.Customer
        WHERE CustomerID = %(id)s
        """,
        {"id": customer_id},
    )
    customer = cursor.fetchone()
    if customer is None:
        raise HTTPException(status_code=401, detail="Customer not found")

    return {
        "id": customer.CustomerID,
        "first_name": customer.FirstName,
        "last_name": customer.LastName,
    }

Bağımlılığı içe aktarın ve main.py öğesine korumalı bir rota ekleyin:

from auth import get_current_customer


@app.get("/me")
def get_me(current_customer: dict = Depends(get_current_customer)):
    return current_customer

İmza anahtarlarını vermek ve döndürmek için bir kimlik sağlayıcısı kullanın. HS256 için JWT_SECRET değerini en az 32 rastgele bayta ayarlayın. Bir üretim imza sırrını depoda veya bir görselde saklamayın.

Uygulamayı test et

FastAPI'ler, TestClient HTTP sunucusu başlatmadan uygulamaya istekler gönderir. Aşağıdaki entegrasyon testleri yapılandırılmış veritabanını kullanır.

Oluştur test_api.py:

import uuid

from fastapi.testclient import TestClient

from main import app


client = TestClient(app)


def test_list_products():
    response = client.get("/products")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data


def test_create_product():
    suffix = uuid.uuid4().hex[:8]
    response = client.post(
        "/products",
        json={
            "name": f"Test Product {suffix}",
            "product_number": f"TEST-{suffix}",
            "price": 19.99,
            "color": "Red",
            "size": "M",
            "category_id": 1,
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["product_number"] == f"TEST-{suffix}"
    assert data["price"] == 19.99


def test_get_product_not_found():
    response = client.get("/products/99999")
    assert response.status_code == 404


def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Testleri proje kökünden çalıştırın:

pytest

Bu testler yapılandırılmış veritabanını kullanır ve test_create_product içine SalesLT.Productbir satır ekler. Özel bir test veritabanı kullanın ve test çalışmaları arasında verilerini sıfırlayın.

Dağıtım denetim listesi

• DATABASE_SERVER, DATABASE_NAME ve JWT_SECRET değerlerini dağıtım platformunun gizli bilgi ve yapılandırma depoları üzerinden ayarlayın.
• Minimum gerekli veritabanı izinlerine sahip özel bir Microsoft Entra kimliği kullanın.
• Havuz boyutunu veritabanının bağlantı sınırının altına ayarlayın ve yönetici erişim ile diğer iş yükleri için kapasite bırakın.
• Veritabanı entegrasyon testlerini yalıtılmış bir test veritabanına karşı çalıştırın.
• Veritabanı istisnaları, talep gecikmesi ve havuz tükenmesi için korunan telemetri yapılandırın.
• Dağıtım ortamlarında Uvicorn'u --reload olmadan çalıştırın.

İlgili içerik

• mssql-python'u FastAPI ile kullanın
• mssql-python ile bağlantı havuzu oluşturma
• Hata işleme ve mssql-python için SQLSTATE kodları
• mssql-python ile Microsoft Entra kimlik doğrulaması
• mssql-python uygulamaları için en iyi güvenlik uygulamaları