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 "Cloud Tasks"

Abstração para criação de HTTP tasks no Google Cloud Tasks, centralizando a comunicação e evitando acoplamento direto à lib google-cloud-tasks nas aplicações consumidoras.

Instalação

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

Ou utilizando poetry:

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

Utilização genérica

from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig, HttpTaskRequest

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="my-queue",
    )
)

task_name = client.create_http_task(
    HttpTaskRequest(
        url="https://my-service/internal/some-job",
        payload={"key": "value"},
        task_name="my-deterministic-task",  # opcional — garante idempotência
    )
)

Retenção por tenant

from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig
from dta_utils_python.cloud_tasks.retention import create_retention_task_for_tenant

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="retention-queue",
    )
)

# Idempotente: chamar duas vezes no mesmo dia/tenant não gera erro
task_name = create_retention_task_for_tenant(
    client=client,
    worker_url="https://my-service/internal/retention/process-tenant",
    tenant_schema="tenant-a",
    retention_days=180,
    batch_size=500,
)

O nome da task gerado segue o padrão retention-{YYYYMMDD}-{tenant_schema}, com caracteres especiais sanitizados.

Autenticação OIDC

Para endpoints protegidos, o Cloud Tasks pode gerar e anexar um OIDC token automaticamente na entrega da task. Basta informar o service account e, opcionalmente, o audience (padrão: a própria URL da task):

from dta_utils_python.cloud_tasks import CloudTasksClient, CloudTasksConfig, HttpTaskRequest, OidcConfig

client = CloudTasksClient(
    CloudTasksConfig(
        project_id="my-project",
        location_id="us-central1",
        queue_id="my-queue",
    )
)

task_name = client.create_http_task(
    HttpTaskRequest(
        url="https://my-service/internal/some-job",
        payload={"key": "value"},
        oidc_token=OidcConfig(
            service_account_email="my-sa@my-project.iam.gserviceaccount.com",
            # audience="https://my-service/internal/some-job",  # opcional
        ),
    )
)

Emulator local

Configure via variável de ambiente:

CLOUD_TASKS_EMULATOR_HOST=localhost:8123

Ou via configuração:

CloudTasksConfig(
    project_id="local-project",
    location_id="us-central1",
    queue_id="my-queue",
    emulator_host="localhost:8123",
)

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_*

Identity V1 e V2

validate_identity_jwt(...) identifica tokens V2 pelo issuer e seleciona automaticamente o JWKS e audience de produção:

issuer:   https://accounts.totvs.app/accounts/api/v2
JWKS:     https://accounts.totvs.app/accounts/api/v2/.well-known/jwks.json
audience: totvsdta

V1 continua sendo o default para tokens sem issuer V2. Para ambientes não produtivos, configure o par confiável IDENTITY_V2_ISSUER e IDENTITY_V2_JWKS_URL, além de IDENTITY_V2_AUDIENCE, ou passe jwks_base_url, identity_audience e identity_issuer nas APIs de autorização. Valores explícitos de V2 têm precedência sobre os defaults automáticos. Para compatibilidade, os defaults V1 conhecidos (DEFAULT_JWKS_BASE_URL e fluig_authenticator_resource) são ignorados quando o token é V2; assim, consumidores legados podem continuar passando esses valores sem quebrar o rollout. Overrides V2 customizados continuam explícitos e devem ser usados com identity_issuer correspondente. O issuer nunca é derivado dinamicamente do token antes da assinatura ser validada. Tokens V2 opacos (access_token) não são JWTs e não podem ser validados via JWKS; use o id_token JWT ou fluxo de introspecção/userinfo do provedor.

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

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.