Testar e implementar aplicações FastAPI com mssql-python

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

Pré-requisitos

• Conclua Utilizar o mssql-python com FastAPI, ou tenha uma aplicação FastAPI equivalente que utilize a base de dados de exemplo AdventureWorksLT. A dependência de autenticação neste artigo questiona 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 as Configurações Pydânticas para carregar valores específicos de implementação a partir de variáveis de ambiente. Esta abordagem mantém segredos fora do código-fonte e dá a cada ambiente a sua própria base 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 maiúsculos das variáveis do ambiente.

Note

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

Configurar o agrupamento de conexões

mssql-python ativa o agrupamento de ligações por predefinição. Configure o pool uma vez, antes de a aplicação criar a sua primeira ligação. Dimensione o pool para a carga de trabalho simultânea esperada da aplicação na base de dados e para o escalão de serviço da base de dados.

Atualize database.py para usar as definições de implementaçã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 gestor de contexto da ligação confirma após o processamento bem-sucedido da solicitação, reverte as alterações quando o processamento da solicitação gera uma exceção e fecha a ligação. Fechar a conexão devolve-a ao pool. Para chaves do conjunto, dimensionamento, isolamento de identidade e orientações sobre exaustão, consulte Agrupamento de ligações com mssql-python.

Lidar com erros na base de dados

Registe tratadores de exceções para que falhas na base de dados devolvam respostas consistentes sem expor detalhes de ligação, consultas ou texto de erro do servidor.

Adicione os tratadores depois 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",
        },
    )

Regista a exceção através do pipeline de telemetria protegida da tua aplicação antes de devolver a resposta. Para a hierarquia de exceções e o tratamento SQLSTATE, veja Gestão de erros e códigos SQLSTATE para mssql-python.

Adicionar dependências de autenticação

Encadear dependências do FastAPI para validar um JSON Web Token (JWT), carregar o cliente AdventureWorksLT correspondente e disponibilizar esse cliente às rotas protegidas. Valide o token antes de adquirir uma ligação à base de dados para que um token inválido não use uma ligaçã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,
    }

Importa a dependência e adiciona 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

Utilize um fornecedor de identidade para emitir e rotacionar chaves de assinatura. Para o HS256, defina JWT_SECRET para pelo menos 32 bytes aleatórios. Não guarde um segredo de assinatura de produção no repositório ou numa imagem.

Testar a aplicação

O TestClient FastAPI envia pedidos para a aplicação sem iniciar um servidor HTTP. Os testes de integração seguintes utilizam a base de dados configurada.

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

Estes testes utilizam a base de dados configurada e test_create_product inserem uma linha em SalesLT.Product. Utilize uma base de dados de teste dedicada e redefina os seus dados entre execuções de teste.

Lista de verificação de implantação

• Defina DATABASE_SERVER, DATABASE_NAME e JWT_SECRET nos repositórios de segredos e de configuração da plataforma de implementação.
• Use uma identidade Microsoft Entra dedicada com as permissões mínimas necessárias para a base de dados.
• Definir o tamanho do pool abaixo do limite de ligação da base de dados e deixar capacidade para acesso administrativo e outras cargas de trabalho.
• Execute testes de integração de base de dados contra uma base de dados de teste isolada.
• Configurar telemetria protegida para exceções na base de dados, latência de pedidos e esgotamento do pool.
• Execute o Uvicorn sem --reload em ambientes de produção.

Conteúdo relacionado

• Utilize o mssql-python com o FastAPI
• Agregação de ligações com mssql-python
• Gestão de erros e códigos SQLSTATE para mssql-python
• Autenticação Microsoft Entra com mssql-python
• Boas práticas de segurança para aplicações mssql-python