Teste e implante aplicações FastAPI com mssql-python

Depois de construir uma aplicação FastAPI com mssql-python, configure-a para implantação, reutilização de conexão, tratamento de erros, autenticação e testes automatizados.

Pré-requisitos

• Conclua Usar mssql-python com FastAPI ou tenha um aplicativo FastAPI equivalente que use o banco de dados de amostra AdventureWorksLT. A dependência de autenticação neste artigo consulta SalesLT.Customer.

• Instale as dependências de produção e teste:

  pip install pydantic-settings pyjwt pytest httpx
  

Definir configurações de implantação

Use Configurações Pydantic para carregar valores específicos de implantação a partir de variáveis de ambiente. Essa abordagem mantém segredos fora do código-fonte e dá a cada ambiente seu próprio banco de dados, pool e configuração de autenticação.

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

Defina DATABASE_SERVER, DATABASE_NAME, e JWT_SECRET no ambiente de implantação. O Pydantic Settings lê automaticamente os nomes de variáveis de ambiente em maiúsculas.

Note

ActiveDirectoryDefault Tenta vários provedores de credenciais em sequência. Em produção, especifique o modo de autenticação para a identidade implantada, como ActiveDirectoryMSI para identidade gerenciada, para evitar percorrer a cadeia de credenciais. Para os modos disponíveis, consulte autenticação do Microsoft Entra com mssql-python.

Configurar o pool de conexões

MSSQL-Python permite o pooling de conexões por padrão. Configure o pool uma vez, antes que o aplicativo crie sua primeira conexão. Dimensione o pool de acordo com a carga de trabalho simultânea esperada do aplicativo no banco de dados e com a camada de serviço do banco de dados.

Atualize database.py para usar as configurações de implantação:

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

O gerenciador de contexto da conexão confirma após o processamento bem-sucedido da requisição, executa rollback quando o processamento da requisição gera uma exceção e fecha a conexão. Fechar a conexão retorna para a piscina. Para chaves de pool, dimensionamento, isolamento de identidade e diretrizes sobre exaustão, consulte Pool de conexões com mssql-python.

Lidar com erros de banco de dados

Registre manipuladores de exceções para que falhas no banco de dados retornem respostas consistentes sem expor detalhes de conexão, consultas ou texto de erro do servidor.

Adicione os manipuladores depois de app = FastAPI(...) em 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",
        },
    )

Registre a exceção pelo pipeline de telemetria protegido da sua aplicação antes de retornar a resposta. Para a hierarquia de exceções e o tratamento de SQLSTATE, veja Gerenciamento de erros e códigos SQLSTATE para mssql-python.

Adicionar dependências de autenticação

Encadeie dependências do FastAPI para validar um JSON Web Token (JWT), carregar o cliente correspondente do AdventureWorksLT e disponibilizar esse cliente para as rotas protegidas. Valide o token antes de adquirir uma conexão de banco de dados para que um token inválido não use uma conexão em pool.

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

Importe a dependência e adicione uma rota protegida para main.py:

from auth import get_current_customer


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

Use um provedor de identidade para emitir e rotacionar as chaves de assinatura. Para HS256, defina JWT_SECRET para pelo menos 32 bytes aleatórios. Não armazene uma chave secreta de assinatura de produção no repositório nem em imagem.

Testar o aplicativo

O TestClient FastAPI envia requisições para a aplicação sem iniciar um servidor HTTP. Os seguintes testes de integração utilizam o banco de dados configurado.

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

Execute os testes a partir da raiz do projeto:

pytest

Esses testes utilizam o banco de dados configurado e test_create_product inserem uma linha em SalesLT.Product. Use um banco de dados de teste dedicado e redefina os dados entre cada execução de teste.

Lista de verificação de implantação

• Defina DATABASE_SERVER, DATABASE_NAME, e JWT_SECRET através dos armazenamentos secretos e de configuração da plataforma de implantação.
• Use uma identidade dedicada da Microsoft Entra com as permissões mínimas necessárias para o banco de dados.
• Defina o tamanho do pool abaixo do limite de conexão do banco de dados e deixe capacidade para acesso administrativo e outras cargas de trabalho.
• Execute testes de integração de banco de dados contra um banco de dados isolado.
• Configure telemetria protegida para exceções no banco de dados, latência de requisição e esgotamento do pool.
• Execute o Uvicorn sem --reload em ambientes de produção.

Conteúdo relacionado

• Use mssql-python com o FastAPI
• Pooling de conexão com mssql-python
• Tratamento de erros e códigos SQLSTATE para mssql-python
• Autenticação do Microsoft Entra com mssql-python
• Melhores práticas de segurança para aplicações mssql-python