FastAPI alkalmazások tesztelése és telepítése mssql-python használatával

Miután mssql-pythonnal FastAPI alkalmazást építettél, konfiguráld telepítésre, kapcsolat újrahasználatra, hibakezelésre, hitelesítésre és automatizált tesztelésre.

Prerequisites

  • Végezze el a Use mssql-python with FastAPI című részt, vagy rendelkezzen egy vele egyenértékű FastAPI-alkalmazással, amely az AdventureWorksLT mintaadatbázist használja. A cikkben található hitelesítési függőség a(z) SalesLT.Customer elemet kérdezi le.

  • Telepítsd a gyártási és teszt függőségeket:

    pip install pydantic-settings pyjwt pytest httpx

Üzembehelyezési beállítások konfigurálása

Használj Pydantic Settings-t a telepítés-specifikus értékek betöltésére környezeti változókból. Ez a megközelítés távol tartja a titkokat a forráskódtól, és minden környezetnek saját adatbázist, készletet és hitelesítési konfigurációt ad.

Hozza létre 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"
    )

Állítsa be a DATABASE_SERVER-t, a DATABASE_NAME-et és a JWT_SECRET-t a telepítési környezetben. A Pydantic Settings automatikusan olvassa a nagybetűs környezeti változó neveket.

Note

ActiveDirectoryDefault Több hitelesítési szolgáltatót próbál egymás után is. Gyártásban megadjuk a telepített identitás hitelesítési módját, például ActiveDirectoryMSI a menedzselt identitásnál, hogy elkerüld a hitelesítési láncban való átjárást. Az elérhető módokért lásd: Microsoft Entra hitelesítés mssql-python-szal.

Kapcsolat-összevonás konfigurálása

Az MSSQL-python alapértelmezés szerint engedélyezi a kapcsolat pooling-et. Egyszer konfiguráld a poolt, mielőtt az alkalmazás létrehozza az első kapcsolatát. Méretezze a poolt az alkalmazás várható egyidejű adatbázis-munkájához és az adatbázis-szolgáltatás szintjéhez.

Frissítse database.py a telepítési beállítások használatára:

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

A kapcsolat kontextuskezelője sikeres kérésfeldolgozás után véglegesíti a tranzakciót, ha a kérés feldolgozása kivételt vált ki, visszagörgeti a tranzakciót, és lezárja a kapcsolatot. A kapcsolat lezárása visszaadja a medencébe. Pool kulcsokért, méretezésért, identitás-izolációért és kimerülési útmutatóért lásd: Connection pooling with mssql-python.

Adatbázis hibák kezelése

Regisztrálja a kivételkezelőket, hogy az adatbázishibák egységes válaszokat adjanak anélkül, hogy felfednék a csatlakozási adatokat, a lekérdezéseket vagy a szerverhibaüzenetek szövegét.

Add hozzá a kezelőket a main.py elemben, a app = FastAPI(...) után:

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",
        },
    )

Naplózza a kivételt az alkalmazás védett telemetriai folyamatán keresztül, mielőtt visszaadná a választ. A kivételhierarchia és az SQLSTATE kezelés érdekében lásd: Hibakezelés és SQLSTATE kódok mssql-pythonhoz.

Hitelesítési függőségek hozzáadása

Fűzz össze FastAPI-függőségeket egy JSON Web Token (JWT) érvényesítéséhez, a hozzá tartozó AdventureWorksLT-ügyfél betöltéséhez, valamint ahhoz, hogy ez az ügyfél elérhető legyen a védett útvonalak számára. Ellenőrizd a tokent az adatbázis-kapcsolat megszerzése előtt, hogy egy érvénytelen token ne használjon pooled kapcsolatot.

Hozza létre 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,
    }

Importáljuk a függőséget, és adjunk hozzá egy védett útvonalat a következőkhöz main.py:

from auth import get_current_customer


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

Használj identitásszolgáltatót az aláíráskulcsok kiadásához és rotációjához. HS256 esetén legalább 32 véletlenszerű bájtra állítsd JWT_SECRET be. Ne tárolj egy gyártási aláírás titkát a tárban vagy képen.

Az alkalmazás tesztelése

A TestClient FastAPI kéréseket küld az alkalmazásnak anélkül, hogy HTTP szervert indítanának. Az alábbi integrációs tesztek a konfigurált adatbázist használják.

Hozza létre 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"

Futtasd le a teszteket a projekt gyökéréből:

pytest

Ezek a tesztek a konfigurált adatbázist használják, és a test_create_product egy sort szúr be a SalesLT.Product táblába. Használj egy dedikált tesztadatbázist, és állítsd vissza az adatokat a tesztfuttatások között.

Üzembe helyezési ellenőrzőlista

  • Állítsd be a(z) DATABASE_SERVER, DATABASE_NAME és JWT_SECRET értékét a telepítési platform titok- és konfigurációtárolóiban.
  • Használj egy dedikált Microsoft Entra identitást, amely a minimális szükséges adatbázis-jogosultságokkal rendelkezik.
  • Állítsd be a pool méretét az adatbázis kapcsolati határa alá, és hagyj kapacitást adminisztratív hozzáférésre és egyéb munkaterhelésre.
  • Futtass adatbázis-integrációs teszteket egy izolált tesztadatbázis ellen.
  • Konfiguráld védett telemetriát adatbázis kivételek, kérelmezési késleltetés és pool kimerülése esetén.
  • Futtasd az Uvicorn-t --reload nélkül éles környezetben.

Kapcsolódó tartalom

  • Last updated on