totvs-dta-utils 1.4.24

Lib for integration with DTA services

Dta Utils 🧰 🛠️

Agilize a integração entre serviços DTA

O que são Serviços DTA?

Uma coleção de serviços para facilitar e acelerar o desenvolvimento e monitoramento de Aplicações, com foco em aplicativos de IA generativa.

Introdução

Esse pacote possui módulos extras que auxiliam o desenvolvimento de integrações com os serviços do DTA.

Extra "Secrets"

Instalação

Instale o módulo secrets com:

pip install "totvs-dta-utils[secrets]"

Ou utilizando poetry:

poetry add "totvs-dta-utils[secrets]"

Configuração inicial:

Adicione as seguintes variaveis no .env do seu projeto:

DTA_ENVIRONMENT="development"
DTA_INTEGRATION_URL="{DTA_INTEGRATION_URL}"

NOTE: Para ambiente em cloud, onde terá acesso irrestrito aos secrets, o valor do DTA_ENVIRONMENTdeve ser production.

Utilização

from dta_utils_python import DtaSecrets

auth = DTA_JWT  # CLIENT AUTHORIZATION

secrets = DtaSecrets(authorization=auth,
                     project="dta-empodera")

all_secrets = secrets.all()  # Get the latest version of all secrets
my_secret = secrets.get("MY_SECRET")  # Get the latest version of a secret
my_secret_v2 = secrets.get("MY_SECRET", version=2)  # Get a specific version of a secret

Observação: Para ambiente em nuvem na rede DTA, nenhuma autenticação é necessária.

Observação 2: Ainda em ambientes de nuvem, usando Cloud Run, lembrar de habilitar TODAS as chamadas de saída do serviço DEVEM passar pela VPC. Selecione Route all traffic to the VPC na configuração de Rede do serviço Cloud Run

Demais configurações:

DtaSecrets(
    authorization=auth,
    project="dta-empodera",
    raise_exception: bool = True,  # Default "False" - Levanta exceção em caso de erro ao obter a secret
    autoload: bool = False,  # Default "True" - Pré-carrega todas as secrets do projeto na inicialização da classe e as mantém em cache de memória
)

Tipos de retorno:

• .get("SECRET_2"): Retorna o valor da secret ou None caso a secret não exista.

any: "321654"

• .all(): Retorna um dicionário (hashmap) contendo a última versão de todas as secrets

dict: {
    "SECRET_1": "123456",
    "SECRET_2": "321654",
    "SECRET_3": "My secret",
}

Extra "Auth" (Identity/JWT)

Desenvolvimento local (Identity)

Para viabilizar autorização via Identity em ambiente local (sem Redis/RAC), configure:

DTA_ENVIRONMENT="development"  # ou "local"
DTA_TENANT_NAME="dta-alisson"
DTA_TOTVS_TENANT_ID="1a2b3c-4d5e6f"
DTA_ROLE_USER="TENANT_ADMIN"
# Opcional: sobrescreve o endpoint JWKS usado para validar o JWT Identity
# DTA_IDENTITY_JWKS_BASE_URL="https://api-fluig.totvs.app/accounts/api/v1/jwks"

Com isso, get_identity_user_from_cache(...) passa a:

• Validar o JWT Identity via JWKS (RS256)
• Ignorar Redis (cache) para o token
• Resolver dta_roles a partir de DTA_*

Extra "Healthcheck"

Componente reutilizável de health check para aplicações FastAPI. Expõe endpoints padronizados de liveness e readiness.

Instalação

pip install "totvs-dta-utils[healthcheck]"

Endpoints

Endpoint	Método	Descrição
/health/live	GET	Liveness — processo está rodando
/health/ready	GET	Readiness — dependências externas OK

Utilização

from fastapi import FastAPI
from dta_utils_python.healthcheck import create_health_router, DtaHealthCheck

app = FastAPI()

async def check_database() -> bool:
    # await db.execute("SELECT 1")
    return True

def check_redis() -> bool:
    # redis_client.ping()
    return True

app.include_router(
    create_health_router(
        readiness_checks=[
            DtaHealthCheck(name="database", check=check_database),
            DtaHealthCheck(name="redis", check=check_redis),
        ]
    )
)

Respostas

GET /health/live — sempre 200:

{"status": "ok"}

GET /health/ready — 200 se todos os checks passarem, 503 caso contrário:

{
  "status": "ok",
  "checks": {
    "database": "ok",
    "redis": "ok"
  }
}

Configurações opcionais

create_health_router(
    readiness_checks=[...],
    prefix="/api/health",   # default: "/health"
    live_path="/ping",      # default: "/live"
    ready_path="/status",   # default: "/ready"
)

DtaHealthCheck(
    name="database",
    check=check_database,
    timeout=5.0,  # segundos; default: 2.0
)

Checks sync e async são suportados. Exceções e timeouts são tratados como falha sem vazar detalhes na resposta.