Prueba y despliega aplicaciones FastAPI con mssql-python

Después de construir una aplicación FastAPI con mssql-python, configúrala para despliegue, reutilización de conexiones, manejo de errores, autenticación y pruebas automatizadas.

Prerequisites

Completa Usar mssql-python con FastAPI, o dispón de una aplicación FastAPI equivalente que utilice la base de datos de ejemplo AdventureWorksLT. La dependencia de autenticación en este artículo consulta SalesLT.Customer.

Instala las dependencias de producción y prueba:

pip install pydantic-settings pyjwt pytest httpx

Configurar los ajustes de implementación

Usa Pydantic Settings para cargar valores específicos del despliegue a partir de variables de entorno. Este enfoque mantiene los secretos fuera del código fuente y proporciona a cada entorno su propia base de datos, pool y configuración de autenticación.

Creación de 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"
    )

Establece DATABASE_SERVER, DATABASE_NAME, y JWT_SECRET en el entorno de despliegue. Pydantic Settings lee automáticamente los nombres de las variables de entorno en mayúsculas.

Nota:

ActiveDirectoryDefault prueba varios proveedores de credenciales de forma secuencial. En producción, especifica el modo de autenticación para la identidad desplegada, como ActiveDirectoryMSI para la identidad gestionada, para evitar recorrer la cadena de credenciales. Para conocer los modos disponibles, consulta Autenticación de Microsoft Entra con mssql-python.

Configuración de la agrupación de conexiones

MSSQL-Python permite la agrupación de conexiones por defecto. Configura el pool una vez, antes de que la aplicación cree su primera conexión. Dimensiona el pool para el trabajo concurrente esperado de la aplicación en bases de datos y el nivel de servicio de base de datos.

Actualiza database.py para usar la configuración de implementación:

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

El gestor de contexto de la conexión confirma después de procesar correctamente la solicitud, revierte los cambios cuando el procesamiento de la solicitud genera una excepción y cierra la conexión. Cerrar la conexión lo devuelve a la piscina. Para obtener información sobre claves de agrupación, dimensionamiento, aislamiento de identidad y recomendaciones sobre agotamiento, consulte Agrupación de conexiones en mssql-python.

Gestionar errores de base de datos

Registra controladores de excepciones para que los fallos de la base de datos devuelvan respuestas consistentes sin exponer detalles de conexión, consultas ni mensajes de error del servidor.

Añadir los manejadores después de app = FastAPI(...) en 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",
        },
    )

Registra la excepción a través de la tubería de telemetría protegida de tu aplicación antes de devolver la respuesta. Para la jerarquía de excepciones y el manejo de SQLSTATE, véase Gestión de errores y códigos SQLSTATE para mssql-python.

Añadir dependencias de autenticación

Encadena las dependencias de FastAPI para validar un JSON Web Token (JWT), cargar el cliente de AdventureWorksLT correspondiente y poner ese cliente a disposición de las rutas protegidas. Valida el token antes de adquirir una conexión a la base de datos para que un token inválido no use una conexión agrupada.

Creación de 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,
    }

Importa la dependencia y añade una ruta protegida a main.py:

from auth import get_current_customer


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

Utiliza un proveedor de identidad para emitir y rotar las claves de firma. Para HS256, configúralo JWT_SECRET en al menos 32 bytes aleatorios. No almacenes un secreto de firma de producción en el repositorio ni en una imagen.

Prueba de la aplicación

FastAPI TestClient envía solicitudes a la aplicación sin iniciar un servidor HTTP. Las siguientes pruebas de integración utilizan la base de datos configurada.

Creación de 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"

Ejecuta las pruebas desde la raíz del proyecto:

pytest

Estas pruebas utilizan la base de datos configurada e test_create_product insertan una fila en SalesLT.Product. Utiliza una base de datos específica para pruebas y restablece sus datos entre ejecuciones de prueba.

Lista de comprobación de la implementación

Establezca DATABASE_SERVER, DATABASE_NAME, y JWT_SECRET a través de los almacenes secretos y de configuración de la plataforma de despliegue.
Utiliza una identidad Microsoft Entra dedicada con los permisos mínimos requeridos para la base de datos.
Establece el tamaño del pool por debajo del límite de conexión de la base de datos y deja capacidad para acceso administrativo y otras cargas de trabajo.
Realiza pruebas de integración de bases de datos contra una base de datos aislada.
Configura la telemetría protegida para excepciones en la base de datos, latencia de solicitudes y agotamiento de pool.
Usa Uvicorn sin --reload en entornos desplegados.

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-09-23