Uji dan terapkan aplikasi FastAPI dengan mssql-python

Setelah Anda membangun aplikasi FastAPI dengan mssql-python, konfigurasikan aplikasi tersebut untuk deployment, penggunaan kembali koneksi, penanganan error, autentikasi, dan pengujian otomatis.

Prasyarat

Selesaikan Penggunaan mssql-python dengan FastAPI, atau miliki aplikasi FastAPI yang setara yang menggunakan database sampel AdventureWorksLT. Dependensi autentikasi yang digunakan dalam artikel ini membuat kueri ke SalesLT.Customer.

Instal dependensi produksi dan pengujian:

pip install pydantic-settings pyjwt pytest httpx

Mengonfigurasi pengaturan penerapan

Gunakan Pydantic Settings untuk memuat nilai khusus deployment dari variabel lingkungan. Pendekatan ini mencegah informasi rahasia disimpan dalam kode sumber dan memberi setiap lingkungan database, pool, serta konfigurasi autentikasinya sendiri.

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

Atur DATABASE_SERVER, DATABASE_NAME, dan JWT_SECRET di lingkungan deployment. Pengaturan Pydantic secara otomatis membaca nama variabel lingkungan dengan huruf kapital.

Note

ActiveDirectoryDefault mencoba beberapa penyedia kredensial secara berurutan. Dalam produksi, tentukan mode autentikasi untuk identitas yang digunakan saat deployment, seperti ActiveDirectoryMSI untuk identitas terkelola, agar tidak menelusuri rantai kredensial. Untuk mode yang tersedia, lihat autentikasi Microsoft Entra dengan mssql-python.

Mengonfigurasi pengumpulan koneksi

mssql-python secara default mengaktifkan connection pooling. Konfigurasikan pool sekali, sebelum aplikasi membuat koneksi pertamanya. Tentukan ukuran pool berdasarkan beban kerja database secara bersamaan yang diharapkan dari aplikasi dan tier layanan database.

Perbarui database.py untuk menggunakan pengaturan penerapan:

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

Pengelola konteks koneksi melakukan commit setelah pemrosesan permintaan berhasil, mundur saat pemrosesan permintaan menimbulkan pengecualian, dan menutup koneksi. Menutup koneksi mengembalikannya ke pool. Untuk kunci pengelompokan koneksi, penentuan ukuran, isolasi identitas, dan panduan tentang kehabisan koneksi, lihat Connection pooling with mssql-python.

Menangani kesalahan basis data

Daftarkan penangan pengecualian agar kegagalan basis data mengembalikan respons yang konsisten tanpa menampilkan detail koneksi, kueri, atau teks kesalahan server.

Tambahkan handler setelah app = FastAPI(...) di main.py:

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

Catat pengecualian melalui pipeline telemetri terlindungi aplikasi Anda sebelum mengembalikan respons. Untuk hierarki pengecualian dan penanganan SQLSTATE, lihat Penanganan kesalahan dan kode SQLSTATE untuk mssql-python.

Tambahkan ketergantungan autentikasi

Rangkai dependensi FastAPI untuk memvalidasi JSON Web Token (JWT), memuat data pelanggan AdventureWorksLT yang sesuai, dan menyediakan pelanggan tersebut untuk rute yang dilindungi. Validasi token sebelum memperoleh koneksi database agar token yang tidak valid tidak menggunakan koneksi pooled.

Buat 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,
    }

Impor dependensi dan tambahkan rute terlindungi ke main.py:

from auth import get_current_customer


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

Gunakan penyedia identitas untuk mengeluarkan dan memutar kunci penandatanganan. Untuk HS256, atur JWT_SECRET ke setidaknya 32 byte acak. Jangan menyimpan rahasia penandatanganan produksi di repositori atau dalam gambar.

Uji aplikasi

FastAPI TestClient mengirim permintaan ke aplikasi tanpa memulai server HTTP. Pengujian integrasi berikut menggunakan database yang telah dikonfigurasi.

Buat 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"

Jalankan pengujian dari root proyek:

pytest

Pengujian ini menggunakan database yang dikonfigurasi, dan test_create_product menyisipkan baris ke dalam SalesLT.Product. Gunakan database pengujian khusus dan atur ulang datanya di antara setiap pelaksanaan pengujian.

Daftar periksa penyebaran

Atur DATABASE_SERVER, DATABASE_NAME, dan JWT_SECRET melalui penyimpanan rahasia dan konfigurasi milik platform deployment.
Gunakan identitas Microsoft Entra khusus dengan izin database minimum yang diperlukan.
Atur ukuran pool di bawah batas koneksi database dan sisakan kapasitas untuk akses administratif serta beban kerja lainnya.
Jalankan pengujian integrasi basis data terhadap basis data pengujian yang terisolasi.
Konfigurasikan telemetri terlindungi untuk eksepsi database, latensi permintaan, dan kehabisan pool.
Jalankan Uvicorn tanpa --reload di lingkungan produksi.

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-09-23